On this page

• Definition
• Behavior
  • Availability
  • Resumability
  • Full Document Lookup of Update Operations
• Access Control
• Examples
  • Open a Change Stream
  • Change Stream with Full Document Update Lookup
  • Change Stream with Aggregation Pipeline Filter
  • Resuming a Change Stream

Definition

db.collection. watch ( pipeline, options )

Opens a change stream cursor on the collection.

Parameter	Type	Description
pipeline	array	A sequence of one or more of the following aggregation stages: $match $project $addFields $replaceRoot $redact See Aggregation for complete documentation on the aggregation framework.
options	document	Optional. Additional options that modify the behavior of watch(). You must pass an empty array [] to the pipeline parameter if you are not specifying a pipeline but are passing the options document.

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.
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.

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.

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 is closed.
  • 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.
• You can only use db.collection.watch() with the Wired Tiger storage engine.
• db.collection.watch() is available for replica sets and sharded clusters:
  • For a replica set, you can issue db.watch() on any data-bearing member.
  • For a sharded cluster, you must issue db.watch() on a mongos instance.

Access Control

For deployments enforcing authentication and authorization, authenticate as a user with the changeStream and find privilege actions on the collection against which you want to open a change stream.

The read built-in role includes the required privileges to support opening a change stream against a collection. Any built-in role or user-defined role that inherits the read role can also support opening change streams against a collection.

Alternatively, use db.createRole to create a user-defined role that grants the changeStream and find privilege actions on the target collection. See User-Defined Roles for more complete documentation.

To associate a built-in role or user-defined role to an existing user, use the db.grantRolesToUser() or db.updateUser() methods. You can also specify the role when creating a new user using db.createUser().

Examples

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

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

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

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

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

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

let watchCursor = db.getSiblingDB("data").sensors.watch();
let firstChange;

while (!watchCursor.isExhausted