$Cambia stream

Non supportato da Elastic Cluster.

La fase di $changeStream aggregazione apre un cursore del flusso di modifiche per monitorare le modifiche in tempo reale a una raccolta. Restituisce i documenti relativi agli eventi di modifica quando si verificano operazioni di inserimento, aggiornamento, sostituzione o eliminazione.

Parametri

• fullDocument: specifica se restituire il documento completo per le operazioni di aggiornamento. Le opzioni sono default o updateLookup.

• resumeAfter: facoltativo. Riprendi il token per continuare da un punto specifico del flusso di modifiche.

• startAtOperationTime: facoltativo. Timestamp da cui iniziare il flusso di modifiche.

• allChangesForCluster: facoltativo. Valore booleano. Quandotrue, controlla tutte le modifiche nel cluster (per il database di amministrazione). Quando false (impostazione predefinita), guarda solo la raccolta specificata.

• showExpandedEvents: facoltativo. Valore booleano. Quandotrue, include le notifiche di eventi DDL (Expanded Data Definition Language), ad esempio createCollection ecreateIndexes, nel flusso di modifiche. Quando false (impostazione predefinita), questi eventi non vengono restituiti. Disponibile in Amazon DocumentDB 8.0 a partire dalla versione del motore 8.0.2.

Esempio (MongoDB Shell)

L'esempio seguente dimostra l'utilizzo dello $changeStream stage per monitorare le modifiche a una raccolta.

Esempio di interrogazione

// 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()));
}

Output

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

Esempi di codice

Per visualizzare un esempio di codice per l'utilizzo della fase di $changeStream aggregazione, scegli la scheda relativa alla lingua che desideri utilizzare:

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