rxjs - 在 rxjs 中链接两个依赖的 observables

Question

我有一个 observable，它发出一系列消息，比如 obs1。然后是第二个 observable，obs2，它需要来自 obs1 发出的最后一条消息的一些数据，并发出另一系列消息。我想“链接”这两个 observable 以生成一个可观察的 obs3，该 obs3 从 obs1 和 obs2 串行发出所有消息。
到目前为止我想出的解决方案是:

obs3 = concat(
  obs1,
  obs1.pipe(
    last(),
    concatMap(lastMessage => obs2(lastMessage)),
);

但这有一个缺陷，即 obs1 被执行(订阅)2 次。
有没有更直接的方法来实现这一目标？有点像 concatMapWithSelf()像这样工作的运算符:

obs3 = obs1.pipe(
  concatMapWithSelf(lastMessage => obs2(lastMessage)),
);

谢谢!

Answer 1

听起来您可以使用 ConnectableObservable。在 RxJS 7 中，我相信使用 multicast() 会更容易，可读性更好。但这将在 RxJS 8 中被弃用，所以唯一的选择可能是用 connectable() 包装源 Observable |然后手动拨打 connect() .

const obs1 = connectable(
  defer(() => {
    console.log('new subscription');
    return of('v1', 'v2', 'v3', 'v4');
  })
);

const obs2 = msg => of(msg);

const obs3 = merge(
  obs1,
  obs1.pipe(
    last(),
    concatMap(lastMessage => obs2(lastMessage))
  )
);

obs3.subscribe(console.log);

obs1.connect();

现场演示: https://stackblitz.com/edit/rxjs-2uheg4?devtoolsheight=60
如 obs1总是异步的，那么你可能可以使用 share()但这对于同步源的行为会有所不同，因此使用 ConnectableObservable 应该更安全。