In Golang Driver for Mongo DB "Collection.Watch" doesn't work with Aggregation Pipeline

user_2476944 0
Viewed 273

I am trying to create a mongodb "Change Stream" on collection and listen to changes on a specific document.

matchID := bson.D{
        {"$match", bson.M{"_id": tid}},
    }
stream, err := collection.Watch(ctx, mongo.Pipeline{matchID}, options.ChangeStream().SetFullDocument(options.UpdateLookup))

Then listen on the stream with a for loop.

for stream.Next(ctx) {
        log.Println("stream.Next")
        if err = stream.Decode(&event); err != nil {
            log.Printf("error decoding: %s", err)
        }
        log.Printf("change event: %v", event)
        publisher.NotifyEvent(event["fullDocument"])
    }

The whole setup works when I remove the pipeline argument, and I start getting data in loop. But when I add the pipeline filter, it stops working.

go mongodb
1 Answers

The filter I used was wrong, since the pipeline filters the event stream rather that the actual document. I had the change the filter to

 matchID := bson.D{
        {"$match", bson.M{"fullDocument._id": tid}},
    }

instead of

matchID := bson.D{
            {"$match", bson.M{"_id": tid}},
        }

the additional keyword "fullDocument" is necessary because the event stream holds the updated document under the field "fullDocument"

More details can be found in this article. https://www.mongodb.com/basics/change-streams

user_24769440
Related