不确定如何根据以前的值重置流

Question

我正在 Angular 应用程序中处理以下 RxJs 流，但在重置值时遇到了麻烦。我的应用程序中有多个位置在此处的 combineLatest 调用中向这三个可观察对象发出值，例如当用户更改过滤器设置或通过输入字段更新页面时。此外，我还有一个延迟加载功能，当用户接近底部时，它会向前移动页面。

当 name 和 filter 更新时，我希望它只返回来自 getContent 的最新数据，但是当 page observable 有一个新值时，我希望它通过 @ 将以前的数据与当前数据结合起来987654326@接线员。我遇到的唯一问题是我似乎无法在扫描中找到执行此操作的最佳方法，因为那时它不知道 mergeMap 来自 name 和 @987654329 的当前值和先前值@.

getContent(name, page filter) {
    return this.http
      .get(
        `${this.API_BASE}/${name}/${filter}/${page}`
)

流如下所示：

this.results$ = combineLatest(
  this.dataService.getName(),
  this.dataService.getPage(),
  this.dataService.getFilter()
).pipe(
  mergeMap(([name, page, filter]) => {
     this.dataService.getContent(name, filter, page);
  }),
  scan(
    (
      acc,
      curr
    ) => {
      this.nextPage = curr[curr.length - 1].id;
      if (acc.length && curr.length) {
         return acc.concat(curr);
      }

      return acc;
    },
    []
  )
);

该模板只是一个 div，它会循环并使用 async 管道进行更新，如果可能的话我想保留它。有没有更好的方法在单个流中处理这个问题，或者有什么方法可以让它按照我需要的方式进行分解？

Answer 1

如果我对问题的理解正确，您可以尝试这些方法

// start creating a stream that is the result of combineLatest, but share it,
// so that any other stream that uses this stream as its upstream will
// use the same shared upstream
// name, page and filter notifications are enriched with a value that says
// whether scan has to be used or not downstream
this.streamsShared$ = combineLatest(
  this.dataService.getName().pipe(res => ({res: res, scan: false})),
  this.dataService.getPage().pipe(res => ({res: res, scan: true})),
  this.dataService.getFilter().pipe(res => ({res: res, scan: false}))
).pipe(
  share()
)

// then you create a stream for the cases where scan has not to be used
this.noScan$ = this.streamsShared$.pipe(
  filter(val => !val.scan),
  // if no scan has to be used, then you just return the value returned 
  // by getContent
  // I used concatMap rather than mergeMap because usually it is the best
  // operator to use with chains of http calls (but this is another story)
  concatMap(([name, page, filter]) => {
     this.dataService.getContent(name, filter, page);
  }),
  // then you just return the value received by getContent in a one-value array
  map(retVal => [retVal])
)

// then you create a stream for the cases where scan has to be used
this.useScan$ = this.streamsShared$.pipe(
  filter(val => !val.scan),
  concatMap(([name, page, filter]) => {
     this.dataService.getContent(name, filter, page);
  }),
  scan(
    (
      acc,
      curr
    ) => {
      this.nextPage = curr[curr.length - 1].id;
      if (acc.length && curr.length) {
         return acc.concat(curr);
      }

      return acc;
    },
    []
  )
)

// eventually you merge the 2 observables to get the final stream which
// has to be passed to the async pipe
this.results$ = merge(this.noScan$, this.useScan$)

老实说，这整个机制对我来说似乎有点复杂，但你实际上在最终流中嵌入的是一个非平庸的状态：姓氏、最后一个过滤器和最后一页的状态以及结果链的状态获取内容。所有这些状态都由流包含和管理，您无需将其保存在实例变量中。

Answer 2

您可以简单地在您的“标准”上使用combineLatest，然后将其提供给switchMap（每当条件更改时提供重置), 然后将标准结果提供给 getPage() 的可观察对象。此时，您拥有调用getContent() 所需的所有 3 个参数。

这样的事情应该适合你：

  results$ = combineLatest([
    this.dataService.getName(),
    this.dataService.getFilter(),
  ]).pipe(
    switchMap(([name, filter]) => this.dataService.getPage().pipe(
      startWith(undefined), // initially emit `undefined` becuase there's not a "next page" cursor for the first call.
      mergeMap(page => this.dataService.getContent(name, filter, page)),
      scan((acc, curr) => {
        this.nextPage = curr[curr.length - 1].id;
  
        if (curr.length) {
          return acc.concat(curr);
        }
  
        return acc;
      }, [])
    )),
  );

---

最好是可观察流不依赖于外部变量。 (this.nextPage)

我们可以通过改变发射的形状来解决这个问题。除了仅发出“结果”之外，我们还可以让它包含“下一页”信息：

  query$ = combineLatest([
    this.dataService.getName(),
    this.dataService.getFilter(),
  ]).pipe(
    switchMap(([name, filter]) => this.dataService.getPage().pipe(
      startWith(undefined),
      mergeMap(page => this.dataService.getContent(name, filter, page)),
      scan(
        (acc, curr) => ({ 
          results  : curr.length ? acc.results.concat(curr) : acc.results, 
          nextPage : curr[curr.length - 1]?.id
        }),
        { results: [] as Result[], nextPage: undefined })
    )),
  );

<ng-container *ngIf="query$ | async as query">

  <div *ngFor="let result of query.results">
    {{ result.label }}
  </div>

  <button *ngIf="query.nextPage" (click)="loadMore(query.nextPage)"> 
    Load More 
  </button>

</ng-container>

这是一个有效的 StackBlitz 演示。

---

最后两个注意事项：

1. 如果getPage()、getFilter()和getName()方法不接受任何参数，您可以简单地将它们声明为可观察对象，而不是返回可观察对象的方法：page$、filter$、name$

2. 数据服务公开了多个可观察对象，因此组件可以订阅它们并将排放反馈给服务自己的 getContent() 方法。由于您的数据服务维护所有可观察的源，因此在服务中而不是在组件中将内容声明为可观察的可能会更简单。

   如果你想查看this StackBlitz:-)