使用 MongoDB Streams 观看文档创建

Question

我想在每次将特定类型的数据插入集合时使用 MongoDB 流来触发事件。

我已经找到了与我正在寻找的内容大致相似的东西，但这仅适用于更改流而不是插入。

知道如何完成这项工作吗？

我正在使用带有 Nodejs 的 Mongodb 驱动程序来完成这项工作，因此我的代码将是这样的：

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

const uri = 'mongodb://localhost:27017/?replicaSet=rs0';
MongoClient.connect(uri, function(err, client) {

const db = client.db('mydb');

// Connect using MongoClient
var filter = [{
    $match: {
        $or: [
            { $or: [{"receipt.receiver": "newdexpocket"}, {"act.account": "newdexpocket"}] }]
    }
}];

var options = { fullDocument: 'updateLookup' };
db.collection('somecollection').watch(filter, options).on('create', data => 
  {
    console.log(data);
  });
});

• 是否需要在过滤器中指定operationType？
• 我还需要获取完整文档，但显然 updateLookup 不是正确的工具，我应该使用什么？
• on 事件可以使用哪些选项？我用过create，但我什至不确定它是否存在，是吗？

很抱歉所有这些问题，但我很难在官方文档中找到一些答案。

解决方案：

请注意不要在您的请求中忘记fullDocument ;-)

function watch_insert(con, db, coll) {
  console.log(new Date() + ' watching: ' + coll);

  const insert_pipeline = [ { $match:
                    { 
                        operationType: 'insert',
                        $or: [
                            { "fullDocument.receipt.receiver": "newdexpocket" },
                            { "fullDocument.act.account": "newdexpocket" }
                        ]
                    }
                }];

  con.db(db).collection(coll).watch(insert_pipeline)
    .on('change', data => {
      console.log(data)
    });
}

async function run(uri) {
    try {
        con = await MongoClient.connect(uri, {"useNewUrlParser": true});
        watch_insert(con, 'EOS', 'action_traces');
    } catch (err) {
        console.log(err);
    }
}

Answer 1

您需要：

1. 指定operationType: 'insert'。由于您不想监控更新，因此不需要updateLookup。
2. 为包含operationType 的过滤器创建适当的aggregation pipeline。
3. 聚合管道过滤由watch() 返回的文档。 Change Events page 中有一个示例输出。

watch() 返回一个ChangeStream。它触发close、change、end 和error 事件。详情请见ChangeStream。

这是一个完整的变更流示例，它侦听数据库test 上的insert 操作@ 集合test。它将输出具有字段{a: 1} ('fullDocument.a': 1) 的文档，并忽略更新、插入其他a 值或任何没有字段a 的内容。

const MongoClient = require('mongodb').MongoClient
const uri = 'mongodb://localhost:27017/test?replicaSet=replset'

const insert_pipeline = [
  {$match: {operationType: 'insert', 'fullDocument.a': 1}}
]

function watch_insert(con, db, coll) {
  console.log(new Date() + ' watching: ' + coll)
  con.db(db).collection(coll).watch(insert_pipeline)
    .on('change', data => {
      console.log(data)
    })
}

async function run() {
  con = await MongoClient.connect(uri, {"useNewUrlParser": true})
  watch_insert(con, 'test', 'test')
}

run()