Change Streams

The Java examples below assume that you have connected to a MongoDB replica set and have accessed a database that 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 database that 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 database that 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 database that 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 database that contains an inventory collection.

mongoc_collection_t *collection;
bson_t pipeline = BSON_INITIALIZER;
bson_t opts = BSON_INITIALIZER;
mongoc_change_stream_t *stream;
const bson_t *change;
bson_iter_t iter;
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 database that contains an inventory collection.

var enumerator = inventory.Watch().ToEnumerable().GetEnumerator();
enumerator.MoveNext();
var next = enumerator.Current;
enumerator.Dispose();

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

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

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:

• $match

• $project

• $addFields

• $replaceRoot

• $redact

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 that filters any operations where the username is alice , or operations where the operationType is delete .

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

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:

• $match

• $project

• $addFields

• $replaceRoot

• $redact

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 by providing an array of one or more of the following pipeline stages when configuring the change stream:

• $match

• $project

• $addFields

• $replaceRoot

• $redact

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:

• $match

• $project

• $addFields

• $replaceRoot

• $redact

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:

• $match

• $project

• $addFields

• $replaceRoot

• $redact

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:

• $match

• $project

• $addFields

• $replaceRoot

• $redact

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:

• $match

• $project

• $addFields

• $replaceRoot

• $redact

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

In the example below, all update operations notifications include 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 updated document, pass { fullDocument: 'updateLookup' } to the collection.watch() method.

In the example below, all update operations notifications include 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 current majority-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 notifications include 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();

$nextChange = $changeStream->current();

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

In the example below, all update operations notifications include 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 updated document, pass the "fullDocument" option with the "updateLookup" value to the mongoc_collection_watch method.

In the example below, all update operations notifications include 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 updated document, pass "FullDocument = ChangeStreamFullDocumentOption.UpdateLookup" to the collection.Watch() method.

In the example below, all update operations notifications include a FullDocument field that represents the current version of the document affected by the update operation.

var options = new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup };
var enumerator = inventory.Watch(options).ToEnumerable().GetEnumerator();
enumerator.MoveNext();
var next = enumerator.Current;
enumerator.Dispose();

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

In the example below, all update operations notifications include 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

In the example below, the resumeToken contains the change stream notification id. The resumeAfter() method takes a parameter that must resolve to a resume token. Passing the resumeToken to the resumeAfter() method directs the change stream to attempt to resume notifications starting after the operation specified in the resume token.

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

In the example below, resumeToken contains the change stream notification id. The resumeAfter takes a parameter that must resolve to a resume token. Passing the resumeToken to the resumeAfter modifier directs the change stream to attempt to resume notifications starting after the operation specified.

const collection = db.collection('inventory');
const changeStream = collection.watch();

let resumeToken, newChangeStream;
changeStream.on('change', next => {
  resumeToken = next._id;
  changeStream.close();

  newChangeStream = collection.watch({ resumeAfter });
  newChangeStream.on('change', next => {
    // process next document
  });
});

In the example below, $resumeToken contains the change stream notification id. The resumeAfter option takes a value that must resolve to a resume token. Passing the $resumeToken to the resumeAfter option directs the change stream to attempt to resume notifications starting after the operation specified in the resume token.

$resumeToken = ($lastChange !== null) ? $lastChange->_id : null;

if ($resumeToken === null) {
    throw new \Exception('resumeToken was not found');
}

$changeStream = $db->inventory->watch([], ['resumeAfter' => $resumeToken]);
$changeStream->rewind();

$nextChange = $changeStream->current();

In the example below, resume_token contains the change stream notification id. The resume_after modifier takes a parameter that must resolve to a resume token. Passing the resume_token to the resume_after modifier directs the change stream to attempt to resume notifications starting after the operation specified in the resume token.

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 options to recreate the stream after it has been destroyed. Passing the _id to the change stream attempts to resume notifications starting after the operation specified.

stream = mongoc_collection_watch (collection, &pipeline, NULL);
if (mongoc_change_stream_next (stream, &change)) {
   bson_iter_init_find (&iter, change, "_id");
   BSON_APPEND_VALUE (&opts, "resumeAfter", bson_iter_value (&iter));

   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 document and passed to the Watch() method as an option. Passing the resumeToken to the Watch() method directs the change stream to attempt to resume notifications starting after the operation specified in the resume token.

var resumeToken = lastChangeStreamDocument.ResumeToken;
  var options = new ChangeStreamOptions { ResumeAfter = resumeToken };
  var enumerator = inventory.Watch(options).ToEnumerable().GetEnumerator();
  enumerator.MoveNext();
  var next = enumerator.Current;
  enumerator.Dispose();

In the example below, resume_token contains the change stream notification id. The resume_after modifier takes a parameter that must resolve to a resume token. Passing the resume_token to the resume_after modifier directs the change stream to attempt to resume notifications starting after the operation specified in the resume token.

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