可观察代码的调用顺序是什么？

Question

在下面的代码中，我很难理解链式调用的顺序是：

 function registerDomain(caseWebsiteUrl) {
    return this._adminService.registerDomain(caseWebsiteUrl.Url)
      .concatMap(registerId => Observable.interval(5000)
        .mergeMap(() => this._adminService.getChange(registerId))
        .takeWhile((info) => info.ChangeInfo.Status.Value !== 'INSYNC')
      )
  }
  registerDomain.subscribe(res => console.log('res:'+res));

我试图理解上面的代码，但我很难理解顺序。当我看到这样一个更简单的例子时，我更容易理解：

function registerDomain() {
    return register()
      .concatMap(registerId => Rx.Observable.interval(5000)
        .mergeMap(() => 
          getChange(registerId))
        .takeWhile((info) => info === 'PENDING')
      );
  }

  registerDomain().subscribe(x=>console.log(x));



  function register()
  {
  return Rx.Observable.of("registerId");
  }


  function getChange(registerId)
  {
   //at some point this will return INSYNC
    return Rx.Observable.of("PENDING");
  }

当我看这段代码时，我的阅读方式是：

1. 调用 register() 返回一个 observable
2. 将 getChange() 和 takeWhile 合并到一个 observable 中
3. 将getChange和TakeWhile的合并结果与区间Observable连接起来。
   4. 每 5 秒调用一次 getChange，直到状态 !== 'PENDING'

我的上述评估是否正确？您如何知道何时使用合并或连接？

如果我想捕获错误，例如 register() 抛出错误，那么它应该会失败。

根据@Mark van Straten 的回答，还有一些问题：

1. 所以如果我想保留顺序，我应该使用 concatMap 吗？在这种情况下，我希望 registerDomain 在调用 getChange 之前完成？

2. 如果我不关心顺序，那么使用mergeMap？

3. 在getRegisterStatus 方法中，到底合并了什么？为什么不使用 concatMap？

Answer 1

扁平化高阶可观察对象的三个运算符：

• mergeMap（又名flatMap） - 合并所有流在一起，排放交错
• concatMap - 在第一个流完成后 concat 下一个等
• switchMap - 放弃第一个流，当它到达时切换到下一个流 所以给定代码：

function registerDomain(caseWebsiteUrl) {
    return this._adminService.registerDomain(caseWebsiteUrl.Url)
      .concatMap(registerId => Observable.interval(5000)
        .mergeMap(() => this._adminService.getChange(registerId))
        .takeWhile((info) => info.ChangeInfo.Status.Value !== 'INSYNC')
      )
  }
  registerDomain.subscribe(res => console.log('res:'+res));

我们可以改写如下：

function registerDomain(caseWebsiteUrl) {
  return this._adminService.registerDomain(caseWebsiteUrl.Url)
    .concatMap(registerId => getRegisterStatus(registerId))
}

function getRegisterStatus(registerId) {
  return Observable.interval(5000)
    .mergeMap(() => this._adminService.getChange(registerId))
    .takeWhile((info) => info.ChangeInfo.Status.Value !== 'INSYNC')
}

1. 第一个 _adminService .registerDomain 返回一个 Observable

2. 在它发出一个值后，我们concatMap第二个从getRegisterStatus返回的可观察对象

3. getRegisterStatus 在检索到的信息变为“INSYNC”时完成，将发出每个非 INSYNC 发射。

我认为 concatMap 并不是真正需要的，可以用 mergeMap 代替，因为 registerDomain 很可能只发出一个值