angular - RxJS 扩展减少循环不返回结果

Question

我正在尝试建立一个分页结果列表。我有循环工作，但我可以获得可观察到的结果以发出最终结果。它循环完成但从不发送给订阅者。不确定使用 EMPTY 是否是错误的完成方式，或者是否因为 EMPTY 而没有命中 reduce 并且永远不会触发最后的结果？

getReportsAll(collection: Collection): Observable<PagedResult<ReportView>> {
return Observable.create(observer => {
  this.collectionService.collectionBy(collection)
  const url = this.collectionService.buildUrl(this.reportsUrl)
  const newPage = new PagedResult<ReportView>([], null, null, 0)
  this.getReportPage(url)
    .pipe(
      expand(result => {
        const skip = collection.pageBy.skip + collection.pageBy.top
        collection.pageBy = new PageBy(collection.pageBy.top, skip)
        this.collectionService.collectionBy(collection)
        const nextUrl = this.collectionService.buildUrl(this.reportsUrl)
        const test = result.count >= collection.pageBy.top ? this.getReportPage(nextUrl) : EMPTY
        console.log('test', test)
        return test
      }),
      reduce((next: PagedResult<ReportView>, data: PagedResult<ReportView>, index: number) => {
        next.value = [...next.value, ...data.value]
        next.count = next.value.length
        console.log('next', next, index)
        return next
      }, newPage),
    )
    // .catch(error => observer.error(error))
    .subscribe(results => {
      console.log('results', results)
    })
})

Answer 1

我解决了这个问题，我缺少 takeWhile() 这里是完整的解决方案，以防其他人想做类似的事情。

  getReportsAll(collection: Collection): Observable<PagedResult<ReportView>> {
this.collectionService.collectionBy(collection)
const url = this.collectionService.buildUrl(this.reportsUrl)
const newPage = new PagedResult<ReportView>([], null, null, 0)
return this.getReportPage(url).pipe(
  expand(result => {
    const skip = collection.pageBy.skip + collection.pageBy.top
    collection.pageBy = new PageBy(collection.pageBy.top, skip)
    this.collectionService.collectionBy(collection)
    const nextUrl = this.collectionService.buildUrl(this.reportsUrl)

    return result.count >= collection.pageBy.top ? this.getReportPage(nextUrl) : of(null)
  }),
  takeWhile((value: PagedResult<ReportView> | Observable<void>, index: number) => {
    if (value === null) {
      return false
    }
    return true
  }),
  reduce((next: PagedResult<ReportView>, data: PagedResult<ReportView>, index: number) => {
    next.value = [...next.value, ...data.value]
    next.count = next.value.length
    return next
  }, newPage),
)