javascript - 如何使用 RxJs 6 GroupBy 来组织流？需要连接来自多个 Grouped Observables 的所有排放

Question

输入可观察流:
数据是从一个可观察的流中获得的，该流是对项目的 REST 请求的结果。数据作为 Observable 获得。

     const project1: Project = {
        id: 1,
        title: 'zebra',
        rootId: 1,
      }
    
      const project2: Project = {
        id: 2,
        title: 'algebra',
        rootId: 2,
      }
    
      const project3: Project = {
        id: 3,
        title: 'Bobcats',
        rootId: 1,
      }
    
      const project4: Project = {
        id: 4,
        rootId: 2,
      }
    
      const project5: Project = {
        id: 5,
        title: 'Marigolds',
        rootId: 1,
      }
    
      const project6: Project = {
        id: 6,
        title: 'whatever',
        rootId: null,
      }
    
      const project7: Project = {
        id: 7,
        title: 'peppercorns',
        rootId: null,
      }
    
    let groupProjects: Observable<ProjectSummary[]> 
= getGroupProjects(of([project1, project2, project3, project4, project5, project6, project7]]));
    
      getGroupProjects(projects$: Observable<ProjectSummary[]>): Observable<ProjectSummary[]> {
        const timer$ = timer(5000);
        const data = projects$.pipe(takeUntil(timer$), flatMap(projects => projects));
        const groupedObservables = data.pipe(
          groupBy(projects => projects.rootId),
          tap( a => console.log('groupBy:' + a.key))
        );
        const merged = groupedObservables.pipe(
          mergeMap(a => a.pipe(toArray())),
          shareReplay(1),
          tap( a => console.log('final:' + JSON.stringify(a)))
        );
        return merged;
      }

所需的输出是:

Object{  //Root of 1
  id: 1,
  title: 'zebra',
  rootId: null
}
Object{
  id: 3, //child of 1
  title: 'Bobcats',
  rootId: 1
} 
Object{
  id: 5, //child of 1
  title: 'Marigolds',
  rootId: 1
}
Object{
  id: 2, //root of 2
  title: 'algebra',
  rootId: 2
}
Object{
  id: 4,  //child of 2
  title: 'dogs',
  rootId: 2
}
Object{
  id: 6,  //unaffiliated
  title: 'whatever',
  rootId: null
}
Object{
  id: 7, //unaffiliated
  title: 'peppercorns',
  rootId: null
}

要求是由 rootId 标识的组按顺序出现在其子项之前(子项出现在其根之后)和非关联组一起列出。当 id = rootId 时识别根，当 rootId != null && id != rootId 时识别 child 。非附属由空根 ID 标识。
目前只发出最后一组。如何返回一个以正确顺序发出所有组的 observable？ - 谢谢

Answer 1

groupBy 获取对象流并在流完成时发出单个组，它不适用于数组流。你想要的是扫描。 Scan 就像一个 reduce，但它在源流每次发出时发出，而不是在最后发出一次。
我不太明白你想从你的问题中得到什么，但这应该让你开始

sourceThatEmitsArrays.pipe(
  scan(
   (results, emittedArray) => functionThatAddsEmittedArrayToResults(results, emittedArray),
   [] // Start with an empty array
  )
)

这与数组上的普通 reduce 函数相同，但每次源发出时都会发出结果。
functionThatAddsEmittedArrayToResults 看起来像

(results, array) => array.reduce(
  (newResults, current) => {
    const group = findCurrentGroupInNewResultsOrCreateNewGroup(newResults, current);
    replacePreviousGroupInResultsOrAddTheNewOne(newResults, group);
    return newResults;
  },
  results // Start with previous results
)