db.collection.watch()

Implantação

db.collection.watch() está disponível para implementações de conjunto de réplica e cluster fragmentado:

• Em um conjunto de réplicas, você pode emitir db.collection.watch() em qualquer membro portador de dados.

• Em um cluster fragmentado, você deve emitir db.collection.watch() em uma instância mongos.

Mecanismo de armazenamento

Você só pode usar db.collection.watch() com o mecanismo de armazenamento Wired Tiger.

Leia o suporte do Concern majority

A partir do MongoDB 4.2, os fluxos de alteração estão disponíveis independentemente do suporte à "majority" de leitura; ou seja, o suporte a majority de leitura pode ser habilitado (padrão) ou desabilitado para usar fluxos de alteração.

No MongoDB 4.0 e versões anteriores, os fluxos de alteração estarão disponíveis somente se "majority" suporte a preocupações de leitura estiver habilitado (padrão).

Retomabilidade

Ao contrário dos drivers do MongoDB, omongosh não tenta retomar automaticamente um cursor de fluxo de alterações após um erro. Os drivers do MongoDB fazem uma tentativa de retomar automaticamente um cursor de fluxo de alterações após determinados erros.

db.collection.watch() usa informações armazenadas no oplog para produzir a descrição do evento de alteração e gerar um token de retomada associado a essa operação. Se a operação identificada pelo token de retomada passada para a opção resumeAfter ou startAfter já tiver sido descartada do oplog, db.collection.watch() não poderá retomar o fluxo de alterações.

Consulte Retomar um fluxo de alterações para obter mais informações sobre como retomar um fluxo de alterações.

Pesquisa completa de documentos de operações de atualização

Por padrão, o cursor de fluxo de alterações retorna alterações/deltas de campos específicos para operações de atualização. Você também pode configurar o fluxo de alterações para procurar e retornar a versão atual de compromisso majoritário do documento alterado. Dependendo de outras operações de gravação que podem ter ocorrido entre a atualização e a pesquisa, o documento retornado pode diferir significativamente do documento no momento da atualização.

Dependendo do número de alterações aplicadas durante a operação de atualização e o tamanho do documento completo, há o risco de que o tamanho do documento do evento de alteração para uma operação de atualização seja maior do que os 16MB que são o limite de documentos BSON. Se isso ocorrer, o servidor fechará o cursor de fluxo de alterações e retornará um erro.

Abrir um fluxo de alterações

A operação a seguir abre um cursor do fluxo de alterações contra a coleção data.sensors:

watchCursor = db.getSiblingDB("data").sensors.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.

Fluxo de alterações com pesquisa completa de atualização de documentos

Defina a opção fullDocument como "updateLookup" para direcionar o cursor de fluxo de alteração para pesquisar a versão da maioria mais atual comprometida do documento associada a um evento de fluxo de alteração de atualização.

A operação a seguir abre um cursor de fluxo de alterações contra a coleção data.sensors usando a opção fullDocument : "updateLookup".

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

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 qualquer operação de atualização, o evento de alteração retorna o resultado da pesquisa do documento no campo fullDocument.

Para obter um exemplo da saída de atualização completa do documento, consulte alterar evento de atualização de fluxo.

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

Alterar fluxo com filtro de aggregation pipeline

A operação a seguir abre um cursor de fluxo de alterações na coleção data.sensors usando um pipeline de agregação para filtrar somente os eventos insert:

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

Itere o cursor para verificar novos eventos. Use o método cursor.isClosed() com o método cursor.hasNext() 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()){
   if (watchCursor.hasNext()){
      printjson(watchCursor.next());
   }
}

O cursor do fluxo de alterações retorna apenas eventos de mudança onde operationType é insert. Para obter documentação completa sobre a saída do change stream, consulte Change Events.

Retomando um fluxo de alterações

Cada documento retornado por um cursor do fluxo de alterações inclui um token de retomada como o campo _id. Para retomar um fluxo de alterações, passe todo o documento _id do evento de alteração do qual você deseja retomar para a opção resumeAfter ou startAfter de watch().

A operação a seguir retoma um cursor de fluxo de alterações em relação à coleção data.sensors usando um token de retomar. Isso pressupõe que a operação que gerou o token de continuação não foi eliminado do oplog do cluster.

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 }
)

Itere o cursor para verificar novos eventos. Use o método cursor.isClosed() com o método cursor.hasNext() 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 (!resumedWatchCursor.isClosed()){
   if (resumedWatchCursor.hasNext()){
      print(resumedWatchCursor.next());
   }
}

Consulte Resume a Change Stream para obter a documentação completa sobre como retomar um fluxo de alterações.