$changeStream

Não é suportado pelo cluster elástico.

O estágio $changeStream de agregação abre um cursor de fluxo de alterações para monitorar as alterações em tempo real em uma coleção. Ele retorna documentos de eventos de alteração quando ocorrem operações de inserção, atualização, substituição ou exclusão.

Parâmetros

• fullDocument: especifica se o documento completo deve ser devolvido para operações de atualização. As opções são default ou updateLookup.

• resumeAfter: opcional. Retomar o token para continuar a partir de um ponto específico no fluxo de alterações.

• startAtOperationTime: opcional. Carimbo de data/hora a partir do qual iniciar o fluxo de alterações.

• allChangesForCluster: opcional. Valor booleano. Quandotrue, observa todas as alterações no cluster (para banco de dados administrativo). Quando false (padrão), observa somente a coleção especificada.

• showExpandedEvents: opcional. Valor booleano. Quandotrue, inclui notificações de eventos de linguagem de definição de dados (DDL) expandida, como createCollection ecreateIndexes, no fluxo de alterações. Quando false (padrão), esses eventos não são retornados. Disponível no Amazon DocumentDB 8.0 a partir da versão 8.0.2 do mecanismo.

Exemplo (MongoDB Shell)

O exemplo a seguir demonstra o uso do $changeStream palco para monitorar alterações em uma coleção.

Exemplo de consulta

// Open change stream first
const changeStream = db.inventory.aggregate([
  { $changeStream: { fullDocument: "updateLookup" } }
]);

// In another session, insert a document
db.inventory.insertOne({ _id: 1, item: "Widget", qty: 10 });

// Back in the first session, read the change event
if (changeStream.hasNext()) {
  print(tojson(changeStream.next()));
}

Saída

{
  _id: { _data: '...' },
  operationType: 'insert',
  clusterTime: Timestamp(1, 1234567890),
  fullDocument: { _id: 1, item: 'Widget', qty: 10 },
  ns: { db: 'test', coll: 'inventory' },
  documentKey: { _id: 1 }
}

Exemplos de código

Para ver um exemplo de código para usar o estágio de $changeStream agregação, escolha a guia do idioma que você deseja usar:

Node.js

const { MongoClient } = require('mongodb');

async function example() {
  const client = await MongoClient.connect('mongodb://<username>:<password>@<cluster-endpoint>:27017/?tls=true&tlsCAFile=global-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false');
  const db = client.db('test');
  const collection = db.collection('inventory');

  // Open change stream
  const changeStream = collection.watch([]);

  changeStream.on('change', (change) => {
    console.log('Change detected:', change);
  });

  // Simulate insert in another operation
  setTimeout(async () => {
    await collection.insertOne({ _id: 1, item: 'Widget', qty: 10 });
  }, 1000);

  // Keep connection open to receive changes
  // In production, handle cleanup appropriately
}

example();

Python

from pymongo import MongoClient
import threading
import time

def example():
    client = MongoClient('mongodb://<username>:<password>@<cluster-endpoint>:27017/?tls=true&tlsCAFile=global-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false')
    db = client['test']
    collection = db['inventory']

    # Open change stream
    change_stream = collection.watch([])

    # Insert document in separate thread after delay
    def insert_doc():
        time.sleep(1)
        collection.insert_one({'_id': 1, 'item': 'Widget', 'qty': 10})

    threading.Thread(target=insert_doc).start()

    # Watch for changes
    for change in change_stream:
        print('Change detected:', change)
        break  # Exit after first change

    client.close()

example()