Change Streams (original) (raw)

Change streams allow applications to access real-time data changes without the prior complexity and risk of manually tailing the oplog. Applications can use change streams to subscribe to all data changes on a single collection, a database, or an entire deployment, and immediately react to them. Because change streams use the aggregation framework, applications can also filter for specific changes or transform the notifications at will.

Starting in MongoDB 5.1, change streams are optimized, providing more efficient resource utilization and faster execution of some aggregation pipeline stages.

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 used on deployments that employ MongoDB'sencryption-at-rest feature.
Replica Set Protocol Version.
The replica sets and sharded clusters must use replica set protocol version 1 (pv1).
Read Concern "majority" Enablement.
Change streams are available regardless of the "majority" read concern support; that is, read concern majority support can be either enabled (default) or disabledto use change streams.

Change streams are included in Stable API V1. However, the showExpandedEventsoption is not included in Stable API V1.

Connections for a change stream can either use DNS seed lists with the +srv connection option or by listing the servers individually in the connection string.

If the driver loses the connection to a change stream or the connection goes down, it attempts to reestablish a connection to the change stream through another node in the cluster that has a matchingread preference. If the driver cannot find a node with the correct read preference, it throws an exception.

For more information, see Connection String URI Format.

You can open change streams against:

Target	Description
A collection	You can open a change stream cursor for a single collection (except system collections, or any collections in theadmin, local, and config databases).The examples on this page use the MongoDB drivers to open and work with a change stream cursor for a single collection. See also the mongosh methoddb.collection.watch().
A database	You can open a change stream cursor for a single database (excludingadmin, local, and config database) to watch for changes to all its non-system collections.For the MongoDB driver method, refer to your driver documentation. See also the mongosh methoddb.watch().
A deployment	You can open a change stream cursor for a deployment (either a replica set or a sharded cluster) to watch for changes to all non-system collections across all databases except for admin, local, andconfig.For the MongoDB driver method, refer to your driver documentation. See also the mongosh methodMongo.watch().

Note

Change Stream Examples

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

If the amount of active change streams opened against a database exceeds theconnection pool size, you may experience notification latency. Each change stream uses a connection and a getMoreoperation on the change stream for the period of time that it waits for the next event. To avoid any latency issues, you should ensure that the pool size is greater than the number of opened change streams. For details see the maxPoolSize setting.

When a change stream is opened on a sharded cluster:

The mongos creates individual change streams on each shard. This behavior occurs regardless of whether the change stream targets a particular shard key range.
When the mongos receives change stream results, it sorts and filters those results. If needed, the mongos also performs afullDocument lookup.

For best performance, limit the use of $lookup queries in change streams.

To open a change stream:

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

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

➤ Use the Select your language drop-down menu in the upper-right to set the language of the examples on this page.

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 Kotlin examples below assume that you are connected to a MongoDB replica set and can access a database that contains the inventory collection. To learn more about completing these tasks, see theKotlin Coroutine Driver Databases and Collections guide.

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

  })

  .once('error', () => {

    // handle error

  });

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


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

const changeStream = collection.watch();

const next = await changeStream.next();

ChangeStream extends EventEmitter.

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

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 Behavior for more information.
If the deployment is a sharded cluster, a shard removal may cause an open change stream cursor to close. The closed change stream cursor may not be fully resumable.

Note

The lifecycle of an unclosed cursor is language-dependent.

➤ Use the Select your language drop-down menu in the upper-right to set the language of the examples on this page.