如何跨两个 RxJS 可观察对象保留“完整”事件？

Question

我有一个可观察的const numbers = from([1,2,3])，它将发出 1、2、3，然后完成。

我需要将其映射到另一个可观察对象，例如像这样：

const mapped = numbers.pipe(
   concatMap(number => Observable.create(observer => {
     observer.next(number);
   }))
 );

但现在生成的 observable mapped 发出 1、2、3 但不是完整的事件。

如何在mapped 中保存完整的事件？

Answer 1

您的代码只给了我“1”（使用 RxJS 6）；你确定你看到了 3 个值吗？

Rx.from([1,2,3]).pipe(
   op.concatMap(number => Rx.Observable.create(observer => {
     observer.next(number);
   }))
 ).forEach(x => console.log(x)).then(() => console.log('done'))

你永远不会完成创建的 Observable（它会发出一个值但永远不会调用 observer.complete()）。这有效：

Rx.from([1,2,3]).pipe(
   op.concatMap(number => Rx.Observable.create(observer => {
     observer.next(number); observer.complete();
   }))
 ).forEach(x => console.log(x)).then(() => console.log('done'))

这一切都表明正确使用Rx.Observable.create() 是多么困难。使用 Rx 的目的是使用更高级别的抽象来编写代码。其中很大一部分是更喜欢使用运算符而不是观察者。例如。在你的情况下（这很简单）：

Rx.from([1,2,3])
  .pipe(op.concatMap(number => Rx.of(number)))
  .forEach(x => console.log(x)).then(() => console.log('done'))