How to use RxJs with Socket.IO on event

Question

I want to use RxJS inside of my socket.on('sense',function(data){});. I am stuck and confused with very few documentation available and my lack of understanding RxJS. Here is my problem.

I have a distSensor.js that has a function pingEnd()

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

Inside my App.js I have

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

The sense function gets lots of sensor data which I want to filter using RxJS and I don't know what should I do next to use RxJs here. Any pointers to right docs or sample would help.

Answer 1

You can use Rx.Observable.fromEvent (https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/fromevent.md).

Here's how I did a similar thing using Bacon.js, which has a very similar API: https://github.com/raimohanska/bacon-minsk-2015/blob/gh-pages/server.js#L13

So in Bacon.js it would go like

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

And in RxJs you'd replace Bacon.fromEvent with Rx.Observable.fromEvent.