RxJS：随时间动态组合多个可观察对象答案

【问题标题】：RxJS: combine multiple observables dynamically over timeRxJS：随时间动态组合多个可观察对象
【发布时间】：2018-02-04 22:02:49
【问题描述】：

我的问题是递归的，不知道什么时候我的 Observable complete() 在递归函数中。

基本上，我的功能是废弃一个具有分页功能的网站，它返回一个 Observable，它使用 .next(data) 方法推送在每个页面上找到的“已解析”项目，并递归执行此操作，直到我们发现自己在最后一个页面，然后我们触发我们的主题的.complete() 方法。

我正在使用concat(Observable) 方法将新的 Observable 与当前的 Observable 递归连接，但它似乎不起作用，当我订阅我的 observable 时，我只得到第一页的项目，这让我猜测concat() 方法不适用于我的情况。

这是我的函数代码的简化版本。

```

crawlList(url) {
    let obsSubject = new Subject();
    request(url, function (err, response, body) {
        //check if everything is alright...
        //parsing data...
        obsSubject.next(parsedData);
        //now we check if we can still paginate (we are not in the last page)
        //if so, we concat our observable with the new observable recursivly
        obsSubject.concat(crawList(url))
        //else ,(we are in the last page
        obsSubject.complete();
    });
    return obsSubject;
}

```

【问题讨论】：

concat 不会修改 Observable/Subject，它会返回一个新的。试试obsSubject = obsSubject.concat(crawList(url));

标签： node.js asynchronous recursion rxjs

【解决方案1】：

一般避免使用Subjects，除非你确定你不能使用操作符。

在这种情况下，我希望 expand 可以在这里工作。将前一个流的结果反馈给操作符，以便递归执行。

类似：

// Convert the callback into an Observable
const rxRequest = Rx.Observable.bindNodeCallback(
  request,
  (response, body) => ({response, body})
);    

// Feed the initial data set, by default we should continue
Observable.of({url: baseUrl, shouldContinue: true})
  .expand(({url, shouldContinue}) => {
    // Return an empty stream to cancel the Observable
    // note this is from the *previous* iteration
    if (!shouldContinue)
      return Rx.Observable.empty();

    // This is how we call the newly created method
    return rxRequest(url)
      .map(({response, body}) => 
        // Parse data
        // Check if you should continue or not
        // We still need to emit this data so we can't cancel until the next
        // go-around
        ({url: newUrl, data: parsedData, shouldContinue})
      );
  })
  // Downstream only cares about the data part
  .pluck('data')

【讨论】：