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.

Cuando se utiliza el controlador Scala, se puede llamar a la watch() método para devolver una instancia de ChangeStreamObservable. Luego, puedes suscribirte a la instancia de ChangeStreamObservable para ver los cambios en los datos nuevos, 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: Supervisa los cambios en todas las colecciones de una base de datos

• MongoCollection: Supervisa 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 de LatchedObserver para recibir y realizar cambios a medida que ocurra:

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

Para comenzar a observar cambios, ejecuta el código anterior. Luego, en una shell separada, ejecuta el siguiente código para actualizar un documento que tenga un valor de campo name igual a "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, se puede pasar una lista de etapas de la pipeline como parámetro al método watch(). Puede incluir las siguientes etapas en la lista:

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

• $matchFiltra los documentos

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

• $replaceWith o $replaceRoot: Reemplaza el documento de entrada por el documento especificado

• $redact:Restringe el contenido de los documentos

• $unsetRemueve campos de 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 saber más sobre las change streams, consulta Change Streams en el manual de MongoDB Server.