Working with Change Streams
1. Opening Change Stream
| Scope | Command |
|---|---|
| Collection | db.coll.watch(pipeline, opts) |
| Database | db.watch() |
| Cluster | client.watch() |
2. Filtering Change Events
| Stage | Detail |
|---|---|
| $match | First stage; filter by operationType, ns, fullDocument fields |
| $project | Reshape change events |
db.orders.watch([{ $match: { operationType: { $in: ["insert","update"] } } }]);3. Understanding Change Event Structure
| Field | Detail |
|---|---|
| _id | Resume token |
| operationType | insert|update|replace|delete|drop|rename|invalidate |
| fullDocument | Current doc (with option) |
| updateDescription | updatedFields, removedFields, truncatedArrays |
| clusterTime | Logical timestamp |
| ns | {db, coll} |
4. Resuming Change Stream
| Option | Detail |
|---|---|
| resumeAfter | From a known token |
| startAfter | Past invalidate events |
| Auto | Drivers auto-resume on transient errors |
5. Using Full Document Option
| Value | Effect |
|---|---|
| default | updateDescription only |
| updateLookup | Latest doc snapshot |
| whenAvailable / required | With pre/post images |
6. Using Full Document Before Change
| Aspect | Detail |
|---|---|
| Requires | collMod with changeStreamPreAndPostImages: { enabled: true } |
| Option | fullDocumentBeforeChange: "whenAvailable" | "required" |
7. Watching Specific Operations
| Filter | Example |
|---|---|
| Inserts | {operationType:"insert"} |
| By field | {"fullDocument.status":"PAID"} |
| By update path | {"updateDescription.updatedFields.price":{$exists:true}} |
8. Setting Start Time
| Option | Detail |
|---|---|
| startAtOperationTime | Begin at given clusterTime (within oplog window) |
9. Handling Change Stream Errors
| Scenario | Action |
|---|---|
| Transient | Driver auto-retries with last token |
| Invalidate | Stream closes; use startAfter to resume |
| Token expired | Restart with startAtOperationTime |
10. Using Change Streams with Aggregation
| Stage | Detail |
|---|---|
| $match, $project, $addFields, $replaceRoot, $redact | Supported |
| $lookup, $group, $out, $merge | Not supported in change pipelines |
11. Monitoring Database/Collection Changes
| Watch Target | Visibility |
|---|---|
| Collection | Single namespace |
| Database | All collections in DB |
| Deployment | All non-system DBs (mongos/replSet) |