node.js - 观看使用 MongoDB Streams 创建文档

Question

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

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

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

我正在使用 Mongodb 驱动程序和 Nodejs 来完成这项工作，所以我的代码应该是这样的:

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:“插入”。由于您不想监视更新，因此不需要 updateLookup。
2. 创建一个合适的 aggregation pipeline用于包含 operationType 的过滤器。
3. 聚合管道过滤由 watch() 返回的文档。示例输出在 Change Events page 中.

watch() 返回一个 ChangeStream。它会触发 close、change、end 和 error 事件。参见 ChangeStream了解更多详情。

这是一个完整的变更流示例，它监听数据库 test 集合 test 上的 insert 操作。它将输出具有字段 {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()