Change Streams

Section ini sudah menyelesaikan tiga tipe collection untuk menyimpan data; sekarang kita melihat cara memantau perubahan data itu sendiri. Selama ini aplikasi kita aktif menanyakan database: “ada pesanan baru?” lewat polling, atau menunggu job terjadwal berjalan. Change streams membalik arahnya — database yang memberi tahu aplikasi setiap kali data berubah, hampir seketika. Chapter ini adalah pengantar ringkas: konsepnya, syaratnya, satu contoh mini yang teruji, lalu alihan ke section khusus yang membahasnya sampai ke arsitektur event-driven.

Konsep: Berlangganan Perubahan #

Change stream adalah cursor yang tidak berhenti di sekumpulan document, melainkan terus mengalirkan change event setiap kali ada insert, update, delete, atau replace. Tiga fakta dasarnya:

• Dibangun di atas oplog replica set. Setiap event berasal dari entri oplog, sehingga change stream hanya tersedia pada replica set (atau sharded cluster) — standalone server menolaknya. Lab kita memakai replica set rs0, jadi semuanya bisa diuji langsung.
• Hanya perubahan yang sudah majority-committed. Event baru mengalir setelah tulisan diakui mayoritas anggota, sehingga aplikasi tidak bereaksi terhadap data yang berpotensi dibatalkan saat failover.
• Bisa dilanjutkan (resumable). Setiap event membawa resume token; bila koneksi terputus, stream bisa dibuka ulang tepat dari posisi terakhir.

Aplikasi klasiknya jelas: invalidasi cache saat data sumber berubah, sinkronisasi ke sistem eksternal, notifikasi real-time, dan audit. Satu catatan khusus section ini: time series collection justru tidak mendukung change streams — kita sudah membuktikan galatnya di Chapter Querying Time Series — jadi demo di sini memakai collection pesanan biasa.

Contoh Mini: Menonton Collection pesanan #

Pola dasarnya tiga langkah: buka stream dengan watch(), lakukan tulisan, lalu ambil event pertama dengan next():

use latihan_timeseries
db.dropDatabase()
const cs = db.pesanan.watch()
db.pesanan.insertOne({ kode: "PSN-001", pelanggan: "Budi", total: 250000 })
cs.next()

Output yang Dihasilkan:

switched to db latihan_timeseries
{ ok: 1, dropped: 'latihan_timeseries' }

{
  acknowledged: true,
  insertedId: ObjectId('6ad3e188302c9deb1db13de8')
}
Warning: If there are no documents in the batch, next will block. Use tryNext if you want to check if there are any documents without waiting, or cursor.disableBlockWarnings() if you want to disable this warning.
{
  _id: {
    _data: '826AD3E188000000022B042C0100296E5A10045C4E90BE68ED4B6C9BE5B484AF000FE2463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F696400646AD3E188302C9DEB1DB13DE8000004'
  },
  operationType: 'insert',
  clusterTime: Timestamp({ t: 1792270728, i: 2 }),
  wallTime: ISODate('2026-10-17T20:58:48.049Z'),
  fullDocument: {
    _id: ObjectId('6ad3e188302c9deb1db13de8'),
    kode: 'PSN-001',
    pelanggan: 'Budi',
    total: 250000
  },
  ns: { db: 'latihan_timeseries', coll: 'pesanan' },
  documentKey: { _id: ObjectId('6ad3e188302c9deb1db13de8') }
}

Satu baris watch() mengembalikan cursor yang menampung event di atas. Peringatan mongosh tentang next yang akan menunggu memang sengaja ditampilkan: next() memang memblokir sampai event berikutnya tiba — itulah sifat stream, dan di aplikasi nyata loop inilah yang berjalan terus di worker terpisah. Isi eventnya sudah informatif: operationType: 'insert', document lengkap di fullDocument, lokasi di ns, dan _id berupa token heksadesimal untuk resume.

Stream juga bisa dipersempit dengan filter pipeline, misalnya hanya insert:

const cs2 = db.pesanan.watch([{ $match: { operationType: "insert" } }])
cs2.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-002", pelanggan: "Siti", total: 175000 })
db.pesanan.updateOne({ kode: "PSN-002" }, { $set: { status: "dibayar" } })
cs2.next().fullDocument
cs2.close()

Output yang Dihasilkan:

ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3e189302c9deb1db13de9')
}
{
  acknowledged: true,
  insertedId: null,
  matchedCount: 1,
  modifiedCount: 1,
  upsertedCount: 0
}
{
  _id: ObjectId('6ad3e189302c9deb1db13de9'),
  kode: 'PSN-002',
  pelanggan: 'Siti',
  total: 175000
}

Insert PSN-002 dan update statusnya sama-sama terjadi, tetapi next() hanya mengembalikan event insert — $match pada pipeline menyaring sisanya. disableBlockWarnings() sekadar mematikan peringatan tadi agar keluaran REPL bersih.

Ke Mana Selanjutnya #

Contoh dua blok ini baru permukaan. Section Change Streams dan Events membahas topik ini secara penuh: struktur setiap field event, menonton satu collection atau seluruh database, resume token untuk melanjutkan stream pasca gangguan, penanganan koneksi yang terputus, sampai merancang arsitektur berbasis event di atasnya. Kalau Anda belum membaca Section Replication, Chapter Oplog di sana adalah bekal yang pas — change stream pada dasarnya adalah antarmuka elegan untuk data yang sama.