Change Streams

New in version 3.6.

Change streams allow applications to access real-time data changeswithout the complexity and risk of tailing the oplog.Applications can use change streams to subscribe to all data changes ona single collection, a database, or an entire deployment, andimmediately react to them. Because change streams use the aggregationframework, applications can also filter for specific changes ortransform the notifications at will.

Availability

Change streams are available for replica sets andsharded clusters:

• Storage Engine.

The replica sets and sharded clusters must use the WiredTiger storage engine. Change streams can also be usedon deployments that employ MongoDB’sencryption-at-rest feature.

• Replica Set Protocol Version.

The replica sets and sharded clusters must use replica set protocolversion 1 (pv1).

• Read Concern “majority” Enablement.

Starting in MongoDB 4.2, change streams areavailable regardless of the "majority" read concernsupport; that is, read concern majority support can be eitherenabled (default) or disabledto use change streams.

In MongoDB 4.0 and earlier, change streams areavailable only if "majority" read concern support isenabled (default).

Watch Collection/Database/Deployment

You can open change streams against:

Change Stream Examples

The examples on this page use the MongoDB drivers to illustrate howto open a change stream cursor for a collection and work with thechange stream cursor.

Open A Change Stream

To open a change stream:

• For a replica set, you can issue the open change stream operationfrom any of the data-bearing members.
• For a sharded cluster, you must issue the open change streamoperation from the mongos.

The following example opens a change stream for a collection anditerates over the cursor to retrieve the change stream documents.[1]

• Python
• Java (Sync)
• Node.js
• PHP
• Motor
• Other
  • C
  • C#
  • Ruby
  • Go

The Python examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

cursor = db.inventory.watch()
document = next(cursor)

The Java examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

MongoCursor<ChangeStreamDocument<Document>> cursor = inventory.watch().iterator();
ChangeStreamDocument<Document> next = cursor.next();

The Node.js examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

The following example uses stream to process the change events.

const collection = db.collection('inventory');
const changeStream = collection.watch();
changeStream.on('change', next => {
  // process next document
});

Alternatively, you can also use iterator to process the change events:

const changeStreamIterator = collection.watch();
const next = await changeStreamIterator.next();

The examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

$changeStream = $db->inventory->watch();
$changeStream->rewind();
 
$firstChange = $changeStream->current();
 
$changeStream->next();
 
$secondChange = $changeStream->current();

The examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

cursor = db.inventory.watch()
document = await cursor.next()

The C examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

mongoc_collection_t *collection;
bson_t *pipeline = bson_new ();
bson_t opts = BSON_INITIALIZER;
mongoc_change_stream_t *stream;
const bson_t *change;
const bson_t *resume_token;
bson_error_t error;
 
collection = mongoc_database_get_collection (db, "inventory");
stream = mongoc_collection_watch (collection, pipeline, NULL /* opts */);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}
 
mongoc_change_stream_destroy (stream);

The C# examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

var cursor = inventory.Watch();
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

The examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

cursor = inventory.watch.to_enum
next_change = cursor.next

The Go examples below assume that you have connected to a MongoDB replica set and have accessed a databasethat contains an inventory collection.

cs, err := coll.Watch(ctx, mongo.Pipeline{})
require.NoError(t, err)
defer cs.Close(ctx)
 
ok := cs.Next(ctx)
next := cs.Current

To retrieve the data change event fromthe cursor, iterate the change stream cursor. For information on thechange stream event, see Change Events.

While the connection to the MongoDB deployment remainsopen, the cursor remains open until one of the following occurs:

• The cursor is explicitly closed.
• An invalidate event occurs.
• If the deployment is a sharded cluster, a shard removal may cause anopen change stream cursor to close, and the closed change stream cursor maynot be fully resumable.

Note

The lifecycle of an unclosed cursor is language-dependent.

Modify Change Stream Output

• Python
• Java (Sync)
• Node.js
• PHP
• Motor
• Other
  • C
  • C#
  • Ruby
  • Go

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

pipeline = [
    {'$match': {'fullDocument.username': 'alice'}},
    {'$addFields': {'newField': 'this is an added field!'}}
]
cursor = db.inventory.watch(pipeline=pipeline)
document = next(cursor)

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

MongoClient mongoClient = new MongoClient( new MongoClientURI("mongodb://host1:port1,host2:port2..."));
 
// Select the MongoDB database and collection to open the change stream against
 
MongoDatabase db = mongoClient.getDatabase("myTargetDatabase");
 
MongoCollection<Document> collection = db.getCollection("myTargetCollection");
 
// Create $match pipeline stage.
List<Bson> pipeline = singletonList(Aggregates.match(Filters.or(
    Document.parse("{'fullDocument.username': 'alice'}"),
    Filters.in("operationType", asList("delete")))));
 
// Create the change stream cursor, passing the pipeline to the
// collection.watch() method
 
MongoCursor<Document> cursor = collection.watch(pipeline).iterator();

The pipeline list includes a single $match stage thatfilters any operations where the username is alice, oroperations where the operationType is delete.

Passing the pipeline to the watch() method directs thechange stream to return notifications after passing them through thespecified pipeline.

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

The following example uses stream to process the change events.

const pipeline = [
  { $match: { 'fullDocument.username': 'alice' } },
  { $addFields: { newField: 'this is an added field!' } }
];
 
const collection = db.collection('inventory');
const changeStream = collection.watch(pipeline);
changeStream.on('change', next => {
  // process next document
});

Alternatively, you can also use iterator to process the change events:

const changeStreamIterator = collection.watch(pipeline);
const next = await changeStreamIterator.next();

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

pipeline = [
    {'$match': {'fullDocument.username': 'alice'}},
    {'$addFields': {'newField': 'this is an added field!'}}
]
cursor = db.inventory.watch(pipeline=pipeline)
document = await cursor.next()

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change =>
        change.FullDocument["username"] == "alice" ||
        change.OperationType == ChangeStreamOperationType.Delete)
    .AppendStage<ChangeStreamDocument<BsonDocument>, ChangeStreamDocument<BsonDocument>, BsonDocument>(
        "{ $addFields : { newField : 'this is an added field!' } }");
 
var collection = database.GetCollection<BsonDocument>("inventory");
using (var cursor = collection.Watch(pipeline))
{
    while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
    var next = cursor.Current.First();
}

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

You can control change stream output byproviding an array of one or more of the following pipeline stages whenconfiguring the change stream:

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith (Available starting in MongoDB 4.2)
• $redact
• $set (Available starting in MongoDB 4.2)
• $unset (Available starting in MongoDB 4.2)

pipeline := mongo.Pipeline{bson.D{{"$match", bson.D{{"$or",
    bson.A{
        bson.D{{"fullDocument.username", "alice"}},
        bson.D{{"operationType", "delete"}}}}},
}}}
cs, err := coll.Watch(ctx, pipeline)
require.NoError(t, err)
defer cs.Close(ctx)
 
ok := cs.Next(ctx)
next := cs.Current

Tip

The _id field of the change streamevent document act as the resume token. Do not use the pipeline to modify or removethe change stream event’s _id field.

Starting in MongoDB 4.2, change streams will throw an exception ifthe change stream aggregation pipeline modifies an event’s _id field.

See Change Events for more information on the change streamresponse document format.

Lookup Full Document for Update Operations

By default, change streams only return the delta of fields during theupdate operation. However, you can configure the change stream toreturn the most current majority-committed version of the updateddocument.

• Python
• Java (Sync)
• Node.js
• PHP
• Motor
• Other
  • C
  • C#
  • Ruby
  • Go

To return the most current majority-committed version of the updateddocument, pass full_document='updateLookup' to thedb.collection.watch() method.

In the example below, all update operations notificationsinclude a full_document field that represents the _current_version of the document affected by the update operation.

cursor = db.inventory.watch(full_document='updateLookup')
document = next(cursor)

To return the most current majority-committed version of the updateddocument, pass FullDocument.UPDATE_LOOKUP to thedb.collection.watch.fullDocument() method.

In the example below, all update operations notificationsinclude a FullDocument field that represents the _current_version of the document affected by the update operation.

cursor = inventory.watch().fullDocument(FullDocument.UPDATE_LOOKUP).iterator();
next = cursor.next();

To return the most current majority-committed version of the updateddocument, pass { fullDocument: 'updateLookup' } to thecollection.watch() method.

In the example below, all update operations notificationsinclude a fullDocument field that represents the _current_version of the document affected by the update operation.

The following example uses stream to process the change events.

const collection = db.collection('inventory');
const changeStream = collection.watch({ fullDocument: 'updateLookup' });
changeStream.on('change', next => {
  // process next document
});

Alternatively, you can also use iterator to process the change events:

const changeStreamIterator = collection.watch({ fullDocument: 'updateLookup' });
const next = await changeStreamIterator.next();

To return the most currentmajority-committed version of the updated document, pass"fullDocument' => \MongoDB\Operation\ChangeStreamCommand::FULL_DOCUMENT_UPDATE_LOOKUP"to the watch() method.

In the example below, all update operations notificationsinclude a fullDocument field that represents the _current_version of the document affected by the update operation.

$changeStream = $db->inventory->watch([], ['fullDocument' => \MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP]);
$changeStream->rewind();
 
$firstChange = $changeStream->current();
 
$changeStream->next();
 
$secondChange = $changeStream->current();

To return the most current majority-committed version of the updateddocument, pass full_document='updateLookup' to thedb.collection.watch() method.

In the example below, all update operations notificationsinclude a `full_document field that represents the _current_version of the document affected by the update operation.

cursor = db.inventory.watch(full_document='updateLookup')
document = await cursor.next()

To return the most current majority-committed version of the updateddocument, pass the "fullDocument" option with the "updateLookup" value to themongoc_collection_watch method.

In the example below, all update operations notificationsinclude a fullDocument field that represents the _current_version of the document affected by the update operation.

BSON_APPEND_UTF8 (&opts, "fullDocument", "updateLookup");
stream = mongoc_collection_watch (collection, pipeline, &opts);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}
 
mongoc_change_stream_destroy (stream);

To return the most current majority-committed version of the updateddocument, pass "FullDocument = ChangeStreamFullDocumentOption.UpdateLookup" to thecollection.Watch() method.

In the example below, all update operations notificationsinclude a FullDocument field that represents the _current_version of the document affected by the update operation.

var options = new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup };
var cursor = inventory.Watch(options);
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

To return the most current majority-committed version of the updateddocument, pass full_document: 'updateLookup' to thewatch() method.

In the example below, all update operations notificationsinclude a full_document field that represents the _current_version of the document affected by the update operation.

cursor = inventory.watch([], full_document: 'updateLookup').to_enum
next_change = cursor.next

To return the most current majority-committed version of theupdated document, SetFullDocument(options.UpdateLookup)change stream option.

cs, err := coll.Watch(ctx, mongo.Pipeline{}, options.ChangeStream().SetFullDocument(options.UpdateLookup))
require.NoError(t, err)
defer cs.Close(ctx)
 
ok := cs.Next(ctx)
next := cs.Current

Note

If there are one or more majority-committed operations that modifiedthe updated document after the update operation but before thelookup, the full document returned may differ significantly from thedocument at the time of the update operation.

However, the deltas included in the change stream document alwayscorrectly describe the watched collection changes that applied tothat change stream event.

See Change Events for more information on the changestream response document format.

Resume a Change Stream

Change streams are resumable by specifying a resume token to eitherresumeAfter orstartAfter when opening the cursor.

resumeAfter for Change Streams

You can resume a change stream after a specific event by passing a resume tokento resumeAfter when opening the cursor. For the resume token, use the_id value of the change stream event document.See Resume Tokens for more information on the resume token.

Important

• The oplog must have enough history to locate the operationassociated with the token or the timestamp, if the timestamp is inthe past.
• You cannot use resumeAfter to resume a change stream after aninvalidate event (for example, a collectiondrop or rename) closes the stream. Starting in MongoDB 4.2, you can usestartAfter to start a new changestream after an invalidate event.

• Python
• Java (Sync)
• Node.js
• PHP
• Motor
• Other
  • C
  • C#
  • Ruby
  • Go

You can use the resume_after modifier to resumenotifications after the operation specified in the resumetoken. The resume_after modifier takes a value that mustresolve to a resume token, e.g. resume_token in theexample below.

resume_token = cursor.resume_token
cursor = db.inventory.watch(resume_after=resume_token)
document = next(cursor)

You can use the resumeAfter() method to resumenotifications after the operation specified in the resumetoken. The resumeAfter() method takes a value that mustresolve to a resume token, e.g. resumeToken in theexample below.

BsonDocument resumeToken = next.getResumeToken();
cursor = inventory.watch().resumeAfter(resumeToken).iterator();
next = cursor.next();

You can use the resumeAfter option to resumenotifications after the operation specified in the resumetoken. The resumeAfter option takes a value that mustresolve to a resume token, e.g. resumeToken in theexample below.

const collection = db.collection('inventory');
const changeStream = collection.watch();
 
let newChangeStream;
changeStream.on('change', next => {
  const resumeToken = changeStream.resumeToken;
  changeStream.close();
 
  newChangeStream = collection.watch({ resumeAfter: resumeToken });
  newChangeStream.on('change', next => {
    // process next document
  });
});

You can use the resumeAfter option to resumenotifications after the operation specified in the resumetoken. The resumeAfter option takes a value that mustresolve to a resume token, e.g. $resumeToken in theexample below.

$resumeToken = $changeStream->getResumeToken();
 
if ($resumeToken === null) {
    throw new \Exception('Resume token was not found');
}
 
$changeStream = $db->inventory->watch([], ['resumeAfter' => $resumeToken]);
$changeStream->rewind();
 
$firstChange = $changeStream->current();

You can use the resume_after modifier to resumenotifications after the operation specified in the resumetoken. The resume_after modifier takes a value that mustresolve to a resume token, e.g. resume_token in theexample below.

resume_token = cursor.resume_token
cursor = db.inventory.watch(resume_after=resume_token)
document = await cursor.next()

In the example below, the resumeAfter option is appended to the stream optionsto recreate the stream after it has been destroyed. Passing the _id tothe change stream attempts to resume notifications starting after theoperation specified.

stream = mongoc_collection_watch (collection, pipeline, NULL);
if (mongoc_change_stream_next (stream, &change)) {
   resume_token = mongoc_change_stream_get_resume_token (stream);
   BSON_APPEND_DOCUMENT (&opts, "resumeAfter", resume_token);
 
   mongoc_change_stream_destroy (stream);
   stream = mongoc_collection_watch (collection, pipeline, &opts);
   mongoc_change_stream_next (stream, &change);
   mongoc_change_stream_destroy (stream);
} else {
   if (mongoc_change_stream_error_document (stream, &error, NULL)) {
      MONGOC_ERROR ("%s\n", error.message);
   }
 
   mongoc_change_stream_destroy (stream);
}

In the example below, the resumeToken is retrieved from the last change stream documentand passed to the Watch() method as an option. Passing the resumeTokento the Watch() method directsthe change stream to attempt to resume notifications starting after theoperation specified in the resume token.

  var resumeToken = previousCursor.GetResumeToken();
  var options = new ChangeStreamOptions { ResumeAfter = resumeToken };
  var cursor = inventory.Watch(options);
  cursor.MoveNext();
  var next = cursor.Current.First();
  cursor.Dispose();

You can use the resume_after modifier to resumenotifications after the operation specified in the resumetoken. The resume_after modifier takes a value that mustresolve to a resume token, e.g. resume_token in theexample below.

  change_stream = inventory.watch
  cursor = change_stream.to_enum
  next_change = cursor.next
  resume_token = change_stream.resume_token
 
  new_cursor = inventory.watch([], resume_after: resume_token).to_enum
  resumed_change = new_cursor.next

You can use ChangeStreamOptions.SetResumeAfterto specify the resumetoken for the change stream. If the resumeAfter option is set,the change stream resumes notifications after the operationspecified in the resume token. The SetResumeAfter takes avalue that must resolve to a resume token, e.g.resumeToken in the example below.

resumeToken := original.ResumeToken()
 
cs, err := coll.Watch(ctx, mongo.Pipeline{}, options.ChangeStream().SetResumeAfter(resumeToken))
require.NoError(t, err)
defer cs.Close(ctx)
 
ok = cs.Next(ctx)
result := cs.Current

startAfter for Change Streams

New in version 4.2.

You can start a new change stream after a specific event by passing a resumetoken to startAfter when opening the cursor. UnlikeresumeAfter, startAfter canresume notifications after an invalidate eventby creating a new change stream. For the resume token, use the _id value ofthe change stream event document. SeeResume Tokens for more information on the resume token.

Important

• The oplog must have enough history to locate the operationassociated with the token or the timestamp, if the timestamp is inthe past.

Resume Tokens

The _id value of a change stream event document acts as the resume token:

{
   "_data" : <BinData|hex string>
}

The resume token _data type depends on the MongoDB versions and,in some cases, the feature compatibility version (fcv) at the timeof the change stream’s opening/resumption (i.e. a change in fcvvalue does not affect the resume tokens for already opened changestreams):

With hex-encoded string resume tokens, you can compare and sort theresume tokens.

Regardless of the fcv value, a 4.0 deployment can useeither BinData resume tokens or hex string resume tokens to resume achange stream. As such, a 4.0 deployment can use a resumetoken from a change stream opened on a collection from a 3.6deployment.

New resume token formats introduced in a MongoDB versioncannot be consumed by earlier MongoDB versions.

Tip

Starting in MongoDB 4.2, change streams will throw an exception ifthe change stream aggregation pipeline modifies an event’s _id field.

Use Cases

Change streams can benefit architectures with reliant business systems,informing downstream systems once data changes are durable. For example,change streams can save time for developers when implementing Extract,Transform, and Load (ETL) services, cross-platform synchronization,collaboration functionality, and notification services.

Access Control

For deployments enforcing Authentication and authorization:

• To open a change stream against specific collection, applicationsmust have privileges that grant changeStream andfind actions on the corresponding collection.

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

• To open a change stream on a single databases, applications must haveprivileges that grant changeStream andfind actions on all non-system collections in adatabase.

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

• To open a change stream on an entire deployment, applications musthave privileges that grant changeStream andfind actions on all non-system collections for alldatabases in the deployment.

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

Event Notification

Change streams only notify on data changes that have persisted to a majorityof data-bearing members in the replica set. This ensures that notifications aretriggered only by majority-committed changes that are durable in failurescenarios.

For example, consider a 3-member replica set with a change streamcursor opened against the primary. If a client issues an insertoperation, the change stream only notifies the application of the data changeonce that insert has persisted to a majority of data-bearing members.

If an operation is associated with a transaction, the change event document includes thetxnNumber and the lsid.

Collation

Starting in MongoDB 4.2, change streams use simple binarycomparisons unless an explicit collation is provided. In earlierversions, change streams opened on a single collection(db.collection.watch()) would inherit that collection’sdefault collation.