Filtering change stream

MongoDB Atlas App Services & Realm
,
1

I’m trying to understand how to filter the results from the change stream. Currently, each insert, update, or delete from the collection is sent to my app:

for await (const change of collection.watch()) {
    console.log(change);
}

Based on the docs, adding $match in the aggregation pipeline should allow me to pick a specific document to watch and ignore every other documents. However, this doesn’t seem to work as any updates done in the collection (whether it is on the target document, or any other document) is still being sent to my app. Can anyone help me with this? Here’s the code:

const pipeline = [ { $match: { "name": "Stella" } } ];
for await (const change of collection.watch(pipeline)) {
   console.log(change);
}

Here’s the sample document schema:

{
   _id: <some-id>,
   name: <some-name>,
   dependents: [
      {
         name: <some-name>,
         relation: <some-relation>,
      },
   ],
   date_time_created: <some-timestamp>,
}
2

Hi. I would recommend reading through the documentation on Change Events. The TLDR though is that MongoDB returns Change Events and if you want to filter the events returned by watch() you need to “filter” on the “change events”. This has fields like FullDocument, UpdateDescription, etc depending on the event type (update, insert, delete, replace)

Let me know if you have any specific questions but the gist is that you need to update your match expression to filter on these events

4

Sorry but this does not even closely close to a good answer. the Change Events page you linked does not contains information about how to correctly use pipelines. why should i filter myself based on the fields themself such ‘FullDocument’,etc. based on your answer i understand that ‘pipelines’ option for watch() does not work as a pipeline, so for what they exists?

5

Hi,

What is it exactly that you are trying to accomplish? If you explain what you would like to do, I would be more than happy to help give specific examples, but the documentation on Change Events and Match Expressions should be the proper place to start. The Watch() API takes a filter that is a $match on the series of Change Events that are sent back. Therefore, if you wanted to filter on all events in which name is “stella”, you would add a match expression of `{“fullDocument.name”: “stella” } and ensure that your trigger it change stream is using the FullDocument option for updates.

Thanks,
Tyler

6

Hate to reply to a year old thread, but I am attempting to do something similar using the mongocxx driver. I have been unable to find any examples of using a pipeline within a changestream, so I’m wondering if it is possible. I have attempted to setup the match in the following manner, but I continue to receive all changes, as if no filter was applied.

matchPipe.match(bsoncxx::builder::basic::make_document(bsoncxx::builder::basic::kvp(“fullDocument._id”, “id-to-search-for”)));

Any help you could provide would be greatly appreciatd.