如何使用 RxJs 运算符进行递归 HTTP 调用？

Question

我在奇怪的情况下使用以下方法通过为每个 HTTP 调用传递 pageIndex (1) 和 pageSize (500) 来检索数据。

this.demoService.geList(1, 500).subscribe(data => {
    this.data = data.items;
});

响应有一个名为isMore 的属性，如果isMore 为真，我想更奇怪地修改我的方法以继续HTTP 调用。我还需要合并返回的值，最后返回累加的值。

例如，假设有 5000 条记录，并且直到第 10 次 HTTP 调用，服务会为 isMore 值返回 true。在第 10 次 HTTP 调用后，它返回 false，然后此方法将 this.data 值设置为合并的 5000 条记录。对于这个问题，我应该使用mergeMap 或expand 还是另一个RxJs 运算符？解决这个问题的正确方法是什么？

更新：我使用下面的方法，但是它没有合并返回值，也没有增加pageIndex。出于这个原因，它不起作用（我尝试进行一些更改，但无法使其起作用）。

let pageIndex = 0;
this.demoService.geList(pageIndex+1, 500).pipe(
    expand((data) => {
        if(data.isComplete) {
            return of(EMPTY);
        } else {
            return this.demoService.geList(pageIndex+1, 500);
        }
    })
).subscribe((data) => {
    //your logic here
});

更新二：

of({
    isMore : true,
    pageIndex: 0,
    items: []
  }).pipe(
    expand(data => demoService.geList(data.pageIndex+1, 100)
    .pipe(
      map(newData => ({...newData, pageIndex: data.pageIndex+1}))
    )),
    // takeWhile(data => data.isMore), //when using this, it does not work if the total record is less than 100
    takeWhile(data => (data.isMore || data.pageIndex === 1)), // when using this, it causing +1 extra HTTP call unnecessarily
    map(data => data.items),
    reduce((acc, items) => ([...acc, ...items]))
  )
  .subscribe(data => {
    this.data = data;
  });  

更新三：

最后我通过修改 Elisseo 的方法使其工作，如下所示。但是，**我需要将其设为无效并在此 getData() 方法中设置 this.data 参数。我该怎么做？

getData(pageIndex, pageSize) {
  return this.demoService.geList(pageIndex, pageSize).pipe(
    switchMap((data: any) => {
      if (data.isMore) {
        return this.getData(pageIndex+1, pageSize).pipe(
          map((res: any) => ({ items: [...data.items, ...res.items] }))
        );
      }
      return of(data);
    })
  );
}

我想将以下订阅部分合并到这种方法，但由于某些错误，例如，我不能“'void' 类型上不存在属性 'pipe'。”

.subscribe((res: any) => {
    this.data = res;
});

Answer 1

我不认为递归是正确的方法：

interval(0).pipe(
    map(count => this.demoService.getList(count + 1, 500)),
    takeWhile(reponse => response.isMore, true),
    reduce((acc, curr) => //reduce any way you like),
).subscribe();

这应该会调用您的端点，直到端点返回 isMore === false。 interval 的美妙之处在于我们可以免费获得 count 变量。

但是，如果您打算使用 recrsion，这里是使用 expand-operator 的 rxjs 方式（请参阅docs）。我发现它的可读性稍差，因为它需要一个 if-else-construct，这会增加代码的复杂性。外部的“计数器”变量也不是最佳的。

let index = 1;
this.demoService.geList(index, 500).pipe(
    expand(response => response.isMore ? this.demoService.geList(++index, 500) : empty()),
    reduce((acc, curr) => //reduce here)
).subscribe();

Answer 2

只要 api 响应为您提供 isMore 标志，您是否完全更改记录并不重要。

我将跳过如何实现减速器动作事件的部分，我假设您已经完成了该部分。所以我将尝试用伪代码来解释。

您有一个带有分页数据的表格或类似的东西。在初始状态下，您可以创建一个 loadModule 效果或使用此 fn：

getPaginationDataWithPageIndex(pageIndex = 1){ this.store.dispatch(new GetPaginationData({ pageIndex: pageIndex, dataSize: 500})); }

在您的 GetPaginationData 效果中

... map(action => {
return apicall.pipe(map((response)=> {
   if(response.isMore){
    return new updateState({data:response.data, isMore: responseisMore})
} else {
   return new updateState({isMore: response.isMore}),
}
}})
`

all you have to left is subscribing store in your .ts if isMore is false you will not display the next page button. and on your nextButton or prevButton's click method you should have to just dispatch the action with pageIndex

Answer 3

getData(pageIndex, pageSize) {
    return this.demoService.getList(pageIndex, pageSize).pipe(
      switchMap((data: any) => {
        if (!data.isCompleted) {
          return this.getData(pageIndex+1, pageSize).pipe(
            map((res: any) => ({ data: [...data.data, ...res.data] }))
          );
        }
        return of(data);
      })
    );
  }

stackblitz 注意：在我写 pageIndex++ 之前，我更新了作为参数 pageIndex+1 的 @mbojko 建议

更新 2

使用扩展运算符，我们需要考虑到我们需要用一个带有 pageIndex 的对象来提供“递归函数”——这在我们的调用中是必需的——为此，当我们制作 this.demoService.getList(data.pageIndex+1,10) 时，我们需要“转换结果”添加一个新的属性“pageIndex”。为此，我们使用“地图”

  getData() {
    //see that initial we create "on fly" an object with properties: pageIndex,data and isCompleted
    return of({
      pageIndex:1,
      data:[],
      isCompleted:false
    }).pipe(
      expand((data: any) => {
        return this.demoService.getList(data.pageIndex,10).pipe(
            //here we use map to create "on fly" and object
            map((x:any)=>({
              pageIndex:data.pageIndex+1, //<--pageIndex the pageIndex +1
              data:[...data.data,...x.data], //<--we concatenate the data using spread operator
              isCompleted:x.isCompleted}))  //<--isCompleted the value
        )
      }),
      takeWhile((data: any) => !data.isCompleted,true), //<--a take while
            //IMPORTANT, use "true" to take account the last call also
      map(res=>res.data)  //finally is we only want the "data" 
                          //we use map to return only this property
    )
  }

我们可以做一个这样的函数：

  getData() {
    of({pageIndex:1,data:[],isCompleted:false}).pipe(
      expand((data: any) => {
        return this.demoService.getList(data.pageIndex,10).pipe(
            tap(x=>{console.log(x)}),
            map((x:any)=>({
              pageIndex:data.pageIndex+1,
              data:[...data.data,...x.data],
              isComplete:x.isComplete}))
        )
      }),
      takeWhile((data: any) => !data.isComplete,true), //<--don't forget the ",true"
    ).subscribe(res=>{
       this.data=res.data
    })
  }

请注意，在这种情况下，我们不返回 else 简单订阅函数并将变量 this.data 与 res.data 相等 - 这就是我们不需要最后一个映射的原因

更新 3Mrk Sef

最后，如果您不希望您的流发出间歇性的值，而您只想要最终的串联数据，您可以从expand 中删除数据串联，然后改用reduce。

  getData() {
    of({
      pageIndex: 1,
      data: [],
      isCompleted: false
    })
      .pipe(
        expand((prevResponse: any) => this.demoService.getList(prevResponse.pageIndex, 10).pipe(
            map((nextResponse: any) => ({
              ...nextResponse,
              pageIndex: prevResponse.pageIndex + 1
            }))
          )
        ),
        takeWhile((response: any) => !response.isCompleted, true),
        // Keep concatenting each new array (data.items) until the stream
        // completes, then emit them all at once
        reduce((acc: any, data: any) => {
          return [...acc, ...data.data];
        }, [])
      )
      .subscribe(items => {
        this.data=items;
      });
  }