javascript - 简化 RxJS 中的嵌套订阅

Question

我有两个可观察对象

clients$ 是客户端连接流

data$ 是一个不断传递数据的流

每当客户端连接时，我都会订阅 data$ 并将其发送给客户端。

我读到嵌套订阅在 RxJS 中是不行的，所以这种方法可能是错误的。

const clients$ = createClientStream()
const data$ = Observable.interval(1000).share()

clients$.subscribe(c => {

  const s = data$.subscribe(d => c.send(d))

  c.on('disconnect', () => s.unsubscribe())

})

惯用的 RxJS 方法是什么？

Answer 1

你可以使用 flatMap .

const clients$ = createClientStream()
const data$ = Observable.interval(1000).share()

clients$
  .flatMap(c => {
     const disconnect$ = Rx.Observable.create (observer => {
       c.on('disconnect', () => {observer.onNext({}); observer.onCompleted();})
     })
     return data$.takeUntil(disconnect$)
  }, (c,d) => c.send(d))