Оператор расширения RXJS

На самом деле это довольно распространенный вариант использования RxJS, поскольку существует множество источников данных с разбивкой на страницы или источников, которые иным образом ограничены тем, что вы можете запросить за один раз.

Мои два цента

На мой взгляд, expand, вероятно, лучший оператор для этого, учитывая, что вы разбиваете страницы на неизвестный источник данных и вам требуется хотя бы один запрос, чтобы определить окончательный счет. Если бы вы знали, сколько данных вы собираетесь запрашивать, более простым вариантом было бы использовать что-то вроде mergeScan, но я отвлекся.

Предлагаемое решение

Это может потребовать некоторых усилий, чтобы разобраться в этом, поэтому я добавил аннотации везде, где это возможно, чтобы понять, как все это работает. Примечание Я на самом деле не проверял это, так что простите меня за любые синтаксические ошибки.

// Your constant limit for any one query
const limit = 500;

// RxJS helper method that wraps the async call into an Observable
// I am basing this on what I saw of your sample which leads me to believe
// that this should work. 
const clientish = Rx.Observable.bindCallback(rets.getAutoLogoutClient);

// A method wrapper around your query call that wraps the resulting promise
// into a defer.
const queryish = (client, params) =>
  // Note the use of defer here is deliberate, since the query returns
  // a promise that will begin executing immediately, this prevents that behavior
  // And forces execution on subscription.
  Rx.Observable.defer(() => client.search.query(SearchType, Class, Query, params));

// This does the actual expansion function
// Note this is a higher order function because the client and the parameters
// are available at different times
const expander = (client) => ({limit, count}) => 
  // Invoke the query method
  queryish(client, {limit, count})
    // Remap the results, update offset and count and forward the whole 
    // package down stream
    .map(results => ({
      limit, 
      count: results.count, 
      offset: offset + limit, 
      results
    }));


// Start the stream by constructing the client
clientish(config.clientSettings)
  .switchMap(client =>
     // This are the arguments for the initial call
     Rx.Observable.of({limit, offset: 0})
       // Call the expander function with the client
       // The second argument is the max concurrency, you can change that if needed
       .expand(expander(client), 1)

       // Expand will keep recursing unless you tell it to stop
       // This will halt the execution once offset exceeds count, i.e. you have
       // all the data
       .takeWhile(({count, offset}) => offset < count)

       // Further downstream you only care about the results
       // So extract them from the message body and only forward them
       .pluck('results')
  )
  .subscribe(results => /*Do stuff with results*/);

const retsConnect = Rx.Observable.create(function(observer) {
  rets.getAutoLogoutClient(config.clientSettings, client => {
    return searchQuery(client, 500, 1, observer);
  });
});

function searchQuery(client, limit, offset, observer) {
  let currentOffset = offset === undefined || offset === 0 ? 1 : offset;
  return client.search.query(SearchType, Class, Query, {limit: limit, offset: currentOffset})
    .then(results => {
      offset += limit;
      observer.next(results.maxRowsExceeded);
      if (results.maxRowsExceeded) {
        console.log(offset);
        return searchQuery(client, limit, offset, observer);
      } else {
        console.log('Completed');
        observer.complete();
      }
    });
}

retsConnect.subscribe(val => console.log(val));

Так что это где-то с тем, что я пробовал здесь. Я все еще в процессе настройки этого. Итак, что я хочу сделать, так это разбить поисковый запрос на более мелкие части. Не уверен, что мне следует передавать тудаObserver.Next, поэтому я собираюсь выяснить, где сопоставить и установить takeUntil, чтобы снова вернуть searchQuery. Я не уверен, что takeUntil примет значение true или false. Все, что мне нужно сделать, это сохранить эти данные в mongodb. Так что, я думаю, я мог бы оставить все как есть и добавить туда свой метод сохранения, но мне все равно нравится это выяснять.

Примечание: results.maxRowsExceeded возвращает значение true, если данных еще больше. Поэтому, как только maxRows вернет false, он остановится, и все данные будут получены.