Monitor Data Changes

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 Get Started with the .NET/C# Driver.

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; }
    [BsonElement("restaurant_id")]
    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; }
    [BsonElement("coord")]
    public double[] Coordinates { get; set; }
    public string Street { get; set; }
    [BsonElement("zipcode")]
    public string ZipCode { get; set; }
}

public class GradeEntry
{
    public DateTime Date { get; set; }
    public string Grade { get; set; }
    public float? Score { get; set; }
}

Monitor Update Events Example

The following example uses the pipeline parameter to open a change stream that records only update operations. Select the Synchronous or Asynchronous tab to see the corresponding code.

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 = collection.Watch(pipeline))
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine("Received the following change: " + change);
    }
}

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 collection.WatchAsync(pipeline))
{
    await cursor.ForEachAsync(change =>
    {
        Console.WriteLine("Received the following change: " + change);
    });
}

Split Large Change Events Example

If your application generates change events that exceed 16 MB in size, the server returns a BSONObjectTooLarge error. To avoid this error, you can use the $changeStreamSplitLargeEvent pipeline stage to split the events into smaller fragments. The .NET/C# Driver aggregation API includes the ChangeStreamSplitLargeEvent() method, which you can use to add the $changeStreamSplitLargeEvent stage to the change stream pipeline.

This example instructs the driver to watch for changes and split change events that exceed the 16 MB limit. The code prints the change document for each event and calls helper methods to reassemble any event fragments:

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .ChangeStreamSplitLargeEvent();
using var cursor = collection.Watch(pipeline);
foreach (var completeEvent in GetNextChangeStreamEvent(cursor.ToEnumerable().GetEnumerator()))
{
    Console.WriteLine("Received the following change: " + completeEvent.BackingDocument);
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .ChangeStreamSplitLargeEvent();
using var cursor = await collection.WatchAsync(pipeline);
await foreach (var completeEvent in GetNextChangeStreamEventAsync(cursor))
{
    Console.WriteLine("Received the following change: " + completeEvent.BackingDocument);
}

The preceding example uses the GetNextChangeStreamEvent(), GetNextChangeStreamEventAsync(), and MergeFragment() methods to reassemble change event fragments into a single change stream document. The following code defines these methods:

// Fetches the next complete change stream event
private static IEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEvent<TDocument>(
IEnumerator<ChangeStreamDocument<TDocument>> changeStreamEnumerator)
{
    while (changeStreamEnumerator.MoveNext())
    {
        var changeStreamEvent = changeStreamEnumerator.Current;
        if (changeStreamEvent.SplitEvent != null)
        {
            var fragment = changeStreamEvent;
            while (fragment.SplitEvent.Fragment < fragment.SplitEvent.Of)
            {
                changeStreamEnumerator.MoveNext();
                fragment = changeStreamEnumerator.Current;
                MergeFragment(changeStreamEvent, fragment);
            }
        }
        yield return changeStreamEvent;
    }
}
// Merges a fragment into the base event
private static void MergeFragment<TDocument>(
    ChangeStreamDocument<TDocument> changeStreamEvent,
    ChangeStreamDocument<TDocument> fragment)
{
    foreach (var element in fragment.BackingDocument)
    {
        if (element.Name != "_id" && element.Name != "splitEvent")
        {
            changeStreamEvent.BackingDocument[element.Name] = element.Value; 
        }
    }
}

// Fetches the next complete change stream event
private static async IAsyncEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEventAsync<TDocument>(
    IAsyncCursor<ChangeStreamDocument<TDocument>> changeStreamCursor)
{
    var changeStreamEnumerator = GetNextChangeStreamEventFragmentAsync(changeStreamCursor).GetAsyncEnumerator();
    while (await changeStreamEnumerator.MoveNextAsync())
    {
        var changeStreamEvent = changeStreamEnumerator.Current;
        if (changeStreamEvent.SplitEvent != null)
        {
            var fragment = changeStreamEvent;
            while (fragment.SplitEvent.Fragment < fragment.SplitEvent.Of)
            {
                await changeStreamEnumerator.MoveNextAsync();
                fragment = changeStreamEnumerator.Current;
                MergeFragment(changeStreamEvent, fragment);
            }
        }
        yield return changeStreamEvent;
    }
}
private static async IAsyncEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEventFragmentAsync<TDocument>(
    IAsyncCursor<ChangeStreamDocument<TDocument>> changeStreamCursor)
{
    while (await changeStreamCursor.MoveNextAsync())
    {
        foreach (var changeStreamEvent in changeStreamCursor.Current)
        {
            yield return changeStreamEvent;
        }
    }
}
// Merges a fragment into the base event
private static void MergeFragment<TDocument>(
    ChangeStreamDocument<TDocument> changeStreamEvent,
    ChangeStreamDocument<TDocument> fragment)
{
    foreach (var element in fragment.BackingDocument)
    {
        if (element.Name != "_id" && element.Name != "splitEvent")
        {
            changeStreamEvent.BackingDocument[element.Name] = element.Value;
        }
    }
}

Collation

To configure collation for your operation, create an instance of the Collation class.

The following table describes the parameters that the Collation constructor accepts. It also lists the corresponding class property that you can use to read each setting's value.

For more information about collation, see the Collation page in the MongoDB Server manual.

Include Pre-Images and Post-Images

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 Synchronous or Asynchronous 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 = collection.Watch(pipeline, options))
{
    foreach (var change in cursor.ToEnumerable())
    {
        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 = await collection.WatchAsync(pipeline, options);
await cursor.ForEachAsync(change =>
{
    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 MongoDB Server manual.

API Documentation

To learn more about any of the methods or types discussed in this guide, see the following API documentation:

• Watch()

• WatchAsync()

• ChangeStreamOptions

• UpdateOne()