Change streams on collections in MongoDB - Mini Project: Build & Apply

Monitor Real-Time Changes with MongoDB Change Streams

📖 Scenario: You are building a simple inventory system for a small store. You want to watch for any changes in the products collection so you can update your dashboard in real-time whenever a product is added, updated, or deleted.

🎯 Goal: Create a MongoDB change stream on the products collection to listen for all changes and print the change events.

📋 What You'll Learn

Create a MongoDB collection named products with some sample documents.

Set up a change stream on the products collection.

Listen for all change events (insert, update, delete).

Print the change event documents as they occur.

💡 Why This Matters

🌍 Real World

Change streams help applications react instantly to database changes, like updating dashboards, syncing data, or triggering alerts.

💼 Career

Understanding change streams is useful for backend developers, database administrators, and anyone building real-time data-driven applications.

1

Create the products collection with sample data

Create a MongoDB collection called products and insert these exact documents: { _id: 1, name: "Apple", quantity: 50 }, { _id: 2, name: "Banana", quantity: 30 }, and { _id: 3, name: "Orange", quantity: 20 }.

MongoDB

// Insert the sample documents into the products collection
// Your code here

Hint

Use db.products.insertMany([...]) to add multiple documents at once.

2

Create a change stream cursor on the products collection

Create a variable called changeStream that opens a change stream on the products collection using db.products.watch().

MongoDB

db.products.insertMany([
  { _id: 1, name: "Apple", quantity: 50 },
  { _id: 2, name: "Banana", quantity: 30 },
  { _id: 3, name: "Orange", quantity: 20 }
])

// Create the changeStream variable to watch the products collection
// Your code here

Hint

Use const changeStream = db.products.watch() to start watching changes.

3

Listen for change events using the change stream cursor

Use a while loop with await changeStream.hasNext() and inside the loop assign const change = await changeStream.next() to get each change event.

MongoDB

db.products.insertMany([
  { _id: 1, name: "Apple", quantity: 50 },
  { _id: 2, name: "Banana", quantity: 30 },
  { _id: 3, name: "Orange", quantity: 20 }
])

const changeStream = db.products.watch()

// Use a while loop to wait for new change events
// Your code here

Hint

Use while (await changeStream.hasNext()) to wait for changes and const change = await changeStream.next() to get each event.

4

Print each change event inside the loop

Inside the while loop, add printjson(change) to display the change event document.

MongoDB

db.products.insertMany([
  { _id: 1, name: "Apple", quantity: 50 },
  { _id: 2, name: "Banana", quantity: 30 },
  { _id: 3, name: "Orange", quantity: 20 }
])

const changeStream = db.products.watch()

while (await changeStream.hasNext()) {
  const change = await changeStream.next()
  // Print the change event document
  // Your code here
}

Hint

Use printjson(change) to nicely print the change event document.

Practice

1. What is the main purpose of using change streams on a MongoDB collection?

easy

A. To create indexes on the collection

B. To backup the collection data automatically

C. To watch for real-time changes happening in the collection

D. To delete documents based on a condition

Solution

1. Step 1: Understand what change streams do

   Change streams allow applications to listen for real-time changes like inserts, updates, or deletes in a collection.

2. Step 2: Compare options with this purpose

   Only To watch for real-time changes happening in the collection describes watching for real-time changes, which matches the purpose of change streams.

3. Final Answer:

   To watch for real-time changes happening in the collection -> Option C

4. Quick Check:

   Change streams = watch real-time changes [OK]

Hint: Change streams = real-time watch on collection changes [OK]

Common Mistakes:

• Thinking change streams backup data
• Confusing change streams with indexing
• Assuming change streams delete documents

2. Which of the following is the correct way to start a change stream on a collection named orders in MongoDB?

easy

A. db.orders.changeStream()

B. db.orders.watch()

C. db.orders.listen()

D. db.orders.observe()

Solution

1. Step 1: Recall the MongoDB method for change streams

   The correct method to open a change stream on a collection is watch().

2. Step 2: Match the method with the collection name

   Using db.orders.watch() correctly starts the change stream on the orders collection.

3. Final Answer:

   db.orders.watch() -> Option B

4. Quick Check:

   Change stream method = watch() [OK]

Hint: Use .watch() to start change streams on collections [OK]

Common Mistakes:

• Using non-existent methods like changeStream()
• Confusing with listen() or observe() which don't exist
• Misspelling the method name

3. Given the following code snippet, what will be the output when a new document is inserted into the products collection?

const changeStream = db.products.watch();
changeStream.on('change', (change) => {
  console.log(change.operationType);
});

medium

A. 'insert'

B. 'update'

C. 'delete'

D. 'replace'

Solution

1. Step 1: Understand the event listener on change stream

   The code listens for any change and logs the operationType of that change.

2. Step 2: Identify the operation type for inserting a document

   When a new document is inserted, the operationType is insert.

3. Final Answer:

   'insert' -> Option A

4. Quick Check:

   Insert operation logs 'insert' [OK]

Hint: Insert triggers operationType 'insert' in change stream [OK]

Common Mistakes:

• Confusing insert with update or delete
• Expecting full document instead of operationType
• Misreading event listener output

4. You wrote this code to watch changes on the users collection but it throws an error:

const changeStream = db.users.watch({ operationType: 'insert' });

What is the mistake?

medium

A. The operationType should be capitalized as 'Insert'

B. The collection name should be user not users

C. You must call start() after watch()

D. The watch method does not accept a filter object directly

Solution

1. Step 1: Check the argument to watch()

   The watch() method expects an array pipeline for filtering, not a direct object.

2. Step 2: Understand correct filtering syntax

   To filter by operationType, you must pass a pipeline like [{ $match: { operationType: 'insert' } }].

3. Final Answer:

   The watch method does not accept a filter object directly -> Option D

4. Quick Check:

   Filter must be pipeline array, not object [OK]

Hint: Use pipeline array for filters in watch() [OK]

Common Mistakes:

• Passing filter as object instead of pipeline array
• Wrong collection name assumption
• Expecting a start() method
• Capitalizing operationType values incorrectly

5. You want to build a live notification system that only alerts when a document in the tasks collection is updated and the status field changes to 'completed'. Which change stream pipeline correctly filters these changes?

hard

A. [{ $match: { operationType: 'update', 'updateDescription.updatedFields.status': 'completed' } }]

B. [{ $match: { operationType: 'insert', 'fullDocument.status': 'completed' } }]

C. [{ $match: { 'updateDescription.updatedFields.status': 'completed' } }]

D. [{ $match: { operationType: 'delete', 'updateDescription.updatedFields.status': 'completed' } }]

Solution

1. Step 1: Identify the operation type needed

   We want to watch only updates, so operationType must be 'update'.

2. Step 2: Filter for status field changed to 'completed'

   The updateDescription.updatedFields.status contains changed fields; we check if it equals 'completed'.

3. Step 3: Combine filters in a pipeline

   The correct pipeline is [{ $match: { operationType: 'update', 'updateDescription.updatedFields.status': 'completed' } }].

4. Final Answer:

   [{ $match: { operationType: 'update', 'updateDescription.updatedFields.status': 'completed' } }] -> Option A

5. Quick Check:

   Filter updates with status='completed' = [{ $match: { operationType: 'update', 'updateDescription.updatedFields.status': 'completed' } }] [OK]

Hint: Filter update and updatedFields.status in $match pipeline [OK]

Common Mistakes:

• Filtering inserts or deletes instead of updates
• Checking fullDocument instead of updateDescription
• Missing operationType filter
• Using wrong field path in filter