如何在 NodeJs 中正确地将 WS 服务器转换为没有主题的 RXJS Api答案

【问题标题】：How to properly turn WS Server to RXJS Api without Subjects in NodeJs如何在 NodeJs 中正确地将 WS 服务器转换为没有主题的 RXJS Api
【发布时间】：2015-02-19 02:59:53
【问题描述】：

将著名的 ws 模块转换为 Node.js 中的反应式 api 的正确方法是什么？我知道主题可以帮助将非反应性事件与反应性事件联系起来，但他们的问题是他们在处理依赖对象时要困难得多。

var WebSocketServer = require('ws').Server;
var wss = new WebSocketServer({ port: 8080 });
var Rx = require('rx');  


var connectionMessageSubject = new Rx.Subject();

wss.on('connection', function connection(client) {
  ws.on('message', function incoming(message) {
    connectionMessageSubject.onNext({
      client: client,
      message: message
    });
  });
});

我不能使用他们内置的 fromEvent 方法，因为它注册了太多不同的事件，以至于 NodeJS 在 30 个或更多客户端连接时会抛出警告。

例如...

var WebSocketServer = require('ws').Server;
var wss = new WebSocketServer({port:8080});

var connectionMessageObservable;

//this uses a tremendous amount of memory and throws warnings that the event emitter has a maximum of 30 listeners 
wss.on('connection', function connection(client){
  connnectionMessageObservable = Rx.Observable.fromEvent(client, 'message');
});

【问题讨论】：

看看Rx.DOM.fromWebSocket()。您应该能够从本质上克隆该方法并将其更改为与 nodeJs ws 模块一起使用。该模式是实现是一个很好的使用。
FWIW，我不久前重写了很多 fromWebSocket。这种模式非常适合在 Node 中包装 WS 之类的东西。
另外，我有一个 singleInstance 运营商的 PR，如果你发现你必须 publish().refCount() 你的 ws包装器并有时完全取消它。

标签： node.js websocket reactive-programming rxjs

【解决方案1】：

以下代码模拟subject 行为。

var WebSocketServer = require('ws').Server;
var wss = new WebSocketServer({port:8080});

var connectionMessage$ = new Rx.Observable(function (observer) {
    wss.on('connection', function connection(client){
        client.on('message', function (message){
            observer.next({
                client: client,
                message: message,
            })
        });
    });    
});

connectionMessage$.subscribe(function (cm) {
    // cm.client for client
    // cm.message for message
});

【讨论】：