Unable to get MongoDB Change stream documents to appear in a watcher service

user_9879577 0
Viewed 716

I have a service that needs to watch a collection on a Mongo DB to create changes in the system. I have managed to establish a connection to a replica set using the C# driver and I'm using the following code to test the change stream.

public async Task WatchLoopAsync()
{
    var options = new ChangeStreamOptions
    {
        FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
    };
            
    using (var cursor = await _collection.WatchAsync(options))
    {
        _logger.LogInformation("Watching collection {String}", 
            _deployments.CollectionNamespace);
                
        await cursor.ForEachAsync(changeStreamDocument =>
        {
            var document = changeStreamDocument.FullDocument;
            _logger.LogInformation("Received document: {String}", 
                document.ToString());
        });
    }
}

The first log appears stating that it is watching the collection with the correct namespace. I then add a document to the collection expecting to see something log as "Received document: ..." but nothing logs.

I followed the async pattern given in the documentation here.

changestream mongodb c#
2 Answers

give the following a try:

using (var cursor = await _collection.WatchAsync(options))
{
    _logger.LogInformation("Watching collection {String}", _collection.CollectionNamespace);

    while (await cursor.MoveNextAsync())
    {
        foreach (var csd in cursor.Current)
        {
            switch (csd.OperationType)
            {
                case ChangeStreamOperationType.Insert:
                case ChangeStreamOperationType.Update:
                case ChangeStreamOperationType.Replace:
                    var document = csd.FullDocument;
                    _logger.LogInformation("Modified document: {String}", document.ToString());
                    break;
                case ChangeStreamOperationType.Delete:
                    var id = csd.DocumentKey["_id"].ToString();
                    _logger.LogInformation("Deleted document: {String}", id);
                    break;
                case ChangeStreamOperationType.Invalidate:
                    _logger.LogInformation("collection dropeed or renamed")
                    break;
            }
        }
    }

if you'd like a simpler implementation for changestreams, have a look at this

user_43684850

The solution to my particular problem was that the change stream receiving code was throwing an exception when it couldn't match a property "name" to an object property of "Name" due to the property casing conventions in C#.

To fix this, I used the following code snippet:

var conventionPack = new ConventionPack
{
    new CamelCaseElementNameConvention()
};
ConventionRegistry.Register("camelCase", conventionPack, t => true);

This registers the DB driver to use the camel casing conventions when mapping entities to BSON.

user_98795770
Related