Mongo.watch()

Definição

Mongo.watch( pipeline, options )

Somente para conjuntos de réplica e clusters fragmentados

Abre um cursor de change stream para um conjunto de réplicas ou um cluster fragmentado para relatar em todas as suas collections que não sejam system em seus bancos de dados, com exceção dos bancos de dados admin, local e config.

Parâmetro	Tipo	Descrição
pipeline	variedade	Opcional. Um pipeline de agregação que consiste em um ou mais dos seguintes estágios de agregação : $addFields $match $project $replaceRoot $replaceWith (Disponível a partir do MongoDB 4.2) $redact $set (Disponível a partir do MongoDB 4.2) $unset (Disponível a partir do MongoDB 4.2) Especifique um pipeline para filtrar/modificar o resultado dos eventos de alteração. A partir do MongoDB 4.2, os fluxos de alteração lançarão uma exceção se o pipeline de agregação do fluxo de alterações modificar o campo _id de um evento.
options	documento	Opcional. Opções adicionais que modificam o comportamento do Mongo.watch().

O documento options pode conter os seguintes campos e valores:

Campo	Tipo	Descrição
resumeAfter	documento	Opcional. Direciona Mongo.watch() para tentar retomar as notificações a partir da operação especificada no token de retomada. Cada documento de evento de fluxo de alteração inclui um token de currículo como o campo _id. Passe todo o _id campo do documento do evento de alteração que representa a operação que você deseja retomar depois. resumeAfter é mutuamente exclusivo com startAfter e startAtOperationTime.
startAfter	documento	Opcional. Direciona Mongo.watch() para tentar iniciar um novo fluxo de alterações após a operação especificada no token de retomada. Permite que as notificações sejam retomadas após um evento de invalidação. Cada documento de evento de fluxo de alteração inclui um token de currículo como o campo _id. Passe todo o _id campo do documento do evento de alteração que representa a operação que você deseja retomar depois. startAfter é mutuamente exclusivo com resumeAfter e startAtOperationTime. Novidades na versão 4.2.
fullDocument	string	Opcional. Por padrão, Mongo.watch() retorna o delta dos campos modificados por uma operação de atualização, em vez de todo o documento atualizado. Defina fullDocument como "updateLookup" para direcionar Mongo.watch() a procurar a versão mais atual confirmada por maioria do documento atualizado. Mongo.watch() retorna um campo fullDocument com a pesquisa do documento, além do delta updateDescription .
batchSize	int	Opcional. Especifica o número máximo de eventos de alteração a serem retornados em cada lote da resposta do cluster MongoDB. Tem a mesma funcionalidade que cursor.batchSize().
maxAwaitTimeMS	int	Opcional. A quantidade máxima de tempo em milissegundos que o servidor aguarda por novas alterações de dados para relatar ao cursor do fluxo de alterações antes de retornar um lote vazio. Padrão para 1000 milissegundos.
collation	documento	Opcional. Passe um documento de agrupamento para especificar um agrupamento para o cursor do change stream. Se omitido, o padrão é simple comparação binária.
startAtOperationTime	Timestamp	Opcional. O ponto de partida para o fluxo de alterações. Se o ponto de partida especificado estiver no passado, ele deverá estar no intervalo de tempo do oplog. Para verificar o intervalo de tempo do oplog, consulte rs.printReplicationInfo(). startAtOperationTime é mutuamente exclusivo com resumeAfter e startAfter.

Retorna:	Um cursor sobre os documentos do evento de alteração. Consulte Eventos de mudança para obter exemplos de documentos de eventos de mudança.

Dica

Veja também:

db.collection.watch() e a db.watch()

Disponibilidade

Comportamento

• Mongo.watch() notifica apenas sobre alterações de dados que persistiram para a maioria dos membros portadores de dados.

• O cursor do fluxo de alterações permanece aberto até que ocorra uma das seguintes situações:

  • O cursor está explicitamente fechado.

  • Ocorre um evento de invalidar; por exemplo, um drop de coleção ou renomear.

  • A conexão com a implantação do MongoDB fecha ou expira. Consulte Comportamentos do cursor para obter mais informações.

  • Se a implantação for um cluster fragmentado, uma remoção de fragmento pode fazer com que um cursor de fluxo de alteração aberto seja fechado e o cursor de fluxo de alterações fechado pode não ser totalmente retomável.

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

npx mongodb-resumetoken-decoder <RESUME TOKEN>

Controle de acesso

Ao executar com controle de acesso, o usuário deve ter as ações de privilégio find e changeStream em todas as collections que não são do sistema em todos os bancos de dados. Ou seja, um usuário deve ter uma role que conceda o seguinte privilégio:

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

O papel do read embutido fornece os privilégios apropriados.

Iteração do cursor

O MongoDB oferece várias maneiras de iterar em um cursor.

O método cursor.hasNext() bloqueia e aguarda o próximo evento. Para monitorar o cursor watchCursor e iterar sobre os eventos, use hasNext() assim:

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

O método cursor.tryNext() não está bloqueando. Para monitorar o cursor watchCursor e iterar sobre os eventos, use tryNext() assim:

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

Exemplo

A operação a seguir no mongosh abre um cursor de change stream em um conjunto de réplicas. O cursor retornado relata as alterações de dados em todas as coleções que não sãosystem em todos os bancos de dados, exceto os bancos de dados admin, local e config .

watchCursor = db.getMongo().watch()

Itere o cursor para verificar novos eventos. Use o método cursor.isClosed() com o método cursor.tryNext() para garantir que o loop só saia se o cursor do fluxo de alteração estiver fechado e não houver objetos restantes no lote mais recente:

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

Para obter a documentação completa sobre a saída do fluxo de alterações, consulte Eventos de alteração.