Change Streams

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;

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.

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());