v9 (current) v8
Edit Page

Change streams

RESTHeart Cloud
Prefer not to run it yourself? Ulabase is this, hosted, with sign-up, payments and an MCP server already on. Free to start. Get started →
Note
 Change streams can be consumed via WebSocket or SSE. This page covers the stream definition format. For the SSE transport see SSE API Overview.

The "streams" collection metadata

In RESTHeart, not only documents but also databases and collections have properties. Some properties are metadata, i.e. they have a special meaning for RESTheart that influences its behavior.

The collection metadata property streams allows to declare change streams that client can watch for to be aware about changes to documents and bind them to given URI.

Change streams need to be defined as collection metadata. This way clients are not able to open up arbitrary change streams but only those defined (and tested) by the developers.

streams is an array of stream definitions.

Stream definition format

{ "streams": [
    {
        "uri": <uri>,
        "stages": [
            "<stage_1>",
            "<stage_2>",
            ...
        ]
    }
]}
Property Description Mandatory

uri

specifies the URI when the stream is bound under the path /<db>/<collection>/_streams

yes

stages

the MongoDB aggregation pipeline stages. For more information refer to Aggregation Pipeline on MongoDb documentation.

yes
Note
 Stages take as input Change Events instead of the documents. For example, the modified version of a document after a PATCH request is present at event.fullDocument property of the stages input event. (See examples).

Optional Stages

The aggregation of the stream can include optional stages. These optional stages are only executed when one or more variables are specified.

For a more comprehensive understanding of how to use optional stages, please refer to the Aggregation documentation.

Note
 An optional stage is a variable in the stages, with the cost described under notify_when: one MongoDB cursor per distinct binding. For a selection that differs per client, prefer notify_when.

Escaped stage properties

MongoDB (up to v5) does not allow to store fields with names starting with $ or containing dots (.).

To store stages with dollar prefixed operators or use the dot notation, RESTHeart automatically and transparently escapes the properties keys as follows:

  • the $ prefix is "underscore escaped", e.g. $exists is stored as _$exists

  • if the dot notation has to be used in a key name, dots are replaced with :: e.g. SD.prop is stored as SD::prop

Per-client filtering: notify_when

Available from RESTHeart 9.4.

notify_when is an optional field in a stream definition that filters which connected clients receive each change event, without requiring a separate MongoDB cursor per client.

Every client of the stream shares one cursor, whatever value it binds. The alternative, a { "$var": …​ } in a stage, changes the pipeline, and RESTHeart opens one cursor per distinct set of bound variables: a thousand users filtering in the stages are a thousand change streams on MongoDB. Keep the stages fixed and put the client-bound variables here.

{
  "streams": [{
    "uri": "by-tenant",
    "stages": [
      { "_$match": { "operationType": { "_$in": ["insert", "update"] } } }
    ],
    "notify_when": {
      "fullDocument::tenantId": { "$var": "tid" }
    }
  }]
}

The client passes its variable value as a query parameter when connecting:

# SSE — only receives events where fullDocument.tenantId == "acme"
curl -N \
  -H 'Accept: text/event-stream' \
  -H 'Authorization: Basic YWRtaW46c2VjcmV0' \
  'http://localhost:8080/mydb/mycoll/_streams/by-tenant?tid=acme'

The variable is bound like any other stream variable (see Using Variables): ?tid=acme and ?avars={"tid":"acme"} are equivalent, and when both are given the avars value wins. Before 9.12.2 notify_when read only the plain query parameter.

Field path notation

The notify_when key uses :: to separate the top-level change event field from the dotted path within it:

"fullDocument::tenantId"       # fullDocument.tenantId
"fullDocument::org.department" # fullDocument.org.department (nested)
"documentKey::_id"             # documentKey._id

Supported predicates

Form Description Example

Scalar equality

Field value (string) equals the bound variable

{ "fullDocument::tenantId": { "$var": "tid" } }

Array membership

Bound variable value is contained in a string array field

{ "fullDocument::recipients": { "$var": "userId" } }

Design notes

  • stages are sent to MongoDB and applied once on the shared cursor. Keep them fixed (e.g. operationType, fullDocument.status): a variable there means one cursor per distinct binding.

  • notify_when is evaluated server-side per event × per connected client. Use it for conditions that depend on per-client variables.

  • If notify_when is absent, all clients receive all events (broadcast — existing behaviour).

  • If a client connects without binding the variable, neither as a query parameter nor in avars, the request fails with 400 and variable not bound, as for an unbound stage variable. Before 9.12.2 such a client received every event of the stream (pass-through).

Warning

notify_when is a delivery filter, not an access control. The variable it binds comes from the client’s own query string, and a client can bind any value it can name and receive the events of that value. Never use it to keep tenants apart.

What does keep them apart, since 9.8.2, is the caller’s permission: a readFilter is applied to the stream exactly as it is to the equivalent GET, and so is projectResponse. Scope the data in the ACL, and use notify_when on top of it to save a client the events it does not care about.

A stage can also scope a stream with a server-resolved variable, { "$match": { "fullDocument::owner": { "$var": "@user.sub" } } }, and so does a readFilter that names the caller: both become stages of the pipeline, and both cost one cursor per caller. That is the price of a boundary. notify_when is the selection that costs none, on top of it. See Predefined variables.

Last-Event-ID limitation

When notify_when is defined, Last-Event-ID is ignored on reconnect and a warning is logged. The shared cursor cannot be rewound for a single client; reconnecting clients should recover missed events via the REST API.

Keeping a WebSocket alive

The server sends a ping on an idle stream every change-streams-keep-alive-ms (20 seconds by default), and since 9.9.2 answers the client’s pings with a pong. Anything else the client sends is ignored.

A connection can die without a close reaching the client, for instance while its computer sleeps: the socket stays open and no event ever arrives. A client that wants to know sends pings and reconnects when the pong does not come back. With websocat:

websocat --ping-interval 20 --ping-timeout 60 autoreconnect:wss://example.com/coll/_streams/all -

Recovering missed events

A WebSocket delivers events only while connected. After a reconnection, ask the REST API for what changed since the last event seen: every write through RESTHeart gives the document a new _etag, an ObjectId that grows with time, so

GET /coll?filter={"_etag":{"$gt":{"$oid":"<last _etag seen>"}}}&sort={"_etag":1}

returns the documents inserted or updated since then, in order. Deleted documents do not show up.

See SSE API Overview for details on the SSE transport and event format.