RxJS：groupBy 与计时器一起使用时不发出任何值

Question

RxJS (^6.5.5) 的新手，我对 groupBy 运算符有疑问。下面是一个精简的例子。

我有一个函数retrieveFiles()，它返回一个字符串数组。

function async retrieveFiles(): Promise<string[]> {
  return ['file1.txt', 'file2.txt', 'file3.txt', 'directory1', 'directory2'];
}

实际上，这会从远程源获取一组数据。

假设我想按前 4 个字符对文件名进行分组。

使用RxJS我可以这样做。

of(retrieveFiles()).pipe(
  concatMap(v => v),
  mergeMap(value => value),
  groupBy(
    (name: string) => name.substr(0, 4),
    (name: string) => name,
  ),
  mergeMap(group$ => group$.pipe(toArray())),
)
.subscribe(console.log);

这将向订阅者发出两个值。

[ 'file1.txt', 'file2.txt', 'file3.txt' ]
[ 'directory1', 'directory2' ]

现在让我们引入timer 并稍微修改一下代码。我们现在基本上是在投票。

timer(0, 1000)
  .pipe(
    concatMap(() => this.retrieveFiles()),
    mergeMap(value => value),
    groupBy(
      (name: string) => name.substr(0, 4),
      (name: string) => name,
    ),
    mergeMap(group$ => group$.pipe(toArray())), 
  )
  .subscribe(console.log);

这不再发出任何值。这两者有什么区别？

Answer 1

在第二种情况下，区别在于 timer 永远不会完成，而 of 会完成，而 toArray 仅在源完成时发出，但源没有完成，因为来自 groupBy 的分组可观察对象是期待未来的排放，因为 timer 是它的来源，所以它永远不会完成它的分组观察。

如果您希望为每个投票生成新组，则需要进行一次小的重组，以便将 retrieveFiles 设为其来源

timer(0, 1000)
  .pipe(
    // recommend use switchMap so a hanging request doesn't clog your stream
    switchMap(() => from(retrieveFiles()).pipe( // need from to make it pipeable
      mergeMap(value => value),
      groupBy(
        (name: string) => name.substr(0, 4),
        (name: string) => name,
      ),
      mergeMap(group$ => group$.pipe(toArray())), 
    )),
  )
  .subscribe(console.log);

这里有一个有效的讨论，关于这是否值得。你可以通过一个简单的map 操作符来实现这个目标，它里面有一个同步数组分组函数，而不是所有的来回流转换。如果您只是在探索图书馆，那很酷，但在实践中，我不建议您这样做。 groupBy 用例并不常见，但应保留用于处理要分组为单独的可观察流的实际值流