订阅来自服务器的事件流答案

【问题标题】：Subscribe to a stream of events from server订阅来自服务器的事件流
【发布时间】：2017-08-11 15:02:30
【问题描述】：

我正在扩展 redux-observable 的 ping-pong 示例，我的最终目标是在从服务器接收到的每个事件上调度一个操作。

但是我在思考如何真正实现这一点时遇到了一些麻烦。

到目前为止我有什么：

1.打开连接后，我的服务器开始发送消息。

// server.js
setInterval(()=>{
    ws.send(JSON.stringify({
        type: 'INCREMENT',
        status: 200
    }))
    console.log("sent some data")
},3000)

2。在客户端上，我已经建立了该 Websocket 连接的 Observable。

const socket$ = Observable.webSocket("ws://localhost:8081")

其余代码与JSBin Example for react类似

我如何为这个任务形成一个史诗？我如何调度一个动作？

【问题讨论】：

github.com/Reactive-Extensions/RxJS/issues/112
@eenagy 是 rxjs v4 顺便说一句，对于 v5 不准确
嘿@n3u3w3不清楚你在问什么。可以修改一下吗？
@jayphelps ，当然。基本上，我需要我的应用程序执行以下操作： 1. 打开一个通过 websocket 连接到服务器的应用程序，应用程序创建该 websocket 连接的 observable。 2. 服务器向任何连接的用户发送无限的“INCREMENT”消息。 3. 收到每条消息后，必须发送一个动作。因此，如果我的state.counter 为零，则在收到事件后它必须变为 1,2,3 等等
你能描述一下 websocket 是如何与这个史诗交互的吗？你需要通过套接字发送东西吗？收东西回来？只听服务器？您需要多路复用还是只需要一个通道？

标签： rxjs redux-observable

【解决方案1】：

我们在 cmets 中讨论了一点，但我恐怕仍然不完全清楚增量/计数器/INCREMENT_APPLY 与套接字的关系，但我可以举一个简单的例子：

const somethingEpic = action$ =>
  action$.ofType('START_SOCKET_OR_WHATEVER')
    .switchMap(action =>
      Observable.webSocket('ws://localhost:8081')
        .map(response => ({ type: 'RECEIVED_MESSAGE', paylod: response }))
    );

在这里，当START_SOCKET_OR_WHATEVER 被调度时，我们将开始监听我们的套接字。每当收到一条消息时，我们都会将其映射到 RECEIVED_MESSAGE 动作中，该动作将在史诗发出后调度。

您会注意到这与您执行单个 ajax 的方式几乎相同。如果您需要将消息发送到服务器或多路复用，它只会变得明显不同。

【讨论】：

非常感谢，@jayphelps，这正是我正在寻找的行为。我刚刚检查了它，它按预期工作 - 我通过调度 SOCKET_START 操作打开一个连接，在调度操作之后，我的应用程序开始接收来自服务器的 ping 并增加计数器值
哇哦！真高兴！
@jayphelps 我不认为你能另外指出正确的方向来监听套接字故障/关闭事件？例如{ type: 'SOCKET_CLOSED' }
@JoshMc webSocket 主题将完成（）或错误（）订阅者，具体取决于它是否干净关闭。您也可以提供 closeObserver 。如果您需要更具体的帮助，例如如何将其转换为您发出的 redux 操作，最好创建一个新的 stackoverflow 问题来描述您的需求。希望这可以帮助！ github.com/ReactiveX/rxjs/blob/…
@jayphelps 感谢您的帮助，我已经尝试过了，但没能走得很远，今晚晚些时候再试一次，但同时我在这里创建了一个问题：stackoverflow.com/questions/57700859/…