node.js - 使用nodejs在mongoDB oplog集合中查找每个新文档不起作用(光标似乎变得无效或死了)

Question

我想要实现的想法是在正在创建新文档(例如插入数据库集合)时读取(查找)mongoDB oplog 集合。

这是我的代码简化:

var MongoClient = require('mongodb').MongoClient;

MongoClient.connect(url, function(err, db) {
if(err){console.error("ERROR",err); return;}

console.log("Connected correctly to server");

db.collection('oplog.rs').find({
        ns: 'cabo_dev.documents',
        op: 'i',
        // ts: {
        //  $gte: $gte
        // }
    }, {
        tailable: true
    })
    .each(function (err, entry) {
        if (err) {
            console.error("Error fetching a document", err, entry);
            return;
        }

        console.log('--- entry', entry);
    });
});

我已经评论了 $gte 值以简化，但这个想法是阅读所有"new"日志，而不是旧日志。我也有类似的代码使用 Mongoose ，而不是原始驱动程序。

根据文档，前面的代码会将代表 insert 的所有 oplog 文档返回到 cabo_dev.documents 集合中(cabo_dev 是 db 的名称)和所有新的后续插入。但是，当它完成返回文档(“旧”文档)并且没有更多文档要返回时，它将下一个输出作为错误返回(每个中的 err var):

{ [MongoError: No more documents in tailed cursor]
  name: 'MongoError',
  message: 'No more documents in tailed cursor',
  tailable: true,
  awaitData: true }

之后，它不再从 oplog 获取更多插入。根据tainlable documentation ，游标变为死或无效的原因之一是:

• 查询没有返回匹配项。

这是我认为发生在我身上的事情。然而，在那种情况下，each 进程永远不会结束(这是我希望当光标变为无效或无效时的情况，不是吗？)。但是我真的很想继续获取后续的插入日志。

我做错了什么？

Answer 1

也许您现在已经找到了答案。但我写这个答案只是为了让任何偶然发现同样问题的人都能得到解决。

就我而言mongodb驱动版本是2.0.33

与 mongodb 服务器建立连接后，请执行以下操作:

db.collection('yourCappedColl', function (err, coll) {
                var stream = coll.find({},
                    {
                        tailable: true,
                        awaitdata: true,
                        numberOfRetries: Number.MAX_VALUE
                    }).stream();

                stream.on('data', function(val) {
                    console.log('Doc: %j',val);
                });

                stream.on('error', function(val) {
                    console.log('Error: %j', val);
                });

                stream.on('end', function(){
                    console.log('End of stream');
                });
        });

即:

1. 使用 stream() 而不是 each()
2. 将 numberOfRetries 设置为 Number.MAX_VALUE
3. 将 awaitData 设置为 true

有关更多详细信息，请参阅此 jira 项目:

MongoError: No more documents in tailed cursor