Watching for changes to specific fields in change stream

[shireefadel]

I tried too many times to listen for changes on a certain fields, which could be achieved directly with MongoDb drivers, but it didn't work for me

Steps to Reproduce

1. Create any document inside with field name test
2. Create a change stream with match stage like the following _$match": { "updateDescription::updatedFields::test": { "ـ$exists": true } }
3. Using any tool like firecamp start listen for changes on that document
4. Send PATCH request with the following body { anotherfield: "NoNo" }
5. a message will be sent to the client and the filteration will fail

Possible Implementation

[shireefadel]

I think you have to add this option "fullDocument": "updateLookup", or at least allow it to be passed in the stream definition.

[ujibang]

Hi @shireefadel "fullDocument": "updateLookup" was added to v4.1.11

Have you tried with latest version?

[shireefadel]

Yes, I've upgraded and I saw that the fullDocument is already included, but filtering with updateDescription::updatedFields is not working for me

[ujibang]

Hi @shireefadel

I tried to reproduce your issue following the steps you gently provided...but it is working for me!

what I suspect is that your changed your stream definition and tried to connect right after. The collection metadata (where the definition is stored) is cached for performance purpose, with default value of 60 seconds. This is controlled by the following configuration options in restheart-platform-core.yml.

local-cache-enabled: true
# TTL in milliseconds; specify a value < 0 to never expire cached entries
local-cache-ttl: 60000

So either you wait a minute after updating the stream definition or disable the caching for testing purposes.

This is what I have done:

1 created two change streams on collection /cs as follows:

• /cs/_streams/cs with a condition on fullDocument
• /cs/_streams/up with a condition on updateDescription

"streams": [
        {
            "stages": [
                {
                    "_$match": {
                        "fullDocument::name": "uji"
                    }
                }
            ],
            "uri": "cs"
        },
        {
            "stages": [
                {
                    "_$match": {
                        "updateDescription::updatedFields::a": {
                            "_$exists": true
                        }
                    }
                }
            ],
            "uri": "ud"
        }
    ]

2 connected with Firecamp to the two streams

3 created a document in /cs with name=uji

POST /cs { "name": "uji" }

HTTP/1.1 201 Created 
Location: http://localhost:8080/cs2/5e106e553784af087fb43bf2

Firecamp web socket on /cs/_streams/cs received the message, the other one no. CORRECT

4 patched document /cs2/5e106e553784af087fb43bf2 adding the property a (the one in the updateDescription stage

the two web socket connections both received the message. CORRECT

This is the message received from /cs/_streams/ud

  "fullDocument": {
    "_id": {
      "$oid": "5e106e553784af087fb43bf2"
    },
    "name": "uji",
    "_etag": {
      "$oid": "5e106f2852b4b878885d3ff7"
    },
    "a": 1
  },
  "documentKey": {
    "_id": {
      "$oid": "5e106e553784af087fb43bf2"
    }
  },
  "updateDescription": {
    "updatedFields": {
      "_etag": {
        "$oid": "5e106f2852b4b878885d3ff7"
      },
      "a": 1
    },
    "removedFields": []
  },
  "operationType": "update"
}

5 patch the document adding a different property

only the change stream /cs/_streams/cs received the message. CORRECT

[ujibang]

closing this, feel free to reopen if your problem persists.

[shireefadel]

Try this change stream definition with two stages

"streams": [{
        "stages": [{
                "_$match": {
                    "fullDocument::datasourceId": {
                        "_$var": "n"
                    }
                }
            },
            {
                "_$match": {
                    "_$and": [{
                            "updateDescription::updatedFields::test": {
                                "_$exists": true
                            }
                        },
                        {
                            "operationType": "update"
                        }
                    ]
                }
            }
        ],
        "uri": "listen"
    },
    {
        "stages": [],
        "uri": "unfilteredstream"
    }
]

it fired with a PATCH request sent with anther property Incorrect

[shireefadel]

I tried to put all conditions in one stage with and operator, but it didn't work either

{

"streams": [{
        "stages": [
            {
                "_$match": {
                    "_$and": [ {
                    "fullDocument::datasourceId": {
                        "_$var": "n"
                    }},{
                            "updateDescription::updatedFields::test": {
                                "_$exists": true
                            }
                        },
                        {
                            "operationType": "update"
                        }
                    ]
                }
            }
        ],
        "uri": "listen"
    },
    {
        "stages": [],
        "uri": "unfilteredstream"
    }
]

}

I think the issue is in the $and operator

[shireefadel]

@ujibang did you check that ?

[ujibang]

Hi @shireefadel

we are creating test cases to cover your use cases. This of course will requires few days of work.

Meanwhile I suggest you to restart RESTHeart after you modify a change stream. Not only the metadata cache, but also a connected client will keep using the old stream definition.

[shireefadel]

restarting RESTHeart and the clients solved the issue, I hope we depend less on restarting RESTHeart for any reason