db.collection.watch()

Definition

db.collection.watch( pipeline, options )

Important

mongosh Method

This page documents a mongosh method. This is not the documentation for database commands or language-specific drivers, such as Node.js.

For the database command, see the aggregate command with the $changeStream aggregation stage.

For MongoDB API drivers, refer to the language-specific MongoDB driver documentation.

For replica sets and sharded clusters only

Opens a change stream cursor on the collection.

Parameter	Type	Description
pipeline	array	Optional. An Aggregation Pipeline consisting of one or more of the following aggregation stages: $addFields $match $project $replaceRoot $replaceWith $redact $set $unset Specify a pipeline to filter/modify the change events output. Starting in MongoDB 4.2, change streams will throw an exception if the change stream aggregation pipeline modifies an event's _id field.
options	document	Optional. Additional options that modify the behavior of watch().

The options document can contain the following fields and values:

Field	Type	Description
resumeAfter	document	Optional. Directs watch() to attempt resuming notifications starting after the operation specified in the resume token. Each change stream event document includes a resume token as the _id field. Pass the entire _id field of the change event document that represents the operation you want to resume after. resumeAfter is mutually exclusive with startAfter and startAtOperationTime.
startAfter	document	Optional. Directs watch() to attempt starting a new change stream after the operation specified in the resume token. Allows notifications to resume after an invalidate event. Each change stream event document includes a resume token as the _id field. Pass the entire _id field of the change event document that represents the operation you want to resume after. startAfter is mutually exclusive with resumeAfter and startAtOperationTime.
fullDocument	string	Optional. By default, watch() returns the delta of those fields modified by an update operation, instead of the entire updated document. Set fullDocument to "updateLookup" to direct watch() to look up the most current majority-committed version of the updated document. watch() returns a fullDocument field with the document lookup in addition to the updateDescription delta.
batchSize	int	Optional. Specifies the maximum number of change events to return in each batch of the response from the MongoDB cluster. Has the same functionality as cursor.batchSize().
maxAwaitTimeMS	int	Optional. The maximum amount of time in milliseconds the server waits for new data changes to report to the change stream cursor before returning an empty batch. Defaults to 1000 milliseconds.
collation	document	Optional. Pass a collation document to specify a collation for the change stream cursor. Defaults to simple binary comparison if omitted.
startAtOperationTime	Timestamp	Optional. The starting point for the change stream. If the specified starting point is in the past, it must be in the time range of the oplog. To check the time range of the oplog, see rs.printReplicationInfo(). startAtOperationTime is mutually exclusive with resumeAfter and startAfter.

Returns:	A cursor that remains open as long as a connection to the MongoDB deployment remains open and the collection exists. See Change Events for examples of change event documents.

Tip

db.watch() and Mongo.watch()

Compatibility

This method is available in deployments hosted in the following environments:

• MongoDB Atlas: The fully managed service for MongoDB deployments in the cloud

• MongoDB Enterprise: The subscription-based, self-managed version of MongoDB

• MongoDB Community: The source-available, free-to-use, and self-managed version of MongoDB

Availability

Behavior

• db.collection.watch() only notifies on data changes that have persisted to a majority of data-bearing members.

• The change stream cursor remains open until one of the following occurs:

  • The cursor is explicitly closed.

  • An invalidate event occurs; for example, a collection drop or rename.

  • The connection to the MongoDB deployment closes or times out. See Cursor Behaviors for more information.

  • If the deployment is a sharded cluster, a shard removal may cause an open change stream cursor to close, and the closed change stream cursor may not be fully resumable.

Access Control

When running with access control, the user must have the find and changeStream privilege actions on the collection resource. That is, a user must have a role that grants the following privilege:

{ resource: { db: <dbname>, collection: <collection> }, actions: [ "find", "changeStream" ] }

The built-in read role provides the appropriate privileges.

Cursor Iteration

MongoDB provides multiple ways to iterate on a cursor.

The cursor.hasNext() method blocks and waits for the next event. To monitor the watchCursor cursor and iterate over the events, use hasNext() like this:

while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}

The cursor.tryNext() method is non-blocking. To monitor the watchCursor cursor and iterate over the events, use tryNext() like this:

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

Examples

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

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

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

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

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

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

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 }
)

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