Open Change Streams

Overview

In this guide, you can learn how to use a change stream to monitor real-time changes to your database. A change stream is a MongoDB server feature that allows your application to subscribe to data changes on a single collection, database, or deployment. You can specify a set of aggregation operators to filter and transform the data your application receives. When connecting to MongoDB v6.0 or later, you can configure the events to include the document data before and after the change.

Learn how to open and configure your change streams in the following sections:

• Open a Change Stream

• Apply Aggregation Operators to your Change Stream

• Split Large Change Stream Events

• Include Pre-images and Post-images

Open a Change Stream

You can open a change stream to subscribe to specific types of data changes and produce change events in your application.

To open a change stream, call the watch() method on an instance of a MongoCollection, MongoDatabase, or MongoClient.

The object on which you call the watch() method on determines the scope of events that the change stream listens for.

If you call watch() on a MongoCollection, the change stream monitors a collection.

If you call watch() on a MongoDatabase, the change stream monitors all collections in that database.

If you call watch() on a MongoClient, the change stream monitors all changes in the connected MongoDB deployment.

// Launch the change stream in a separate coroutine,
// so you can cancel it later.
val job = launch {
    val changeStream = collection.watch()
    changeStream.collect {
        println("Received a change event: $it")
    }
}
// Perform MongoDB operations that trigger change events...
// Cancel the change stream when you're done listening for events.
job.cancel()

Received a change event: ChangeStreamDocument{
   operationType='insert',
   resumeToken={"_data": "825EC..."},
   namespace=myDb.myChangeStreamCollection,
   ...
}

Apply Aggregation Operators to your Change Stream

You can pass an aggregation pipeline as a parameter to the watch() method to specify which change events the change stream receives.

To learn which aggregation operators your MongoDB server version supports, see Modify Change Stream Output.

val pipeline = listOf(
    Aggregates.match(Filters.`in`("operationType",
        listOf("insert", "update")))
)
// Launch the change stream in a separate coroutine,
// so you can cancel it later.
val job = launch {
    val changeStream = collection.watch(pipeline)
    changeStream.collect {
        println("Received a change event: $it")
    }
}
// Perform MongoDB operations that trigger change events...
// Cancel the change stream when you're done listening for events.
job.cancel()

Received a change event: ChangeStreamDocument{
operationType=update,
resumeToken={...},
...

Split Large Change Stream Events

When connecting to MongoDB v7.0 or later, you can use the $changeStreamSplitLargeEvent aggregation operator to split event documents that exceed 16 MB into smaller fragments.

Use the $changeStreamSplitLargeEvent operator only when you expect the change stream events to exceed the document size limit. For example, you might use this feature if your application requires full document pre-images or post-images.

A $changeStreamSplitLargeEvent aggregation stage returns fragments sequentially. You can access the fragments by using a change stream cursor. Each fragment document includes a splitEvent object that contains the following fields:

The following example opens a change stream that includes an aggregation pipeline with an $changeStreamSplitLargeEvent aggregation stage to split large events:

val pipeline = listOf(BsonDocument().append("\$changeStreamSplitLargeEvent", BsonDocument()))
val job = launch {
    val changeStream = collection.watch(pipeline)
    changeStream.collect {
        println("Received a change event: $it")
    }
}

To learn more about the $changeStreamSplitLargeEvent aggregation operator, see $changeStreamSplitLargeEvent (aggregation) in the Server manual.

Include Pre-images and Post-images

You can configure the change event to contain or omit the following data:

• The pre-image which is a document that represents the version of the document before the operation if it exists

• The post-image which is a document that represents the version of the document after the operation if it exists

To receive change stream events that include a pre-image or post-image, you must connect to a MongoDB v6.0 or later deployment and set up the following:

• Enable pre-images and post-images for the collection on your MongoDB deployment.

  Tip

  To learn how to enable these on your deployment, see the Change Streams with Document Pre- and Post-Images MongoDB server manual page.

  To learn how to instruct the driver to create a collection with pre-images and post-images enabled, see the Create a Collection with Pre-Image and Post-Images Enabled section.

• Configure your change stream to retrieve either or both the pre-images and post-images.

  Tip

  To configure your change stream to include the pre-image, see the Pre-image Configuration Example.

  To configure your change stream to include the post-image, see the Post-image Configuration Example.

val collectionOptions = CreateCollectionOptions()
collectionOptions.changeStreamPreAndPostImagesOptions(ChangeStreamPreAndPostImagesOptions(true))
database.createCollection("myChangeStreamCollection", collectionOptions)

val job = launch {
    val changeStream = collection.watch()
        .fullDocumentBeforeChange(FullDocumentBeforeChange.REQUIRED)
    changeStream.collect {
        println(it)
    }
}
// Perform MongoDB operations that trigger change events...
// Cancel the change stream when you're done listening for events.
job.cancel()

Received a change event: ChangeStreamDocument{
   operationType=update,
   resumeToken={...}
   namespace=software.libraries,
   destinationNamespace=null,
   fullDocument=null,
   fullDocumentBeforeChange=Document{{_id=6388..., latestVersion=2.0.0, ...}},
   ...

val job = launch {
    val changeStream = collection.watch()
        .fullDocument(FullDocument.UPDATE_LOOKUP)
    changeStream.collect {
        println(it)
    }
}
// Perform MongoDB operations that trigger change events...
// Cancel the change stream when you're done listening for events.
job.cancel()

Received a change event: ChangeStreamDocument{
   operationType=update,
   resumeToken={...},
   namespace=censusData.cities,
   destinationNamespace=null,
   fullDocument=Document{{_id=6388..., city=Springfield, population=950, ...}},
   updatedFields={"population": 950}, ...
   ...