Supervisar los cambios en los datos

Overview

En esta guía, puedes aprender a usar un flujo de cambios para supervisar los cambios en tiempo real de tus datos. Un flujo de cambios es una funcionalidad de MongoDB Server que permite que la aplicación se suscriba a cambios en los datos de una colección, base de datos o implementación.

Abre un flujo de cambios

Para abrir un flujo de cambios, llama al método watch(). La instancia sobre la que llama el método determina el alcance de los eventos a los que el flujo de cambios escucha. Puedes llamar al método watch() en instancias de las siguientes clases:

• MongoClient:Para monitorear todos los cambios en la implementación de MongoDB

• MongoDatabase: Para supervisar los cambios en todas las colecciones de la base de datos

• MongoCollection:Para monitorear cambios en la colección

El siguiente ejemplo abre un flujo de cambios en la colección restaurants y muestra los cambios a medida que ocurren:

// Opens a change stream and prints the changes as they're received
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch();
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

Para empezar a observar los cambios, ejecute la aplicación. A continuación, en una aplicación o shell independiente, realice una operación de escritura en la colección restaurants. Actualizar un documento cuyo valor de campo "name" es "Blarney Castle" genera la siguiente salida del flujo de cambios:

Received change: ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null, fullDocument=null,
fullDocumentBeforeChange=null, documentKey={"_id": {"$oid": "..."}}, clusterTime=Timestamp{...},
updateDescription=UpdateDescription{removedFields=[], updatedFields={"cuisine": "Traditional Irish"},
truncatedArrays=[], disambiguatedPaths=null}, txnNumber=null, lsid=null,
splitEvent=null, wallTime=BsonDateTime{value=...}}

Modificar la salida del flujo de cambios

Puede pasar una canalización de agregación como parámetro al método watch() para modificar la salida del flujo de cambios. Este parámetro le permite observar únicamente los eventos de cambio especificados.

Puede especificar las siguientes etapas de agregación en el parámetro pipeline:

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

El siguiente ejemplo pasa un pipeline de agregación a un flujo de cambios para registrar solo las operaciones de actualización:

// Creates a change stream pipeline
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.eq("operationType", "update"))
);
// Opens a change stream and prints the changes as they're received
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch(pipeline);
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

Para obtener más información sobre cómo modificar la salida de tu flujo de cambios, consulta la sección Modificar salida del flujo de cambio en el manual del MongoDB Server.

Modificar el comportamiento de watch()

Puedes encadenar métodos al método watch() que representan opciones que puedes usar para configurar la operación del flujo de cambios. Si no especificas ninguna opción, el driver no personaliza la operación.

La siguiente tabla describe los métodos que puedes encadenar a watch() para personalizar su comportamiento:

// Creates a change stream pipeline
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.eq("operationType", "update"))
);
// Opens a change stream and prints the changes as they're received including the full
// document after the update
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch(pipeline)
        .fullDocument(FullDocument.UPDATE_LOOKUP);
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

Información Adicional

Para saber más sobre las change streams, consulta Change Streams en el manual de MongoDB Server.