Change Streams

Change Streams

Watching Data with Change Streams

watch() opens a tailable cursor over the oplog and hands you one event per committed change. It exists on a collection, a database and the client itself, and takes an aggregation pipeline, so you filter events on the server rather than in your process. Only majority-committed changes are reported, so no event is taken back. An update event carries just the delta by default; ask for fullDocument: 'updateLookup' to get the current document, or enable pre- and post-images and use fullDocumentBeforeChange, the only way to see a deleted one.

Insert, update and delete events from one collection

await db.createCollection('orders', { changeStreamPreAndPostImages: { enabled: true } });
const stream = db.collection('orders').watch([], { fullDocument: 'updateLookup',
  fullDocumentBeforeChange: 'whenAvailable' });
stream.on('change', e => console.log(e.operationType, JSON.stringify(e.fullDocument),
  JSON.stringify(e.updateDescription), JSON.stringify(e.fullDocumentBeforeChange)));
await orders.insertOne({ _id: 7, item: 'kettle', qty: 1, status: 'new' });
await orders.updateOne({ _id: 7 }, { $set: { status: 'paid' }, $inc: { qty: 2 } });
await orders.deleteOne({ _id: 7 });

Output

insert {"_id":7,"item":"kettle","qty":1,"status":"new"} undefined undefined
update {"_id":7,"item":"kettle","qty":3,"status":"paid"} {"updatedFields":{"qty":3,
  "status":"paid"},"removedFields":[],"truncatedArrays":[]} {"_id":7,"item":"kettle",
  "qty":1,"status":"new"}
delete undefined undefined {"_id":7,"item":"kettle","qty":3,"status":"paid"}

updateDescription is the useful part of an update event: it names exactly the fields that changed, which is what you want when invalidating a cache or pushing a patch to a browser. Every event also carries _id (the resume token), operationType, clusterTime, wallTime and ns. Pre- and post-images live in config.system.preimages and cost write throughput and disk, so enable them per collection and set a retention with the changeStreamOptions cluster parameter. Time series collections support none.

MENU