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.

Al utilizar el controlador Scala, puede llamar al watch() Método para devolver una instancia de ChangeStreamObservable. Luego, puede suscribirse a la instancia ChangeStreamObservable para ver nuevos cambios en los datos, como actualizaciones, inserciones y eliminaciones.

val database: MongoDatabase = client.getDatabase("sample_restaurants")
val collection: MongoCollection[Document] = database.getCollection("restaurants")

case class LatchedObserver() extends Observer[ChangeStreamDocument[Document]] {
  val latch = new CountDownLatch(1)
  override def onSubscribe(subscription: Subscription): Unit = subscription.request(Long.MaxValue) // Request data
  override def onNext(changeDocument: ChangeStreamDocument[Document]): Unit = println(changeDocument)
  override def onError(throwable: Throwable): Unit = {
      println(s"Error: '$throwable")
      latch.countDown()
  }
  override def onComplete(): Unit = latch.countDown()
  def await(): Unit = latch.await()
}

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 watch() determina el alcance de los eventos que supervisa el flujo de cambios. Puede llamar al método watch() en instancias de las siguientes clases:

• MongoClient:Supervisa los cambios en todas las colecciones en todas las bases de datos de una implementación, excluyendo las colecciones del sistema o las colecciones en las admin local config bases de datos, y

• MongoDatabase:Monitorea los cambios en todas las colecciones en una base de datos

• MongoCollection:Monitorea los cambios en una colección

El siguiente ejemplo llama al método watch() para abrir un flujo de cambios en la colección restaurants. El código crea una instancia LatchedObserver para recibir y generar los cambios a medida que ocurren:

val observer = LatchedObserver()
collection.watch().subscribe(observer)
observer.await()

Para empezar a detectar cambios, ejecute el código anterior. Luego, en un shell independiente, ejecute el siguiente código para actualizar un documento cuyo valor de campo name sea "Blarney Castle":

val filter = equal("name", "Blarney Castle")
val update = set("cuisine", "American")
collection.updateOne(filter, update)
          .subscribe((res: UpdateResult) => println(res),
                    (e: Throwable) => println(s"There was an error: $e"))

Cuando ejecuta el código anterior para actualizar la colección, la aplicación de flujo de cambios imprime el cambio a medida que ocurre. El evento de cambio impreso se asemeja a la siguiente salida:

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

Modificar la salida del flujo de cambios

Para modificar la salida del flujo de cambios, puede pasar una lista de etapas de la canalización como parámetro al método watch(). Puede incluir las siguientes etapas en la lista:

• $addFields o $set: Agrega nuevos campos a los documentos

• $match:Filtra los documentos

• $project: Proyecta un subconjunto de los campos del documento

• $replaceWith o $replaceRoot: reemplaza el documento de entrada con el documento especificado

• $redact:Restringe el contenido de los documentos

• $unset:Elimina campos de los documentos

El controlador Scala proporciona la clase Aggregates, que incluye métodos auxiliares para construir las etapas de la canalización anteriores.

El siguiente ejemplo crea un pipeline que utiliza el método Aggregates.filter() para compilar la etapa $match. Luego, el código pasa este pipeline al método watch() e instruye a watch() salir de eventos solo cuando ocurren operaciones de actualización:

val observer = LatchedObserver()
collection.watch(Seq(Aggregates.filter(Filters.in("operationType", "update"))))
observer.await()

Modificar el comportamiento de watch()

Puede modificar el comportamiento del método watch() encadenando los métodos proporcionados por la clase ChangeStreamObservable. La siguiente tabla describe algunos de estos métodos:

Para una lista completa de las opciones de watch(), consulte ChangeStreamObservable en la documentación de la API.

val observer = LatchedObserver()
collection.watch()
          .fullDocument(FullDocument.UPDATE_LOOKUP)
          .subscribe(observer)
observer.await()

ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null,
fullDocument=Iterable((_id,BsonObjectId{...}), (address,{"building": "202-24",
"coord": [-73.9250442, 40.5595462], "street": "Rockaway Point Boulevard",
"zipcode": "11697"}), (borough,BsonString{value='Queens'}),
(cuisine,BsonString{value='Irish'}), (grades,BsonArray{values=[...]}),
(name,BsonString{value='Blarney Castle'}), (restaurant_id,BsonString{...}),
(blank,BsonString{value='Irish'})), fullDocumentBeforeChange=null,
documentKey={"_id": {"$oid": "..."}}, clusterTime=Timestamp{...}, updateDescription=
UpdateDescription{removedFields=[], updatedFields={"cuisine": "Irish"},
truncatedArrays=[], disambiguatedPaths=null}, txnNumber=null, lsid=null,
splitEvent=null, wallTime=BsonDateTime{...}}

Información Adicional

Para obtener más información sobre los flujos de cambio, consulte Flujos de cambio en el manual de MongoDB Server.