javascript - 我如何知道 Stream(Node.js、MongoDB)何时准备好进行更改？

Question

我想确保 Stream 已准备好进行更改，然后才能将数据发送到客户端。我的代码:

// Get the Stream (MongoDB collection)
let stream = collection.watch()
// Generate identifier to send it to the client
const uuid = uuid()
// Listen for changes
stream
  .on('change', () => {
    // Send something to WebSocket client
    webSocket.emit('identifier', uuid)
  })

// Mutate data base collection to kick off the "change" event
await collection.updateOne()

webSocket.emit 是我的问题。我如何知道 Stream 是否已经准备好接收事件？ change 事件从未发生，因此 webSocket.emit 永远不会被调用。

长话短说

基本上，我需要向客户端发送一些东西，但需要确保 Stream 在此之前准备好接收事件。

Answer 1

这看起来像是一个竞争条件，您的更新查询在 changeStream 聚合管道到达服务器之前执行。基本上，您需要等待设置流游标，然后再触发更改。

我找不到任何“光标准备就绪”事件，因此作为解决方法，您可以检查它的 ID。它由服务器分配，因此当它在客户端可用时，它可以保证捕获所有连续的数据更改。

像这样的东西应该可以完成工作:

async function streamReady(stream) {
    return new Promise(ok => {
        const i = setInterval(() => {
            if (stream.cursor.cursorState.cursorId) {
                clearInterval(i);
                return ok()
            }
        }, 1)
    });
}

然后在你的代码中:

// Get the Stream (MongoDB collection)
let stream = collection.watch()
// Generate identifier to send it to the client
const uuid = uuid()
// Listen for changes
stream
  .on('change', () => {
    // Send something to WebSocket client
    webSocket.emit('identifier', uuid)
  })

await streamReady(stream);    

// Mutate data base collection to kick off the "change" event
await collection.updateOne()

---

免责声明:

上面的streamReady 函数依赖于cursorState。这是一个内部字段，即使在驱动程序的补丁版本更新中也可以更改，恕不另行通知。