node.js - 如何在 NodeJs 中正确地将 WS 服务器转换为 RXJS Api 而无需主题

Question

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

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 方法，因为它注册了太多不同的事件，以至于当 30 个或更多客户端连接时 NodeJS 会抛出警告。

例如...

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');
});

Answer 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
});