将流中的每一项映射到一个 Promise 并返回它的值

Question

我正在尝试将流的每个项目映射到一个承诺，就像这样

myStream$
  .flatMap(id => Rx.Observable.fromPromise(database.get(id)))
  .subscribe(val => console.log(val));

myStream$ 大约有 15 个项目。由于在数据库中找不到这些项目，因此每个承诺都将被拒绝。我期待 15 个日志输出打印错误。但是我得到的只是一个错误

rx.js:77 未捕获 {"status":404,"name":"not_found","message":"missing","re​​ason":"missing"}

为什么我只收到一个错误而不是 15 个？

Answer 1

这种行为是意料之中的 - 每当抛出错误时，流都会结束（换句话说：“停止并取消订阅所有订阅者”）。

如果您希望流正确完成而不管一个或所有 db-request 失败，您必须处理包装承诺的子流中的错误，因为 reject 将被评估为 error在 RxJS 中。

Rx.Observable.from([1,2,3,4,5])
  .flatMap(
    id => mockRequest(id)
      .catch(error => {
        console.error(error);
        return Rx.Observable.empty();  // here we just return an empty stream, so the "main"-stream won't receive the error and continue with the other ids in the queue
      })
  )
  .subscribe(
    val => console.log(val),
    error => console.error("Stream hit an error and will finalize", error),
    complete => console.log("Done!")
  );
  
function mockRequest(id) {
  return Rx.Observable.throw("Request failed for: " + id);
}

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

注意：您当然可以通过 Observable.of("myFallbackValue") 返回任何备用值，而不是 return Rx.Observable.empty();

---

旁注和建议：当您直接使用 RxJS 进行休息调用而不需要包装承诺时，您会更容易。 （虽然从技术上讲，两者都是完全有效的方式）