Supervisar los cambios en los datos

Overview

En esta guía, aprenderá a usar un flujo de cambios para supervisar los cambios en sus datos en tiempo real. Un flujo de cambios es una función de MongoDB Server que permite que su aplicación se suscriba a los 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, llame al método watch(). La instancia en la que llama al método determina el alcance de los eventos que el flujo de cambios escucha. Puede 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 monitorear 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 una canalización de agregación a un flujo de cambios para registrar solo 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 su flujo de cambios, consulte la sección Modificar la salida del flujo de cambios en el manual de MongoDB Server.

Modificar watch() el comportamiento

Puede encadenar métodos al método watch() que representan opciones que puede usar para configurar la operación del flujo de cambios. Si no especifica ninguna opción, el controlador 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 obtener más información sobre los flujos de cambio, consulte Flujos de cambio en el manual de MongoDB Server.