Как правильно сгладить поток наблюдаемых? - PullRequest
Новая
       9

Как правильно сгладить поток наблюдаемых?

0 голосов
/ 18 июня 2020

Вот что я пытаюсь сделать:

Учитывая разбитый на страницы API, получить все ресурсы с помощью параллельных запросов.

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

Вот моя идея (но получаю предупреждение, потому что я использую плоский в ответе), поэтому, возможно, есть способ лучше сделать это.

  1. Получить общее количество элементов.
  2. Учитывая количество и лимит, вычислите, сколько запросов необходимо для получения всех данных.
  3. Запускать все запросы параллельно и объединять все данные в уплощенный массив.

Вот пример:

https://stackblitz.com/edit/paginated-api?embed=1&file=index.ts&hideExplorer=1&devtoolsheight=100

getCount().pipe(
  mergeMap(count => range(0, Math.ceil(count / limit))),
  map(offset => getDevices(offset, limit)),
  combineAll(),
).subscribe(res => {
  const a = res.flat(); // <--- warning: Property 'flat' does not exist on type '{ name: string; }[][]'.
  console.log(JSON.stringify(a));
});

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

Ответы [ 2 ]

3 голосов
/ 19 июня 2020

Для каждого внутреннего Observable нам нужен другой оператор сглаживания.

Таким образом, будет работать что-то вроде этого: 

getCount().pipe(

  mergeMap(count => range(0, Math.ceil(count / limit))),

  mergeMap(offset => getDevices(offset, limit)),

  mergeAll(),
  toArray()

).subscribe(res => {
  console.log('result', JSON.stringify(res));
});

Первый mergeMap сглаживает внутренний range Observable. Второй mergeMap сглаживает getDevices, который, как я полагаю, возвращает Observable.

mergeAll() объединяет все отдельные значения, которые являются объектами.

The toArray() затем складывает все объекты в один массив.

Результат: 

result
[{"name":"dev-1"},{"name":"dev-2"},{"name":"dev-3"},{"name":"dev-4"},{"name":"dev-5"},{"name":"dev-6"},{"name":"dev-7"},{"name":"dev-8"},{"name":"dev-9"},{"name":"dev-10"},{"name":"dev-11"},{"name":"dev-12"},{"name":"dev-13"},{"name":"dev-14"},{"name":"dev-15"},{"name":"dev-16"},{"name":"dev-17"},{"name":"dev-18"},{"name":"dev-19"},{"name":"dev-20"}]

Надеюсь, это поможет.

0 голосов
/ 19 июня 2020

3 способа сделать это:

Используя mergeMap, все запросы запускаются параллельно. Однако конечный результат будет зависеть от порядка прибытия. Это означает, что если ваш API отсортирован, это может сломать его. 

const getAllOffsets = () => pipe(
  mergeMap((count: number) => range(0, Math.ceil(count / limit))),
  toArray(),
  // all requests in parallel and results in order (order of creation)
  concatMap(r => forkJoin(...r.map(offset => getDevices(offset, limit)))),
  map(a => a.flat())
);

Используя concatMap, это гарантирует порядок. Однако каждый запрос выполняется один за другим. Это будет проблема с производительностью 

const getAllOffsets2 = () => pipe(
  mergeMap((count: number) => range(0, Math.ceil(count / limit))),
  // all requests in parallel but order is not guarantee (response order)
  concatMap(offset => getDevices(offset, limit)),
  mergeAll(),
  toArray()
);

Наконец, используя forkJoin. Это будет выполнять все запросы параллельно и сохранять их в порядке создания. Как Promise.all. Это наиболее оптимальный вариант для получения всех ресурсов из API с разбивкой на страницы. 

const getAllOffsets3 = () => pipe(
  mergeMap((count: number) => range(0, Math.ceil(count / limit))),
  toArray(),
  // all requests in parallel and results in order (order of creation)
  concatMap(r => forkJoin(...r.map(offset => getDevices(offset, limit)))),
  map(a => a.flat())
);

Полный рабочий пример: https://stackblitz.com/edit/paginated-api-ozcgwg?file=index.ts&hideExplorer=1&devtoolsheight=100

getCount().pipe(getAllOffsets3()).subscribe(res => {
  console.log({res, size: res.length || 0});
  console.log(JSON.stringify(res))
  // console.log('end', res.map(d => d.name));
});
...