A change stream delivers notifications when data changes in a MongoDB collection, database, or deployment. An application can react to inserts, updates, and deletes without repeatedly polling the database or tailing the oplog. It can also use an aggregation pipeline to filter or reshape events before receiving them. section 9.2 section 1.3.1
To watch changes in practice:
- Use a replica set or sharded cluster with WiredTiger and replica set protocol PV1. Choose a collection, database, or deployment scope; system collections and certain databases are excluded. section 9.2 section 9.2.1
- In mongosh, run `watchCursor = db.getSiblingDB("sample_mflix").watch()` to open a database-wide cursor. Insert a document into a nonsystem collection from another connection, then read the resulting event with `watchCursor.tryNext()`. section 9.2.1
- For an application, call `sessionsCollection.watch()` with the MongoDB Node.js driver and iterate over the returned change stream. Pass a pipeline to `watch(pipeline)` when only selected or modified events are needed. section 9.2.2 section 9.2.3
Filter events before processing
A watcher for the `sessions` collection can pass `[{ $match: { 'fullDocument.user_id': '12345' } }, { $addFields: { newField: 'this is an added field!' } }]` to `watch(pipeline)`. The application then receives matching events with the added field, rather than processing every change itself. section 9.2.3
Listing 9.8 Change-stream cursor
{
_id: {
_data: '8266073FD1000000022B022C0100296E5A1004E43
➥C0CEEB6D74D84814318056E6EECDC46645F6964006466073FD1359139DD
➥FE70B1D70004'
},
operationType: 'insert',
clusterTime: Timestamp({ t: 1711751121, i: 2 }),
wallTime: ISODate("2024-03-29T22:25:21.073Z"),
fullDocument: {
_id: ObjectId("66073fd1359139ddfe70b1d7"),
user_id: '12345',
jwt: 'token123456'
},
ns: {
db: 'sample_mflix',
coll: 'sessions'
},
documentKey: {
_id: ObjectId("66073fd1359139ddfe70b1d7")
}
}
Inspect an insert event: `operationType` identifies the action, `fullDocument` contains the inserted document, and `_id` contains the resume token. section 9.2.1
Keep each event’s `_id` intact if the stream may need to resume: supply its token through `resumeAfter` or `startAfter` when reopening the cursor. Drivers attempt to reconnect through a node matching the read preference and raise an exception if none is available. Events are majority-committed, ordered across shards, and subject to role-based access control. Atlas Database Triggers also consume change streams to invoke functions for matching collection changes; local Atlas search uses them to keep Lucene indexes updated. section 9.2.1 section 9.2 section 19.2.1 section 13.1
