使用流式 JSON 输出构建简单的 Nodejs API

我正在尝试构建一个简单的基于 Node.js 的流 API。我想做的就是当我点击服务器 URL 时，输出应该流式传输一组测试数据（JSON），如 Twitter 流 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' });
    }));
});

所以，如果我这样做curl http://localhost:8083/，我想要输出类似：

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

我是 Node.js 和 Web 套接字的新手。我可能对节点工作原理的基础知识有严重错误，请让我知道最好的解决方案。

首先，最好将 JSONStream 部分放入中间件中，如下所示：

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

然后，假设您希望在每次有人连接到 socket.io 时通过此 API 收到通知

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

最后，如果您想在不连接到 socket.io 的情况下测试此代码，请使用以下模拟器：

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