How do you configure and use MongoDB Change Streams to track data changes in real-time?

MongoDB Change Streams allow application developers to monitor changes to their MongoDB data in real-time. Change Streams can watch for changes at the individual document level or at the database or collection level, and they can be configured to send notifications to applications whenever documents are inserted, modified, or deleted.

To use MongoDB Change Streams, you first need to create a cursor on a MongoDB collection or database. You can do this by calling the ‘watch()‘ method on the collection object, passing in a query object that specifies the data changes you want to track. For example, to track all insertions into a collection, you can call:

const cursor = db.collection('mycollection').watch([{ $match: { operationType: 'insert' } }])

Once you have a cursor, you can use it to read pending changes from the MongoDB server in real-time. The cursor is updated every time a change is detected, so you can wait for new changes to arrive by polling the cursor periodically:

while (!cursor.isClosed()) {
  if (cursor.hasNext()) {
    const next = cursor.next()
    console.log(next)
  } else {
    sleep(1000)
  }
}

Change Stream events are represented as plain JavaScript objects, containing information about the type of change (‘insert‘, ‘update‘, ‘replace‘, ‘delete‘, ‘invalidate‘), the database and collection affected by the change, and the specific document(s) that were changed. For example:

{
  "_id": {
    "_data": "825FAD8D53...C57E8915023",
    "operationType": "insert"
  },
  "fullDocument": {
    "_id": "5c003155abf51d0fd7b23d72",
    "name": "Alice",
    "age": 30
  },
  "ns": {
    "db": "mydatabase",
    "coll": "mycollection"
  },
  "documentKey": {
    "_id": "5c003155abf51d0fd7b23d72"
  }
}

Change Streams can also be customized to include additional filtering, sorting, or aggregation operations before they are read by the client. For example:

const pipeline = [
  { $match: { operationType: 'insert', 'fullDocument.age': { $gt: 20 } } },
  { $sort: { 'fullDocument.age': -1 } },
  { $group: { _id: null, count: { $sum: 1 } } }
]

const options = {
  fullDocument: 'updateLookup',
  startAtOperationTime: new Timestamp(0, Date.now() / 1000),
  batchSize: 100
}

const cursor = db.collection('mycollection').watch(pipeline, options)

Here, we are watching for ‘insert‘ operations where the ‘age‘ field of the new document is greater than ‘20‘. We are then sorting the resulting documents by age in descending order and aggregating them into a single document that contains a count of the number of matching documents.

Overall, MongoDB Change Streams offer a powerful and flexible tool for monitoring live data changes in MongoDB. By configuring and using Change Streams effectively, developers can build applications that respond to data changes in real-time, improving their responsiveness and utility for end-users.