File-sharing with sockets

Question

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