I'm trying to get some confidence with RXJS and FRP. This code handles file sharing via socket.io. I need some advice about general RXJS style and reactive programming.
io.sockets.on('connection', function (socket) {
var msgStream = Rx.Observable.fromEvent(socket, 'message');
msgStream.filter(function(x) {
return x.type === 'get';
}).do(function() {
socket.send({type: 'start'});
}).do(function() {
msgStream.filter(function (x) { return x.type === 'next'; })
.zip(
RxNode.fromReadableStream(
fs.createReadStream('file.mpg')
),
function(a, b) {·
return [a, b];
}
)
.pluck('1')
.do(function(data) {
socket.send({type: 'chunk', data: data});
}).subscribe(undefined, undefined, function() {
socket.send({type: 'end'});
});
}).subscribe();
});