Building simple nodejs API with streaming JSON out

问题:

I am trying to build a simple node.js based streaming API. All I want to do is as I hit the server url, the output should stream a set of test data(JSON) like twitter streaming API.

var app = require('express')();
var server = require('http').Server(app);
var io = require('socket.io')(server);

server.listen(8083);

app.get('/', function (req, res) {
    res.write(io.on('connection', function (socket) {
              socket.emit('item', { hello: 'world' });
    }));
});

So, If i do curl http://localhost:8083/, I want output something like:

$ curl http://localhost:8083/
{hello: 'world'}
{hello: 'world'}
{hello: 'world'}
{hello: 'world'}
...

I am new to node.js and web sockets. I might be horribly wrong on the basics of how node works, let me know the best solution.

回答1:

First, it's better to put the JSONStream part inside a middleware like so:

var _ = require('lodash');
// https://github.com/smurthas/Express-JSONStream/blob/master/index.js
function jsonStream(bytes) {
  return function jsonStream(req, res, next) {
    // for pushing out jsonstream data via a GET request
    var first = true;
    var noop = function () {};
    res.jsonStream = function (object, f) {
      f = _.isFunction(f) ? f : noop;
      if (!(object && object instanceof Object)) {
        return f();
      }

      try {
        if (first) {
          first = false;
          res.writeHead(200, {
            'Content-Type': 'application/json',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive'
          });
        }
        res.write(JSON.stringify(object) + '\n');
      } catch (err) {
        return _.defer(f.bind(null, err));
      }
      f();
    };
    next();
  };
}

Then, let's say you want to be notified via this API each time someone connects to socket.io

var app = require('express')();
var server = require('http').Server(app);
var io = require('socket.io')(server);
var _ = require('lodash');
var EventEmitter = require('events').EventEmitter;
server.listen(8083);

var mediator = new EventEmitter();

io.on('connection', function (socket) {
  mediator.emit('io:connection:new', socket);
});

// the second parameter, specify an array of middleware, 
// here we use our previously defined jsonStream
app.get('/', [jsonStream()], function (req, res) {

  function onNewConnection(socket) {
    res.jsonStream({
      type: 'newConnection',
      message: 'got a new connection',
      socket: {
        id: socket.id
      }
    });
  }

  // bind `onNewConnection` on the mediator, we have to use an mediator gateway
  // because socket.io does not offer a nice implementation of "removeListener" in 1.1.0
  // this way each time someone will connect to socket.io
  // the current route will add an entry in the stream
  mediator.on('io:connection:new', onNewConnection);

  // unbind `onNewConnection` from the mediator
  // when the user disconnects
  req.on('close', function () {
    mediator.removeListener('connection', onNewConnection);
  });

  res.jsonStream({
    type: 'welcome',
    message: 'waiting for connection'
  });
});

Finally, if you want to test this code without connecting to socket.io use the following simulator:

// Simulate socket.io connections using mediator
(function simulate() {
  var dummySocket = {
    id: ~~(Math.random() * 1000)
  };
  mediator.emit('io:connection:new', dummySocket);
  setTimeout(simulate, Math.random() * 1000);
})();