RxJS回调中使用的Observable是异步的

Question

我在我的rxjs流程中有一个地图操作

streaming.map((data) => {
  //example async call
  methodCall.then((response) => {
    return data.test
  })
})
.filter((value) => ...);

问题在于过滤器在数据.test被自然返回之前就被调用了。因此，我尝试将其转换为switchMap并返回一个可观察对象。

streaming.switchMap((data) => {
  return Observable.create((observer) => {
    //example async call
    methodCall.then((response) => {
      observer.next(data.test);
      observer.complete();
    });
  });
})
.filter((value) => ......);

我认为在这种情况下，由于我们返回的是一个带有明确时间的可观察对象，因此只有在观察者完成调用后才会调用筛选器。但是，在值变量未定义的情况下，筛选器提前被调用了。

通常使用rxJs，我该如何解决这个问题？

- MilindaD

Answer 1

我认为您可能缺少了一个返回语句或其他内容，这就是为什么会出现意外结果的原因。 switchMap() 操作符订阅从其回调返回的 Observable，并重新发出它的所有项，直到返回另一个 Observable。它不会等待完成通知。

以下代码应该模拟您的示例：

function methodCall() {
  return new Promise(resolve => setTimeout(() => {
    resolve(123);
  }, 1000));
}

Observable.of(42)
  .do(value => console.log('start: ' + value))
  .switchMap((data) => {
    return Observable.create((observer) => {
      //example async call
      methodCall().then((response) => {
        observer.next(response);
        observer.complete();
      });
    });
  })
  .filter(value => true)
  .subscribe(value => console.log('next: ' + value));

请查看实时演示：https://jsbin.com/focili/2/edit?js,console 请注意，当Promise解决时，会发出123。

顺便说一下，也许你根本不需要使用switchMap()。RxJS 5将Observables、Promises、数组、类似数组的对象等视为同等处理方式。这意味着您可以将Observables替换为上述任何内容之一。例如，您可以仅使用concatMap()，结果将是相同的：

function methodCall() {
  return new Promise(resolve => setTimeout(() => {
    resolve(123);
  }, 1000));
}

Observable.of(42)
  .do(value => console.log('start: ' + value))
  .concatMap(value => methodCall())
  .filter(value => true)
  .subscribe(value => console.log('next: ' + value));

请查看实时演示：https://jsbin.com/wipofiv/2/edit?js,console 请注意，concatMap()接收一个Promise，但仍然正常工作，您甚至不需要将其转换为Observable。

- martin

Answer 2

通常情况下，您可以使用fromPromise将Promise转换为Observable。

streaming.switchMap((data) => Observable.fromPromise(methodCall))
         .filter((value) => ...);

这个工具会处理所有繁琐的工作，让你只需享受 RxJS 编程。

在使用 filter 之前，你可能需要应用一些 .map 来返回 data.test 而不是 data

- smnbbrv