我如何附加到可观察对象本身内部的可观察对象答案

【问题标题】：How do I append to an observable inside the observable itself我如何附加到可观察对象本身内部的可观察对象
【发布时间】：2021-11-13 02:09:14
【问题描述】：

我的情况如下：我正在执行顺序 HTTP 请求，其中一个 HTTP 请求依赖于前一个的响应。我想将所有这些 HTTP 请求的响应数据合并到一个 observable 中。在使用异步生成器之前，我已经实现了这一点。这个代码比较简单：

async function* AsyncGeneratorVersion() {
  let moreItems = true; // whether there is a next page
  let lastAssetId: string | undefined = undefined; // used for pagination
  while (moreItems) {
    // fetch current batch (this performs the HTTP request)
    const batch = await this.getBatch(/* arguments */, lastAssetId);
    moreItems = batch.more_items;
    lastAssetId = batch.last_assetid;
    yield* batch.getSteamItemsWithDescription();
  }
}

我正在尝试远离异步生成器，而转向 RxJs Observables。我最好的（和工作的）尝试如下：

const observerVersion = new Observable<SteamItem>((subscriber) => {
  (async () => {
    let moreItems = true;
    let lastAssetId: string | undefined = undefined;
    while (moreItems) {
      // fetch current batch (this performs the HTTP request)
      const batch = await this.getBatch(/* arguments */, lastAssetId);
      moreItems = batch.more_items;
      lastAssetId = batch.last_assetid;
      const items = batch.getSteamItemsWithDescription();
      for (const item of items) subscriber.next(item);
    }
    subscriber.complete();
  })();
});

现在，我相信一定有一些方法可以改进这个 Observer 变体 - 这段代码对我来说似乎不是很被动。我使用pipe 尝试了几件事，但不幸的是这些都没有成功。我发现concatMap 接近解决方案。这让我可以将下一个 HTTP 请求连接为可观察的（使用 this.getBatch 方法完成），但是我找不到不放弃当前 HTTP 请求响应的好方法。

如何做到这一点？简而言之，我相信这个问题可以描述为将数据附加到可观察对象本身内部的可观察对象。（但也许这不是处理这种情况的好方法）

【问题讨论】：

标签： rxjs reactivex

【解决方案1】：

TLDR;

Here 是一个有效的 StackBlitz 演示。

说明

这是我的方法：

// Faking an actual request
const makeReq = (prevArg, response) =>
  new Promise((r) => {
    console.log(`Running promise with the prev arg as: ${prevArg}!`);
    setTimeout(r, 1000, { prevArg, response });
  });

// Preparing the sequential requests.
const args = [1, 2, 3, 4, 5];

from(args)
  .pipe(
    // Running the reuqests sequantially.
    mergeScan(
      (acc, crtVal) => {
        // `acc?.response` will refer to the previous response
        // and we're using it for the next request.
        return makeReq(acc?.response, crtVal);
      },
      // The seed(works the same as `reduce`).
      null,
      // Making sure that only one request is run at a time.
      1
    ),

    // Combining all the responses into one object
    // and emitting it after all the requests are done.
    reduce((acc, val, idx) => ({ ...acc, [`request${idx + 1}`]: val }), {})
  )
  .subscribe(console.warn);

首先，from(array) 会从数组中同步地、一个一个地发出每一项。

然后，有mergeScan。这正是结合scan 和merge 的结果。使用scan，我们可以累积值（在这种情况下，我们使用它来访问上一个请求的响应）并且merge 所做的是允许我们使用observables。
为了让事情更容易理解，想想Array.prototype.reduce 函数。它看起来像这样：

[].reduce((acc, value) => { return { ...acc }}, /* Seed value */{});

merge 在mergeScan 中所做的是允许我们使用累加器 类似(acc, value) => new Observable(...) 而不是return { ...acc }。后者表示同步行为，而前者我们可以有异步行为。

让我们一步一步来：

当1 发出时，makeReq(undefined, 1) 将被调用
在第一个makeReq（从上面）解析后，makeReq(1, 2) 将被调用
makeReq(1, 2) 解析后，makeReq(2, 3) 将被调用，依此类推...

【讨论】：

感谢您的回答！这种方法对我来说很有意义。我发布了一个不同的解决方案，我认为它可能更优雅一些。它使用 expand 而不是 mergeScan 来实现几乎相同的目标

【解决方案2】：

我就此事咨询过的人提出了这个解决方案，我认为它非常优雅：

defer(() => this.getBatch(options)).pipe(
  expand(({ more_items, last_assetid }) =>
    more_items
      ? this.getBatch({ ...options, startAssetId: last_assetid })
      : EMPTY,
  ),
  concatMap((batch) => batch.getSteamItemsWithDescription()),
);

据我了解，这里使用expand 与@Andrei 的答案中使用mergeScan 非常相似

【讨论】：