$ChangeStream

Tidak didukung oleh cluster Elastic.

T $changeStream ahap agregasi membuka kursor aliran perubahan untuk memantau perubahan real-time pada koleksi. Ini mengembalikan dokumen peristiwa perubahan saat menyisipkan, memperbarui, mengganti, atau menghapus operasi terjadi.

Parameter

• fullDocument: Menentukan apakah akan mengembalikan dokumen lengkap untuk operasi pembaruan. Opsi adalah default atauupdateLookup.

• resumeAfter: Opsional. Lanjutkan token untuk melanjutkan dari titik tertentu dalam aliran perubahan.

• startAtOperationTime: Opsional. Stempel waktu untuk memulai aliran perubahan.

• allChangesForCluster: Opsional. Nilai Boolean. Kapantrue, menonton semua perubahan di seluruh cluster (untuk database admin). Ketika false (default), hanya menonton koleksi yang ditentukan.

• showExpandedEvents: Opsional. Nilai Boolean. Kapantrue, sertakan pemberitahuan peristiwa bahasa definisi data yang diperluas (DDL), seperti createCollection dancreateIndexes, dalam aliran perubahan. Ketika false (default), peristiwa ini tidak dikembalikan. Tersedia di Amazon DocumentDB 8.0 dimulai dengan mesin versi 8.0.2.

Contoh (MongoDB Shell)

Contoh berikut menunjukkan penggunaan pang $changeStream gung untuk memantau perubahan pada koleksi.

Contoh kueri

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

Keluaran

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

Contoh kode

Untuk melihat contoh kode untuk menggunakan tahap $changeStream agregasi, pilih tab untuk bahasa yang ingin Anda gunakan:

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