nodejs中的Mongodb Tailable Cursor，如何停止流

Question

我使用下面的代码从 mongodb capped collection 中获取数据

函数监听（条件，回调）{

        db.openConnectionsNew( [req.session.client_config.db] , function(err, conn){

            if(err) {console.log({err:err}); return next(err);}

            coll = db.opened[db_name].collection('messages');

        latestCursor = coll.find(conditions).sort({$natural: -1}).limit(1)
        latestCursor.nextObject(function(err, latest) {
            if (latest) {
                conditions._id = {$gt: latest._id}
            }
            options = {
                tailable: true,
                awaitdata: true,
                numberOfRetries: -1
            }
            stream = coll.find(conditions, options).sort({$natural: -1}).stream()
            stream.on('data', callback)
        });

        });
    }

然后我使用 sockets.broadcast(roomName,'data',document);

在客户端

io.socket.get('/get_messages/', function(resp){
});
io.socket.on('data', function notificationReceivedFromServer ( data ) {
  console.log(data);

});

这非常有效，因为我可以看到插入到 db 中的任何新文档。

我可以在 mongod -verbose 中看到，每 200 毫秒后会有一个使用查询 {$gt:latest_id} 运行的查询，这很好，但我不知道如何关闭这个查询 :( 我很新nodejs 和第一次使用 mongodb tailable 选项，我完全迷路了，非常感谢任何帮助或线索

Answer 1

Cursor 对象从.find() 返回的.stream() 方法返回的是node transform stream 接口的实现。具体来说，这是一个“可读”流。

因此，只要接收到新数据并在要读取的流中可用，就会发出它的“数据”事件。

还有其他方法，例如.pause() 和.resume()，可用于控制这些事件的流程。通常，您会将这些“内部”称为“数据”事件回调，您希望确保在处理“下一个”数据事件之前执行该回调中的代码：

stream.on("data", function(data) {
   // pause before processing
   stream.pause();

  // do some work, possibly with a callback
  something(function(err,result) {

      // Then resume when done
      stream.resume();
  });
});

但这当然只是“范围界定”的问题。因此，只要“流”变量定义在另一段代码可以访问它的范围内，那么您可以随时调用任一方法。

同样，通过范围界定，您可以在代码中的任何位置“取消定义”“流”对象，从而使“事件处理”变得多余。

// Just overwrite the object
scope = undefined;

所以值得知道。事实上，较新的“2.x 版”节点驱动程序将“流接口”直接包装到标准的Cursor 对象中，而无需调用.stream() 进行转换。节点流是非常有用和强大的东西，在接受它们的使用时非常值得。