Почему подписка не вызывается после оператора groupBy?

Question

Почему метод подписки во втором примере не вызывается?

Все журналы в конвейере работают, как и ожидалось, в обоих примерах.

ПРИМЕР РАБОТЫ: (но с использованием жестко закодированных данных и-creator):

---

of([
  {tableName: 'table1', firstName: 'name1', lastName: 'lastName1'},
  {tableName: 'table1', firstName: 'name2', lastName: 'lastName2'},
  {tableName: 'table2', firstName: 'name3', lastName: 'lastName3'},
  {tableName: 'table2', firstName: 'name4', lastName: 'lastName4'}
]).pipe(
  tap(data => console.log('amount of records', data.length)),
  mergeMap((searchResult) => searchResult),
  groupBy(b => b.tableName),
  mergeMap(group => {
    console.log('groupKey', group.key);
    return group.pipe(toArray());
  })
).subscribe(val => console.log(val));

КОД ПРОБЛЕМЫ: (используя данные непосредственно из хранилища ngrx и ту же структуру данных, что и в приведенном выше примере кода)

---

this.store.pipe(
  select(selectSearchResult),
  tap(data => console.log('amount of records', data.length)),
  mergeMap((searchResult) => searchResult),
  groupBy(b => b.tableName),
  mergeMap(group => {
    console.log('groupKey', group.key);
    return group.pipe(toArray());
  })
).subscribe(val => console.log(val)); // <- this is not called!?

Мне бы хотелось, чтобы метод подписки вызывался для каждой группы, как в приведенном выше рабочем примере с жестко закодированными данными. Но метод подписки по иронии судьбы не вызывается?!

Пример элемента из массива, доставленного магазином:

Answer 1

Я думаю, что проблема заключается здесь:

Магазин является долгоживущим Наблюдаемым. Это означает: он не будет завершать сам по себе.

toArray собирает все уведомления об исходном потоке до тех пор, пока этот поток не завершится и только затем отправляет один aуведомление с массивом элементов.

Иначе как бы он узнал, когда массив готов к отправке следующему оператору? Всегда может быть другое уведомление, которое может сделать массив еще больше.

Обратите внимание, что GroupedObservable, создаваемый groupBy, не будет завершен, пока не будет завершен его источник. Поскольку источник store не будет завершен, toArray никогда не получит сообщение complete.

Решение заключается в следующем: рефакторинг вашего кода, чтобы вы могли получить выполнимую наблюдаемую с from. from принимает массив в качестве аргумента, отправляет одно уведомление для каждого элемента массива, а затем отправляет полное уведомление , указывающее toArray, что он может отправить уведомление. Это должно работать:

mergeMap((searchResult) => 
  from(searchResult).pipe(
    groupBy(b => b.tableName),
    mergeMap(group => {
      console.log('groupKey', group.key);
      return group.pipe(toArray());
  })
))

Источники: Код RxJS Посмотрите, когда завершится GroupedObservable.

Answer 2

Кажется, вы путаете groupBy как функцию, которая работает с массивами вместо потока, функции RxJs применяются к потоку, а не со значениями, которые выдаются потоком.

groupBy не группирует массивы, которыеполучает излучение потоком, он группирует поток. Так как каждый раз, когда наблюдаемое испускает объект, оно добавляется в группу. Если вы хотите сгруппировать массив, который находится в магазине, вам нужна функция groupBy, которая работает с массивами.

Вот пример написанной мной функции groupBy, которая принимает массив https://stackblitz.com/edit/typescript-ezydzv

map - это функция RxJs, которая управляет объектом, излучаемым потоком, и на карте вы можете применить groupBy к массиву.

this.store.pipe(
  select(selectSearchResult),
  tap(data => console.log('amount of records', data.length)),
  map(data => groupBy(data, { keys: ['tableName'] }))
).subscribe(val => console.log(val));