为什么在 groupBy 运算符之后不调用 subscribe？

Question

为什么不调用第二个示例的订阅方法？

在这两个示例中，管道内的所有日志都按预期工作。

工作示例：（但使用硬编码数据和创建者）：

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-store 的数据和与上面的代码工作示例相同的数据结构）

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!?

我希望为每个组调用 subscribe 方法，就像上面使用硬编码数据的工作示例一样。但具有讽刺意味的是，订阅方法没有被调用？！

Answer 1

我认为问题出在这里：Store 是一个长期存在的Observable。这意味着：它不会自行完成。toArray收集其源流的所有通知，直到该流完成，然后才发送带有项目数组的单个通知。否则，它如何知道数组何时准备好发送给下一个操作员？总有另一个通知可以使阵列更大。请注意，GroupedObservable所生产groupBy将无法完成，直到它的来源是完整的。因为源store不会完成，toArray所以永远不会收到complete消息。解决方案是：重构您的代码，以便使用from. from以一个数组为参数，为每个数组项发送一个通知，然后发送一个完整的通知，表示toArray它可以发送一个通知。这应该有效：mergeMap((searchResult) =>&nbsp;&nbsp; from(searchResult).pipe(&nbsp; &nbsp; groupBy(b => b.tableName),&nbsp; &nbsp; mergeMap(group => {&nbsp; &nbsp; &nbsp; console.log('groupKey', group.key);&nbsp; &nbsp; &nbsp; return group.pipe(toArray());&nbsp; })))来源：RxJS 代码查看 GroupedObservable 何时完成。

Answer 2

它认为您将 groupBy 混淆为处理数组而不是流的函数，RxJs 函数适用于流，而不是流发出的值。groupBy 不会对流发出的数组进行分组，而是对流进行分组。每次 observable 发出一个对象时，它都会被添加到组中。如果要对存储中的数组进行分组，则需要使用数组的 groupBy 函数。这是我编写的 groupBy 函数的示例，该函数采用数组https://stackblitz.com/edit/typescript-ezydzvmap 是 RxJs 函数，它操作由流发出的对象，在 map 中，您可以将 groupBy 应用于数组。this.store.pipe(&nbsp; select(selectSearchResult),&nbsp; tap(data => console.log('amount of records', data.length)),&nbsp; map(data => groupBy(data, { keys: ['tableName'] }))).subscribe(val => console.log(val));