如何在事件中使用RxJs和Socket.IO

Question

我想在我的内部使用RxJS socket.on('sense',function(data){});.我很困惑并且很少使用可用的文档和我对RxJS缺乏了解.这是我的问题.

我有一个distSensor.js函数pingEnd()

function pingEnd(x){
socket.emit("sense", dist); //pingEnd is fired when an Interrupt is generated.
}

我的App.js里面有

io.on('connection', function (socket) {
    socket.on('sense', function (data) {
        //console.log('sense from App4 was called ' + data);
    });
});

感知功能获取了大量我想用RxJS过滤的传感器数据,我不知道接下来我应该怎么做才能使用RxJs.任何指向正确文档或示例的指针都会有所帮助.

Answer 1

我使用fromEvent方法遇到了一些奇怪的问题,所以我更喜欢创建自己的Observable:

function RxfromIO (io, eventName) {
  return Rx.Observable.create(observer => {
    io.on(eventName, (data) => {
        observer.onNext(data)
    });
    return {
        dispose : io.close
    }
});

我可以这样使用:

let $connection = RxfromIO(io, 'connection');

Answer 2

我想你可以使用Rx.Observable.fromEvent(https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/fromevent.md).

以下是我使用Bacon.js做类似的事情,它有一个非常相似的API:https://github.com/raimohanska/bacon-minsk-2015/blob/gh-pages/server.js#L13

所以在Bacon.js中它会像

io.on('connection', function(socket){
  Bacon.fromEvent(socket, "sense")
    .filter(function(data) { return true })
    .forEach(function(data) { dealWith(data) })
})

而在RxJs你替换Bacon.fromEvent用Rx.Observable.fromEvent.

Answer 3

您可以像这样创建一个Observable:

var senses = Rx.Observable.fromEventPattern(
    function add (h) {
      socket.on('sense',h);
    }
  );

然后senses像任何其他Observable一样使用.