View a markdown version of this page

$ChangeStream - Amazon DocumentDB
Exemple (MongoDB Shell)Exemples de code

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

$ChangeStream

Non pris en charge par Elastic Cluster.

L'étape $changeStream d'agrégation ouvre un curseur de flux de modifications pour surveiller en temps réel les modifications apportées à une collection. Il renvoie les documents relatifs aux événements de modification lorsque des opérations d'insertion, de mise à jour, de remplacement ou de suppression sont effectuées.

Paramètres

  • fullDocument: Spécifie s'il faut renvoyer le document complet pour les opérations de mise à jour. Les options sont default ou updateLookup.

  • resumeAfter : facultatif. Token de reprise pour continuer à partir d'un point précis du flux de modifications.

  • startAtOperationTime : facultatif. Horodatage à partir duquel démarrer le flux de modifications.

  • allChangesForCluster : facultatif. Valeur booléenne. Quandtrue, surveille tous les changements dans le cluster (pour la base de données d'administration). Quand false (par défaut), surveille uniquement la collection spécifiée.

  • showExpandedEvents : facultatif. Valeur booléenne. Quandtrue, inclut des notifications d'événements en langage de définition des données étendu (DDL), telles que createCollection etcreateIndexes, dans le flux de modifications. Quand false (par défaut), ces événements ne sont pas renvoyés. Disponible dans Amazon DocumentDB 8.0 à partir de la version 8.0.2 du moteur.

Exemple (MongoDB Shell)

L'exemple suivant montre comment utiliser la $changeStream scène pour surveiller les modifications apportées à une collection.

Exemple de requête 

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

Sortie

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

Exemples de code

Pour afficher un exemple de code permettant d'utiliser l'étape d'$changeStreamagrégation, choisissez l'onglet correspondant à la langue que vous souhaitez utiliser :

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