Flujos de cambio

Requisitos previos

Debe configurar los siguientes componentes para ejecutar los ejemplos de código de esta guía:

• Una test.restaurants colección rellena con documentos del restaurants.json archivo en los activos de documentación de GitHub.

• Las siguientes declaraciones de importación:

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;

Conectarse a una implementación de MongoDB

Primero, conéctese a una implementación de MongoDB, luego declare y defina las instancias MongoDatabase y MongoCollection.

El siguiente código se conecta a una implementación independiente de MongoDB que se ejecuta en localhost, en el puerto 27017. Luego, define la variable database para referirse a la base de datos test y la variable collection para referirse a la colección restaurants:

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

Para obtener más información sobre cómo conectarse a implementaciones de MongoDB, consulte el tutorial Conectarse a MongoDB.

Esté atento a los cambios en una colección

Para crear un flujo de cambios, utilice uno de los métodos MongoCollection.watch().

En el siguiente ejemplo, el flujo de cambios imprime todos los cambios que observa:

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

Esté atento a los cambios en una base de datos

Las aplicaciones pueden abrir un único flujo de cambios para supervisar todas las colecciones de una base de datos que no pertenecen al sistema. Para crear dicho flujo de cambios, utilice uno de los métodos MongoDatabase.watch().

En el siguiente ejemplo, el flujo de cambios imprime todos los cambios que observa en la base de datos dada:

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

Esté atento a los cambios en todas las bases de datos

Las aplicaciones pueden abrir un único flujo de cambios para supervisar todas las colecciones no del sistema de todas las bases de datos en una implementación de MongoDB. Para crear ese flujo de cambios, utiliza uno de los métodos MongoClient.watch().

En el siguiente ejemplo, el flujo de cambios imprime todos los cambios que observa en la implementación a la que está conectado MongoClient:

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

Filtrado de contenido

Puede pasar una lista de etapas de agregación al método watch() para modificar los datos devueltos por el operador $changeStream.

En el siguiente ejemplo, el flujo de cambios imprime todos los cambios que observa correspondientes a las operaciones insert, update, replace y delete.

Primero, la canalización incluye una etapa $match para filtrar documentos donde operationType es insert, update, replace o delete. Luego, establece fullDocument en FullDocument.UPDATE_LOOKUP para que el documento posterior a la actualización se incluya en los resultados:

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