db.colección.watch() (método mongosh)

Implementación

db.collection.watch() está disponible para implementaciones de Sets de réplicas y clústeres:

• Para un set de réplicas, se puede emitir db.collection.watch() en cualquier Nodo que contenga datos.

• Para un clúster particionado, se debe emitir db.collection.watch() en una instancia mongos.

Motor de almacenamiento

Solo se puede usar db.collection.watch() con el motor de almacenamiento WiredTiger.

Leer Preocupación majority Soporte

A partir de MongoDB,4.2 losflujos de cambio están disponibles independientemente del "majority" soporte de lectura de preocupación; es decir, el soporte de lectura de majority preocupación puede estar habilitado (predeterminado) o deshabilitado para usar flujos de cambio.

En MongoDB 4.0 y versiones anteriores, losflujos de cambio solo están disponibles si "majority" el soporte de lectura de inquietudes está habilitado (predeterminado).

Reanudabilidad

A diferencia de los controladores de MongoDB, mongosh no intenta automáticamente reanudar un cursor de flujo de cambios después de un error. Los controladores de MongoDB hacen un intento de reanudar automáticamente un cursor de flujo de cambios después de ciertos errores.

db.collection.watch() utiliza la información almacenada en el oplog para generar la descripción del evento de cambio y crear un token de reanudación asociado a esa operación. Si la operación identificada por el token de reanudación pasado a la opción resumeAfter o startAfter ya ha salido del oplog, db.collection.watch() no puede reanudar el flujo de cambios.

Consulte Reanudar un flujo de cambios para obtener más información sobre cómo reanudar un flujo de cambios.

Consulta completa de documentos de operaciones de actualización

Por defecto, el cursor de flujo de cambios devuelve cambios/deltas específicos de campo para las operaciones de actualización. También puedes configurar el flujo de cambios para buscar y devolver la versión actual del documento modificado que ha sido comprometida por la mayoría. Dependiendo de otras operaciones de guardado que puedan haber ocurrido entre la actualización y la búsqueda, el documento devuelto puede diferir significativamente del documento en el momento de la actualización.

Dependiendo del número de cambios aplicados durante la operación de actualización y del tamaño del documento completo, existe el riesgo de que el tamaño del documento de evento de cambio para una operación de actualización sea mayor que el límite de 16 MB para documentos BSON. Si esto ocurre, el servidor cierra el cursor del flujo de cambios y devuelve un error.

Abre un flujo de cambios

La siguiente operación abre un cursor de flujo de cambios en la colección data.sensors:

watchCursor = db.getSiblingDB("data").sensors.watch()

Itera el cursor para verificar si hay nuevos eventos. Utiliza el método cursor.isClosed() junto con el método cursor.tryNext() para asegurarte de que el bucle solo salga si el cursor del flujo de cambios está cerrado y no quedan objetos en el último grupo:

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

Para obtener la documentación completa sobre la salida del flujo de cambios, consultar Eventos de cambios.

Flujo de cambios con búsqueda de actualización de documentos completos

Configurar la opción fullDocument en "updateLookup" para dirigir el cursor del flujo de cambios a buscar la versión más reciente comprometida por mayoría del documento asociado a un evento de actualización del flujo de cambios.

La siguiente operación abre un cursor de flujo de cambios contra la colección data.sensors usando la opción fullDocument : "updateLookup".

watchCursor = db.getSiblingDB("data").sensors.watch(
   [],
   { fullDocument : "updateLookup" }
)

Itera el cursor para verificar si hay nuevos eventos. Utiliza el método cursor.isClosed() junto con el método cursor.tryNext() para asegurarte de que el bucle solo salga si el cursor del flujo de cambios está cerrado y no quedan objetos en el último grupo:

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

Para cualquier operación de actualizar, el evento de cambio devuelve el resultado de la búsqueda del documento en el campo fullDocument.

Para un ejemplo del resultado completo de la actualización del documento, vea evento de actualización del flujo de cambios.

Para obtener la documentación completa sobre la salida del flujo de cambios, consultar Eventos de cambios.

Change Streams con imágenes previas y posteriores de los documentos

A partir de MongoDB 6.0, puedes utilizar eventos de flujo de cambios para generar la versión de un documento antes y después de los cambios (las imágenes previas y posteriores del documento):

• La imagen previa es el documento antes de reemplazarlo, actualizarlo o borrarlo. No existe una imagen previa para un documento insertado.

• La imagen posterior es el documento tras insertarse, sustituirse o actualizarse. No hay imagen posterior para un documento borrado.

• Activa changeStreamPreAndPostImages para una colección con db.createCollection(), create o collMod.

Las imágenes previas y posteriores no están disponibles para un evento de flujo de cambios si las imágenes fueron:

• No está habilitado en la colección en el momento de la operación de actualización o eliminación de un documento.

• Eliminado después del tiempo de retención de imágenes previas y posteriores establecido en expireAfterSeconds.

  • El siguiente ejemplo configura expireAfterSeconds en 100 segundos en todo el clúster:

    use admin
    db.runCommand( {
       setClusterParameter:
          { changeStreamOptions: {
             preAndPostImages: { expireAfterSeconds: 100 }
          } }
    } )

  • El siguiente ejemplo devuelve la configuración actual de changeStreamOptions, incluyendo expireAfterSeconds:

    db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

  • Configurar expireAfterSeconds en off utiliza la política de retención por defecto: las imágenes previas y posteriores se conservan hasta que los eventos correspondientes del flujo de cambios se eliminan del oplog.

  • Si se elimina un evento de flujo de cambios del oplog, las imágenes previas y posteriores correspondientes también se eliminan independientemente del tiempo de retención de las imágenes previas y posteriores expireAfterSeconds.

Consideraciones adicionales:

• Habilitar las imágenes previas y posteriores consume espacio de almacenamiento y aumenta el tiempo de procesamiento. Habilita solo las imágenes previas y de publicación si las necesitas.

• Limita el tamaño del evento del flujo de cambios a menos de 16 mebibytes. Para limitar el tamaño del evento, puedes:

  • Limita el tamaño del documento a 8 megabytes. Puedes solicitar imágenes previas y posteriores simultáneamente en la salida del flujo de cambios si otros campos de eventos del flujo de cambios como updateDescription no son grandes.

  • Solicita solo imágenes posteriores en la salida del flujo de cambios para documentos de hasta 16 mebibytes si otros campos de eventos del flujo de cambios como updateDescription no son grandes.

  • Solicita solo imágenes previas en la salida del flujo de cambios para documentos de hasta 16 mebibytes si:

    • las actualizaciones de documento afectan solo a una pequeña fracción de la estructura o el contenido del documento, y

    • no cause un evento de cambio replace. Un evento replace siempre incluye la imagen de publicación.

• Para solicitud de una pre-imagen, configura fullDocumentBeforeChange en required o whenAvailable en db.collection.watch(). Para solicitar una imagen de publicación, configura fullDocument usando el mismo método.

• Las preimágenes se escriben en la colección config.system.preimages.

  • La colección config.system.preimages puede agrandarse. Para limitar el tamaño de la colección, puedes establecer el tiempo a expireAfterSeconds para las imágenes previas como se mostró antes.

  • Las imágenes previas se eliminan de forma asincrónica mediante un proceso en segundo plano.

db.createCollection(
   "temperatureSensor",
   { changeStreamPreAndPostImages: { enabled: true } }
)

db.temperatureSensor.insertMany( [
   { "_id" : 0, "reading" : 26.1 },
   { "_id" : 1, "reading" : 25.9 },
   { "_id" : 2, "reading" : 24.3 },
   { "_id" : 3, "reading" : 22.4 },
   { "_id" : 4, "reading" : 24.6 }
] )

watchCursorFullDocumentBeforeChange = db.temperatureSensor.watch(
   [],
   { fullDocumentBeforeChange: "whenAvailable" }
)

while ( !watchCursorFullDocumentBeforeChange.isClosed() ) {
   if ( watchCursorFullDocumentBeforeChange.hasNext() ) {
      printjson( watchCursorFullDocumentBeforeChange.next() );
   }
}

db.temperatureSensor.updateOne(
   { _id: 2 },
   { $set: { reading: 22.1 } }
)

{
   "_id" : {
      "_data" : "82624B21...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1649090957, 1),
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 2
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 22.1
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   },
   "fullDocumentBeforeChange" : {
      "_id" : 2,
      "reading" : 24.3
   }
}

watchCursorFullDocument = db.temperatureSensor.watch(
   [],
   { fullDocument: "whenAvailable" }
)

while ( !watchCursorFullDocument.isClosed() ) {
   if ( watchCursorFullDocument.hasNext() ) {
      printjson( watchCursorFullDocument.next() );
   }
}

db.temperatureSensor.updateOne(
   { _id: 1 },
   { $set: { reading: 29.5 } }
)

{
   "_id" : {
      "_data" : "8262474D...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1648840090, 1),
   "fullDocument" : {
      "_id" : 1,
      "reading" : 29.5
   },
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 1
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 29.5
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   }
}

Flujo de cambios con filtro de canalización de agregación

La siguiente operación abre un cursor de flujo de cambios en la colección data.sensors utilizando una canalización de agregación para filtrar solo los eventos insert:

watchCursor = db.getSiblingDB("data").sensors.watch(
   [
      { $match : {"operationType" : "insert" } }
   ]
)

Itera el cursor para verificar si hay nuevos eventos. Utiliza el método cursor.isClosed() junto con el método cursor.hasNext() para asegurarte de que el bucle solo salga si el cursor del flujo de cambios está cerrado y no quedan objetos en el último grupo:

while (!watchCursor.isClosed()){
   if (watchCursor.hasNext()){
      printjson(watchCursor.next());
   }
}

El cursor de flujo de cambios solo devuelve eventos de cambio donde el operationType es insert. Para obtener la documentación completa sobre la salida del flujo de cambios, consulta Eventos de cambio.

Reanudar un flujo de cambios

Cada documento devuelto por un cursor de flujo de cambios incluye un token de reanudación como el campo _id. Para reanudar un flujo de cambios, pase el documento _id completo del evento de cambio del que desea reanudar a la opción resumeAfter o startAfter de watch().

La siguiente operación reanuda un cursor de flujo de cambios en la colección data.sensors utilizando un token de reanudación. Esto asume que la operación que generó el token de reanudación no se ha eliminado del oplog del clúster.

let watchCursor = db.getSiblingDB("data").sensors.watch();
let firstChange;
while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}
watchCursor.close();
let resumeToken = firstChange._id;
resumedWatchCursor = db.getSiblingDB("data").sensors.watch(
[],
   { resumeAfter : resumeToken }
)

Itera el cursor para verificar si hay nuevos eventos. Utiliza el método cursor.isClosed() junto con el método cursor.hasNext() para asegurarte de que el bucle solo salga si el cursor del flujo de cambios está cerrado y no quedan objetos en el último grupo:

while (!resumedWatchCursor.isClosed()){
   if (resumedWatchCursor.hasNext()){
      print(resumedWatchCursor.next());
   }
}

Consulta Reanudar un flujo de cambios para obtener documentación completa sobre cómo reanudar un flujo de cambios.