Mongo.watch() (método mongosh)

Definición

Mongo.watch( pipeline, options )

Solo para Sets de réplicas y clústeres fragmentados

Abre un(a) cambiar el cursor del flujo de cambios para un set de réplicas o un clúster para informar sobre todas sus colecciones no-system en sus bases de datos, con la excepción de las bases de datos admin, local y config.

Parameter	Tipo	Descripción
pipeline	arreglo	Opcional. Una canalización de agregación que consiste en una o más de las siguientes etapas de agregación: $addFields $match $project $replaceRoot $replaceWith $redact $set $unset Especifica un pipeline para filtrar o modificar la salida de los eventos de cambio. A partir de MongoDB 4.2, los flujos de cambios lanzarán una excepción si la pipeline de agregación de flujos de cambios modifica el campo _id de un evento.
options	Documento	Opcional. Opciones adicionales que modifican el comportamiento de Mongo.watch().

El documento options puede contener los siguientes campos y valores:

Campo	Tipo	Descripción
resumeAfter	Documento	Opcional. Indica a Mongo.watch() que intente reanudar las notificaciones comenzando después de la operación especificada en el token de reanudación. Cada documento de evento de flujo de cambios incluye un token de reanudación como el campo _id. Pasa todo el campo _id del documento de evento de cambio que representa la operación que deseas reanudar después. resumeAfter es mutuamente excluyente con startAfter y startAtOperationTime.
startAfter	Documento	Opcional. Indica a Mongo.watch() que intente iniciar un nuevo flujo de cambios después de la operación especificada en el token de reanudación. Permite que las notificaciones se reanuden después de un evento de invalidación. Cada documento de evento de flujo de cambios incluye un token de reanudación como el campo _id. Pasa todo el campo _id del documento de evento de cambio que representa la operación que deseas reanudar después. startAfter es mutuamente excluyente con resumeAfter y startAtOperationTime.
fullDocument	string	Opcional. Por defecto, Mongo.watch() devuelve el delta de los campos que han sido modificados por una operación de actualización, en lugar de todo el documento actualizado. Configura fullDocument en "updateLookup" para dirigir a Mongo.watch() a buscar la versión más reciente del documento actualizado que haya sido comprometida por la mayoría. Mongo.watch() devuelve un campo fullDocument con la búsqueda de documentos además del delta de updateDescription.
batchSize	Int	opcional. Especifica el número máximo de eventos de cambio a devolver en cada agrupar de la respuesta del clúster de MongoDB. Tiene la misma funcionalidad que cursor.batchSize().
maxAwaitTimeMS	Int	Opcional. El tiempo máximo en milisegundos que el servidor espera para que los nuevos cambios de datos se informen al cursor del flujo de cambios antes de devolver una agrupación vacía. Se establece por defecto en 1000 milisegundos.
collation	Documento	opcional. Pase un documento de intercalación para especificar la intercalación para el cursor del flujo de cambios. Si se omite, por defecto se realizará la comparación binaria simple.
startAtOperationTime	Marca de tiempo	Opcional. El punto de inicio para el flujo de cambios. Si el punto de inicio especificado está en el pasado, debe encontrarse dentro del rango de tiempo del oplog. Para comprobar el rango temporal del oplog, consulta rs.printReplicationInfo(). startAtOperationTime es mutuamente excluyente con resumeAfter y startAfter.

Devuelve:	Un cursor sobre los documentos de eventos de cambio. Consulta Eventos de cambio para ver ejemplos de documentos de eventos de cambio.

Tip

db.collection.watch() y db.watch()

Compatibilidad

Este método está disponible en implementaciones alojadas en los siguientes entornos:

• MongoDB Atlas: El servicio totalmente gestionado para implementaciones de MongoDB en la nube

• MongoDB Enterprise: La versión basada en suscripción y autogestionada de MongoDB

• MongoDB Community: La versión de MongoDB con código fuente disponible, de uso gratuito y autogestionada.

Disponibilidad

Comportamiento

• Mongo.watch() solo notifica a la mayoría de los miembros portadores de datos sobre cambios en los datos que han persistido.

• El cursor de flujo de cambios permanece abierto hasta que ocurra una de las siguientes situaciones:

  • El cursor se cierra explícitamente.

  • Se produce un evento de invalidación; por ejemplo, un descarte o cambio de nombre de una colección.

  • La conexión a la implementación de MongoDB se cierra o se agota el tiempo de espera. Consulta Comportamiento para obtener más información.

  • Si la implementación es un clúster particionado, la eliminación de una partición puede provocar que un cursor del flujo de cambios abierto se cierre. Es posible que el cursor cerrado del flujo de cambios no sea completamente reanudable.

snippet install resumetoken
decodeResumeToken('<RESUME TOKEN>')

npx mongodb-resumetoken-decoder <RESUME TOKEN>

Control de acceso

Al ejecutar con control de acceso, el usuario debe tener las acciones de privilegio find y changeStream en todas las colecciones que no sean del sistema en todas las bases de datos. Es decir, un usuario debe tener un rol que otorgue el siguiente privilegio:

{ resource: { db: "", collection: "" }, actions: [ "find", "changeStream" ] }

El rol read incorporado proporciona los privilegios apropiados.

Iteración de cursor

MongoDB ofrece múltiples maneras de iterar sobre un cursor.

El cursor.hasNext() método se bloquea y espera el siguiente evento. Para supervisar el cursor watchCursor e iterar sobre los eventos, utiliza hasNext() de esta manera:

while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}

El cursor.tryNext() método es no bloqueante. Para supervisar el cursor watchCursor e iterar sobre los eventos, utiliza tryNext() de esta manera:

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

Ejemplo

La siguiente operación en mongosh abre un cursor de flujo de cambios en un set de réplicas. El cursor devuelto informa sobre los cambios en los datos en todas las colecciones que no sonsystem, en todas las bases de datos excepto en las bases de datos admin, local y config.

watchCursor = db.getMongo().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.