db.collection.watch()

Deployment

db.collection.watch() is available for replica set and sharded cluster deployments :

• For a replica set, you can issue db.collection.watch() on any data-bearing member.

• For a sharded cluster, you must issue db.collection.watch() on a mongos instance.

Storage Engine

You can only use db.collection.watch() with the Wired Tiger storage engine.

Read Concern majority Support

Change streams are available regardless of the "majority" read concern support; that is, read concern majority support can be either enabled (default) or disabled to use change streams.

Resumability

Unlike the MongoDB Drivers, mongosh does not automatically attempt to resume a change stream cursor after an error. The MongoDB drivers make one attempt to automatically resume a change stream cursor after certain errors.

db.collection.watch() uses information stored in the oplog to produce the change event description and generate a resume token associated to that operation. If the operation identified by the resume token passed to the resumeAfter or startAfter option has already dropped off the oplog, db.collection.watch() cannot resume the change stream.

See Resume a Change Stream for more information on resuming a change stream.

Full Document Lookup of Update Operations

By default, the change stream cursor returns specific field changes/deltas for update operations. You can also configure the change stream to look up and return the current majority-committed version of the changed document. Depending on other write operations that may have occurred between the update and the lookup, the returned document may differ significantly from the document at the time of the update.

Depending on the number of changes applied during the update operation and the size of the full document, there is a risk that the size of the change event document for an update operation is greater than the 16MB BSON document limit. If this occurs, the server closes the change stream cursor and returns an error.

Open a Change Stream

The following operation opens a change stream cursor against the data.sensors collection:

watchCursor = db.getSiblingDB("data").sensors.watch()

Iterate the cursor to check for new events. Use the cursor.isClosed() method with the cursor.tryNext() method to ensure the loop only exits if the change stream cursor is closed and there are no objects remaining in the latest batch:

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

For complete documentation on change stream output, see Change Events.

Change Stream with Full Document Update Lookup

Set the fullDocument option to "updateLookup" to direct the change stream cursor to lookup the most current majority-committed version of the document associated to an update change stream event.

The following operation opens a change stream cursor against the data.sensors collection using the fullDocument : "updateLookup" option.

watchCursor = db.getSiblingDB("data").sensors.watch(
   [],
   { fullDocument : "updateLookup" }
)

Iterate the cursor to check for new events. Use the cursor.isClosed() method with the cursor.tryNext() method to ensure the loop only exits if the change stream cursor is closed and there are no objects remaining in the latest batch:

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

For any update operation, the change event returns the result of the document lookup in the fullDocument field.

For an example of the full document update output, see change stream update event.

For complete documentation on change stream output, see Change Events.

Change Streams with Document Pre- and Post-Images

Starting in MongoDB 6.0, you can use change stream events to output the version of a document before and after changes (the document pre- and post-images):

• The pre-image is the document before it was replaced, updated, or deleted. There is no pre-image for an inserted document.

• The post-image is the document after it was inserted, replaced, or updated. There is no post-image for a deleted document.

• Enable changeStreamPreAndPostImages for a collection using db.createCollection(), create, or collMod.

Pre- and post-images are not available for a change stream event if the images were:

• Not enabled on the collection at the time of a document update or delete operation.

• Removed after the pre- and post-image retention time set in expireAfterSeconds.

  • The following example sets expireAfterSeconds to 100 seconds on an entire cluster:

    use admin
    db.runCommand( {
       setClusterParameter:
          { changeStreamOptions: {
             preAndPostImages: { expireAfterSeconds: 100 }
          } }
    } )

  • The following example returns the current changeStreamOptions settings, including expireAfterSeconds:

    db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

  • Setting expireAfterSeconds to off uses the default retention policy: pre- and post-images are retained until the corresponding change stream events are removed from the oplog.

  • If a change stream event is removed from the oplog, then the corresponding pre- and post-images are also deleted regardless of the expireAfterSeconds pre- and post-image retention time.

Additional considerations:

• Enabling pre- and post-images consumes storage space and adds processing time. Only enable pre- and post-images if you need them.

• Limit the change stream event size to less than 16 megabytes. To limit the event size, you can:

  • Limit the document size to 8 megabytes. You can request pre- and post-images simultaneously in the change stream output if other change stream event fields like updateDescription are not large.

  • Request only post-images in the change stream output for documents up to 16 megabytes if other change stream event fields like updateDescription are not large.

  • Request only pre-images in the change stream output for documents up to 16 megabytes if:

    • document updates affect only a small fraction of the document structure or content, and

    • do not cause a replace change event. A replace event always includes the post-image.

• To request a pre-image, you set fullDocumentBeforeChange to required or whenAvailable in db.collection.watch(). To request a post-image, you set fullDocument using the same method.

• Pre-images are written to the config.system.preimages collection.

  • The config.system.preimages collection may become large. To limit the collection size, you can set expireAfterSeconds time for the pre-images as shown earlier.

  • Pre-images are removed asynchronously by a background process.

db.createCollection(
   "temperatureSensor",
   { changeStreamPreAndPostImages: { enabled: true } }
)

db.temperatureSensor.insertMany( [
   { "_id" : 0, "reading" : 26.1 },
   { "_id" : 1, "reading" : 25.9 },
   { "_id" : 2, "reading" : 24.3 },
   { "_id" : 3, "reading" : 22.4 },
   { "_id" : 4, "reading" : 24.6 }
] )

watchCursorFullDocumentBeforeChange = db.temperatureSensor.watch(
   [],
   { fullDocumentBeforeChange: "whenAvailable" }
)

while ( !watchCursorFullDocumentBeforeChange.isClosed() ) {
   if ( watchCursorFullDocumentBeforeChange.hasNext() ) {
      printjson( watchCursorFullDocumentBeforeChange.next() );
   }
}

db.temperatureSensor.updateOne(
   { _id: 2 },
   { $set: { reading: 22.1 } }
)

{
   "_id" : {
      "_data" : "82624B21...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1649090957, 1),
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 2
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 22.1
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   },
   "fullDocumentBeforeChange" : {
      "_id" : 2,
      "reading" : 24.3
   }
}

watchCursorFullDocument = db.temperatureSensor.watch(
   [],
   { fullDocument: "whenAvailable" }
)

while ( !watchCursorFullDocument.isClosed() ) {
   if ( watchCursorFullDocument.hasNext() ) {
      printjson( watchCursorFullDocument.next() );
   }
}

db.temperatureSensor.updateOne(
   { _id: 1 },
   { $set: { reading: 29.5 } }
)

{
   "_id" : {
      "_data" : "8262474D...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1648840090, 1),
   "fullDocument" : {
      "_id" : 1,
      "reading" : 29.5
   },
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 1
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 29.5
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   }
}

Change Stream with Aggregation Pipeline Filter

The following operation opens a change stream cursor against the data.sensors collection using an aggregation pipeline to filter only insert events:

watchCursor = db.getSiblingDB("data").sensors.watch(
   [
      { $match : {"operationType" : "insert" } }
   ]
)

Iterate the cursor to check for new events. Use the cursor.isClosed() method with the cursor.hasNext() method to ensure the loop only exits if the change stream cursor is closed and there are no objects remaining in the latest batch:

while (!watchCursor.isClosed()){
   if (watchCursor.hasNext()){
      printjson(watchCursor.next());
   }
}

The change stream cursor only returns change events where the operationType is insert. For complete documentation on change stream output, see Change Events.

Resuming a Change Stream

Every document returned by a change stream cursor includes a resume token as the _id field. To resume a change stream, pass the entire _id document of the change event you want to resume from to either the resumeAfter or startAfter option of watch().

The following operation resumes a change stream cursor against the data.sensors collection using a resume token. This assumes that the operation that generated the resume token has not rolled off the cluster's oplog.

let watchCursor = db.getSiblingDB("data").sensors.watch();
let firstChange;
while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}
watchCursor.close();
let resumeToken = firstChange._id;
resumedWatchCursor = db.getSiblingDB("data").sensors.watch(
[],
   { resumeAfter : resumeToken }
)

Iterate the cursor to check for new events. Use the cursor.isClosed() method with the cursor.hasNext() method to ensure the loop only exits if the change stream cursor is closed and there are no objects remaining in the latest batch:

while (!resumedWatchCursor.isClosed()){
   if (resumedWatchCursor.hasNext()){
      print(resumedWatchCursor.next());
   }
}

See Resume a Change Stream for complete documentation on resuming a change stream.