Monitor Data Changes
On this page
Overview
In this guide, you can learn how to use a change stream to monitor real-time changes to your data. A change stream is a feature that allows your application to subscribe to data changes on a collection, database, or deployment.
Sample Data
The examples in this guide use the sample_restaurants.restaurants
collection
from the Atlas sample datasets. To learn how to create a
free MongoDB Atlas cluster and load the sample datasets, see the Quick Start.
The examples on this page use the following Restaurant
, Address
, and GradeEntry
classes as models:
public class Restaurant { public ObjectId Id { get; set; } public string Name { get; set; } [ ] public string RestaurantId { get; set; } public string Cuisine { get; set; } public Address Address { get; set; } public string Borough { get; set; } public List<GradeEntry> Grades { get; set; } }
public class Address { public string Building { get; set; } [ ] public double[] Coordinates { get; set; } public string Street { get; set; } [ ] public string ZipCode { get; set; } }
public class GradeEntry { public DateTime Date { get; set; } public string Grade { get; set; } public float? Score { get; set; } }
Note
The documents in the restaurants
collection use the camel-case naming
convention. The examples in this guide use a ConventionPack
to deserialize the fields in the collection into Pascal case and map them to
the properties in the Restaurant
class.
To learn more about custom serialization, see Custom Serialization.
Open a Change Stream
To open a change stream, call the Watch()
or WatchAsync()
method. The instance on which you
call the method determines the scope of events that the change
stream listens for. You can call the Watch()
or WatchAsync()
method on the following
classes:
MongoClient
: To monitor all changes in the MongoDB deploymentDatabase
: To monitor changes in all collections in the databaseCollection
: To monitor changes in the collection
The following example opens a change stream on the restaurants
collection
and outputs the changes as they occur. Select the
Asynchronous or Synchronous tab to see the corresponding
code.
var database = client.GetDatabase("sample_restaurants"); var collection = database.GetCollection<Restaurant>("restaurants"); // Opens a change streams and print the changes as they're received using var cursor = await collection.WatchAsync(); await cursor.ForEachAsync(change => { Console.WriteLine("Received the following type of change: " + change.BackingDocument); });
var database = client.GetDatabase("sample_restaurants"); var collection = database.GetCollection<Restaurant>("restaurants"); // Opens a change stream and prints the changes as they're received using (var cursor = collection.Watch()) { foreach (var change in cursor.ToEnumerable()) { Console.WriteLine("Received the following type of change: " + change.BackingDocument); } }
To begin watching for changes, run the application. Then, in a separate
application or shell, modify the restaurants
collection. Updating a document
that has a "name"
value of "Blarney Castle"
results in the following
change stream output:
{ "_id" : { "_data" : "..." }, "operationType" : "update", "clusterTime" : Timestamp(...), "wallTime" : ISODate("..."), "ns" : { "db" : "sample_restaurants", "coll" : "restaurants" }, "documentKey" : { "_id" : ObjectId("...") }, "updateDescription" : { "updatedFields" : { "cuisine" : "Irish" }, "removedFields" : [], "truncatedArrays" : [] } }
Modify the Change Stream Output
You can pass the pipeline
parameter to the Watch()
and WatchAsync()
methods to modify the change stream output. This parameter allows you to watch
for only specified change events. Create the pipeline by using the
EmptyPipelineDefinition
class and appending the relevant aggregation stage methods.
You can specify the following aggregation stages in the pipeline
parameter:
$addFields
$match
$project
$replaceRoot
$replaceWith
$redact
$set
$unset
To learn how to build an aggregation pipeline by using the
PipelineDefinitionBuilder
class, see Build an Aggregation Pipeline in
the Operations with Builders guide.
The following example uses the pipeline
parameter to open a change stream
that records only update operations. Select the Asynchronous or Synchronous tab to see the
corresponding code.
var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>() .Match(change => change.OperationType == ChangeStreamOperationType.Update); // Opens a change stream and prints the changes as they're received using (var cursor = await _restaurantsCollection.WatchAsync(pipeline)) { await cursor.ForEachAsync(change => { Console.WriteLine("Received the following change: " + change); }); }
var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>() .Match(change => change.OperationType == ChangeStreamOperationType.Update); // Opens a change streams and print the changes as they're received using (var cursor = _restaurantsCollection.Watch(pipeline)) { foreach (var change in cursor.ToEnumerable()) { Console.WriteLine("Received the following change: " + change); } }
To learn more about modifying your change stream output, see the Modify Change Stream Output section in the manual.
Modify Watch()
Behavior
The Watch()
and WatchAsync()
methods accept optional parameters, which represent
options you can use to configure the operation. If you don't specify any
options, the driver does not customize the operation.
The following table describes the options you can set to customize the behavior
of Watch()
and WatchAsync()
:
Option | Description |
---|---|
FullDocument | Specifies whether to show the full document after the change, rather
than showing only the changes made to the document. To learn more about
this option, see Include Pre-Images and Post-Images. |
FullDocumentBeforeChange | Specifies whether to show the full document as it was before the change, rather
than showing only the changes made to the document. To learn more about
this option, see Include Pre-Images and Post-Images. |
ResumeAfter | Directs Watch() or WatchAsync() to resume returning changes after the
operation specified in the resume token.Each change stream event document includes a resume token as the _id
field. Pass the entire _id field of the change event document that
represents the operation you want to resume after.ResumeAfter is mutually exclusive with StartAfter and StartAtOperationTime . |
StartAfter | Directs Watch() or WatchAsync() to start a new change stream after the
operation specified in the resume token. Allows notifications to
resume after an invalidate event.Each change stream event document includes a resume token as the _id
field. Pass the entire _id field of the change event document that
represents the operation you want to resume after.StartAfter is mutually exclusive with ResumeAfter and StartAtOperationTime . |
StartAtOperationTime | Directs Watch() or WatchAsync() to return only events that occur after the
specified timestamp.StartAtOperationTime is mutually exclusive with ResumeAfter and StartAfter . |
MaxAwaitTime | Specifies the maximum amount of time, in milliseconds, the server waits for new
data changes to report to the change stream cursor before returning an
empty batch. Defaults to 1000 milliseconds. |
ShowExpandedEvents | Starting in v6.0, change streams support change notifications
for Data Definition Language (DDL) events, such as the createIndexes and dropIndexes events. To
include expanded events in a change stream, create the change stream
cursor and set this parameter to True . |
BatchSize | Specifies the maximum number of change events to return in each batch of the
response from the MongoDB cluster. |
Collation | Specifies the collation to use for the change stream cursor. |
Comment | Attaches a comment to the operation. |
Include Pre-Images and Post-Images
Important
You can enable pre-images and post-images on collections only if your deployment uses MongoDB v6.0 or later.
By default, when you perform an operation on a collection, the
corresponding change event includes only the delta of the fields
modified by that operation. To see the full document before or after a
change, create a ChangeStreamOptions
object and specify the
FullDocumentBeforeChange
or the FullDocument
options. Then, pass the
ChangeStreamOptions
object to the Watch()
or WatchAsync()
method.
The pre-image is the full version of a document before a change. To include the
pre-image in the change stream event, set the FullDocumentBeforeChange
option to one of the following values:
ChangeStreamFullDocumentBeforeChangeOption.WhenAvailable
: The change event includes a pre-image of the modified document for change events only if the pre-image is available.ChangeStreamFullDocumentBeforeChangeOption.Required
: The change event includes a pre-image of the modified document for change events. If the pre-image is not available, the driver raises an error.
The post-image is the full version of a document after a change. To include the
post-image in the change stream event, set the FullDocument
option to
one of the following values:
ChangeStreamFullDocumentOption.UpdateLookup
: The change event includes a copy of the entire changed document from some time after the change.ChangeStreamFullDocumentOption.WhenAvailable
: The change event includes a post-image of the modified document for change events only if the post-image is available.ChangeStreamFullDocumentOption.Required
: The change event includes a post-image of the modified document for change events. If the post-image is not available, the driver raises an error.
The following example opens a change stream on a collection and includes the post-image
of updated documents by specifying the FullDocument
option. Select the
Asynchronous or Synchronous tab to see the corresponding
code.
var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>() .Match(change => change.OperationType == ChangeStreamOperationType.Update); var options = new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup, }; using var cursor = await _restaurantsCollection.WatchAsync(pipeline, options); await cursor.ForEachAsync(change => { Console.WriteLine(change.FullDocument.ToBsonDocument()); });
var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>() .Match(change => change.OperationType == ChangeStreamOperationType.Update); var options = new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup, }; using (var cursor = _restaurantsCollection.Watch(pipeline, options)) { foreach (var change in cursor.ToEnumerable()) { Console.WriteLine(change.FullDocument.ToBsonDocument()); } }
Running the preceding code example and updating a document that has a "name"
value of "Blarney Castle"
results in the following change stream output:
{ "_id" : ObjectId("..."), "name" : "Blarney Castle", "restaurant_id" : "40366356", "cuisine" : "Traditional Irish", "address" : { "building" : "202-24", "coord" : [-73.925044200000002, 40.5595462], "street" : "Rockaway Point Boulevard", "zipcode" : "11697" }, "borough" : "Queens", "grades" : [...] }
To learn more about pre-images and post-images, see Change Streams with Document Pre- and Post-Images in the manual.
Additional Information
To learn more about change streams, see Change Streams in the manual.
API Documentation
To learn more about any of the methods or types discussed in this guide, see the following API documentation: