javascript - 使用 RxJS 对 WebSocket 流建模

Question

使用 RxJS 对 WebSocket 流进行建模的方法有哪些。

我看到的最明显的东西是套接字流，它们发出消息流。

如果我创建一个套接字流，我如何创建它们的消息流并仍然保留发送这些消息的人？

套接字流是我的第一步:

const socket$ = Observable.create(({complete, next}) => {

  const server = new WebSocketServer({server: someHttpServer})

  server.on('connection', next)

  return () => {
    server.close()
    complete()
  }

})

但是消息流有点难，因为我需要从中获取消息的套接字。

这是我第一次天真的建模尝试:

const message$ = socket$.flatMap(socket => Observable.create(({complete, next}) => {

  socket.on('message', next)
  socket.on('close', complete)

  return () => socket.close()

})).share()

从所有套接字流式传输所有套接字消息的可观察对象。但是如果我订阅它，我就没有套接字了，这使得它是单向的。

我要

socket$ -> message$ -> server-processing -> socket$

但是响应、广播、多播和单播有多个用例。

Answer 1

我发现 flatMap 采用第二个函数，它接收 flatMap 的参数值和 flatMap 的(扁平化)返回值>。此函数可以返回一个新值，供以后的所有运算符使用。

const socketMessage$ = socket$.flatMap(

  socket => Observable.create(({complete, next}) => {

    socket.on('message', next)
    socket.on('close', complete)

    return () => socket.disconnect()

  }),

  (socket, message) => ({socket, message})

).share()