Наблюдаемая подписка не срабатывает

Question

У меня есть следующий фрагмент:

static groupAlphabetically<T>(observable:Observable<T[]>): Observable<AlphabeticalGroup<T>[]>{
return (observable.pipe(
  switchMap(data => data as T[]), 
  reduce((r, e:T) => {
    let group = e["name"][0] as string;
    var currentGroup = r.filter(x => x.letter == group)[0];
    if(!currentGroup) 
    {
      currentGroup = new AlphabeticalGroup<T>();
      currentGroup.letter = group;
      r.push(currentGroup);
    }

    currentGroup.items.push(e);
    return r;
  }, new Array<AlphabeticalGroup<T>>())
  ));
}

По какой-то причине эта наблюдаемая по трубопроводу никогда не заканчивается. Что я делаю неправильно? При отладке я могу перейти к оператору return r, но подписка никогда не работает:

subscribe(x => {
// this code is never reached
})

Answer 1

switchMap не расширяет массив, поэтому в вашей функции сокращения e! = T, но e = T []. Используйте оператор groupBy в RxJS.

См. пример ниже

export const groupAlphabetical = (observable: Observable<hasName[]>) => observable.pipe(
      flatMap(x => x),
      groupBy(x => x.name[0],x=> x),
      mergeMap( (group$) => group$.pipe(reduce((acc, cur) =>{
          acc.items = [...acc.items, cur];
          return acc;
      }, {
        letter: group$.key,
        items:[]
      })))
    );

интерфейс hasName:

export interface hasName {
  name: string;
}