Change Streams

This version of the documentation is archived and no longer supported. View the current documentation to learn how to upgrade your version of the Java Reactive Streams driver.

MongoDB 3.6 introduces the $changeStream aggregation pipeline operator.

Change streams provide a way to watch changes to documents in a collection. To improve the usability of this new stage, the MongoCollection type includes a new watch() method. The ChangeStreamPublisher instance sets up the change stream and automatically attempts to resume if it encounters a potentially recoverable error.

Prerequisites

You must set up the following components to run the code examples in this guide:

A test.restaurants collection populated with documents from the restaurants.json file in the documentation assets GitHub.
The following import statements:

import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.changestream.FullDocument;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import org.bson.Document;

Important

This guide uses the Subscriber implementations, which are described in the Quick Start Primer.

Connect to a MongoDB Deployment

First, connect to a MongoDB deployment, then declare and define MongoDatabase and MongoCollection instances.

The following code connects to a standalone MongoDB deployment running on localhost on port 27017. Then, it defines the database variable to refer to the test database and the collection variable to refer to the restaurants collection:

MongoClient mongoClient = MongoClients.create();
MongoDatabase database = mongoClient.getDatabase("test");
MongoCollection<Document> collection = database.getCollection("restaurants");

To learn more about connecting to MongoDB deployments, see the Connect to MongoDB tutorial.

Watch for Changes on a Collection

To create a change stream use one of the MongoCollection.watch() methods.

In the following example, the change stream prints out all changes it observes:

collection.watch().subscribe(new PrintDocumentSubscriber());

Watch for Changes on a Database

Applications can open a single change stream to watch all non-system collections of a database. To create such a change stream, use one of the MongoDatabase.watch() methods.

In the following example, the change stream prints out all the changes it observes on the given database:

database.watch().subscribe(new PrintDocumentSubscriber());

Watch for Changes on All Databases

Applications can open a single change stream to watch all non-system collections of all databases in a MongoDB deployment. To create such a change stream, use one of the MongoClient.watch() methods.

In the following example, the change stream prints out all the changes it observes on the deployment to which the MongoClient is connected:

client.watch().subscribe(new PrintDocumentSubscriber());

Filtering Content

You can pass a list of aggregation stages to the watch() method to modify the data returned by the $changeStream operator.

Note

Not all aggregation operators are supported. See Change Streams in the Server manual to learn more.

In the following example, the change stream prints out all changes it observes corresponding to insert, update, replace and delete operations.

First, the pipeline includes a $match stage to filter for documents where the operationType is either an insert, update, replace or delete. Then, it sets the fullDocument to FullDocument.UPDATE_LOOKUP, so that the document after the update is included in the results:

collection.watch(
    asList(
        Aggregates.match(
            Filters.in("operationType", asList("insert", "update", "replace", "delete"))
        )
    )
).fullDocument(FullDocument.UPDATE_LOOKUP).subscribe(new PrintDocumentSubscriber());

Back

Aggregation Framework

Text Search