如何在RxJS Observable内部追加数据实现顺序分页HTTP请求结果合并

你这个场景完全可以用RxJS内置的expand操作符实现，它天生适配需要根据前一次响应结果触发下一次请求的递归分页场景，不需要手动维护外部状态和自定义Observable包装，代码如下：

import { from, EMPTY, expand, concatMap, type Observable } from 'rxjs';

// 最终输出的Observable，发射类型为SteamItem
const steamItem$: Observable<SteamItem> = 
  // 首次请求，lastAssetId初始为undefined
  from(this.getBatch(/* 你的请求参数 */, undefined))
  .pipe(
    // 核心递归分页逻辑
    expand(prevBatch => {
      // 没有下一页时返回EMPTY终止递归
      if (!prevBatch.more_items) return EMPTY;
      // 携带上一页的lastAssetId发起下一次请求
      return from(this.getBatch(/* 你的请求参数 */, prevBatch.last_assetid));
    }),
    // 展开每个批次的商品数组，逐个发射单个SteamItem，和原异步生成器的yield*行为完全一致
    concatMap(batch => batch.getSteamItemsWithDescription())
  );

实现说明 #

• 如果你的this.getBatch方法本身已经返回Observable，可以直接去掉外层的from包装，直接传入初始请求即可
• expand操作符会自动递归处理分页逻辑，不需要手动维护moreItems、lastAssetId等外部可变变量
• 所有请求严格按顺序发起，后一次请求会等前一次请求返回后才触发，符合你的依赖要求
• 天然支持RxJS的生态能力，比如需要请求失败重试只需在pipe中加retry(3)，需要异常拦截加catchError即可，不需要修改核心逻辑

相比你当前版本的优势 #

• 完全符合响应式编程范式，没有嵌套的异步自执行函数
• 没有手动管理订阅者和循环逻辑，避免了潜在的内存泄漏和状态异常问题
• 代码更简洁，核心逻辑仅需几行即可实现，可维护性更高

内容的提问来源于stack exchange，提问作者Tim