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.

Altere fluxos com pré e pós-imagens de documentos

A partir do MongoDB 6.0, você pode usar eventos change stream para produzir a versão de um documento antes e depois das alterações (pré e pós-imagens do documento):

• A pré-imagem é o documento antes de ser substituído, atualizado ou excluído. Não há pré-imagem para um documento inserido.

• A pós-imagem é o documento após ter sido inserido, substituído ou atualizado. Não há pós-imagem para um documento excluído.

• Habilite changeStreamPreAndPostImages para uma coleção usando db.createCollection(), create ou collMod.

As imagens pré e pós não estarão disponíveis para um change stream se as imagens forem:

• Não habilitadas na coleção no momento de uma operação de atualização ou exclusão de documento.

• Removido após o tempo de retenção pré e pós-imagem definido em expireAfterSeconds.

  • O exemplo a seguir define expireAfterSeconds para 100 segundos em um cluster inteiro:

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

  • O exemplo a seguir retorna as configurações atuais do changeStreamOptions, incluindo expireAfterSeconds:

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

  • Definir expireAfterSeconds para off usa a política de retenção: pré-imagens e pós-imagens são retidas até os eventos de change streams serem removidos do oplog.

  • Se um change stream for removido do oplog, as imagens pré e pós correspondentes também serão excluídas, independentemente do tempo de retenção pré e pós-imagem expireAfterSeconds .

Considerações adicionais:

• Habilitar pré e pós-imagens consome espaço de armazenamento e adiciona tempo de processamento. Ative as imagens anteriores e posteriores somente se precisar delas.

• Limite o tamanho do evento de transmissão da alteração para menos de 16 megabytes. Para limitar o tamanho do evento, você pode:

  • Limite o tamanho do documento a 8 megabytes. Você pode solicitar imagens pré e pós simultaneamente na saída do change stream se outros campos de evento de change stream, como updateDescription não forem grandes.

  • Solicite apenas pós-imagens na saída de fluxo de alteração para documentos de até 16 megabytes se outros campos de evento de fluxo de alteração como updateDescription não forem grandes.

  • Solicite somente pré-imagens na saída do change stream para documentos de até 16 megabytes se:

    • as atualizações do documento afetam apenas uma pequena fração da estrutura ou conteúdo do documento, e

    • não causa um evento de alteração replace . Um evento replace sempre inclui o pós-imagem.

• Para solicitar uma pré-imagem, defina fullDocumentBeforeChange como required ou whenAvailable em db.collection.watch(). Para solicitar uma pós-imagem, defina fullDocument usando o mesmo método.

• As pré-imagens são escritas na coleção config.system.preimages.

  • A coleção config.system.preimages pode ficar grande. Para limitar o tamanho da coleção, você pode definir expireAfterSeconds tempo para as pré-imagens, conforme mostrado anteriormente.

  • As pré-imagens são removidas de forma assíncrona por um processo de plano de fundo.

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" : [ ]
   }
}

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.