Observable Query Demultiplexer

The Observable Query Demultiplexer is a composite real-time streaming endpoint that multiplexes multiple observable query subscriptions over a single persistent connection.

Rather than creating one WebSocket or SSE connection per query, the demultiplexer provides two fixed, well-known endpoints — one for WebSocket and one for Server-Sent Events — that all observable queries in an application route through.

Endpoints

Transport	Endpoint	Direction
WebSocket	`/.cratis/queries/ws`	Bidirectional — subscribe, unsubscribe, ping/pong
Server-Sent Events (SSE)	`/.cratis/queries/sse`	Server → client; one query per connection

Both endpoints are registered automatically by UseCratisArc() and require no manual configuration.

Why a Composite Demultiplexer?

Individual per-query WebSocket endpoints work but come with drawbacks at scale:

Each browser tab opens a separate WebSocket per observable query.
HTTP/1.1 limits the number of concurrent connections per origin.
SSE has the same constraint and typically falls back to polling when the connection limit is reached.

The demultiplexer solves this by allowing a single WebSocket to carry updates for many queries simultaneously, and by providing a single, predictable SSE endpoint that clients can connect to for any query.

Protocol

All messages exchanged over the demultiplexer share a common envelope:

{
  "type": "<ObservableQueryHubMessageType>",
  "queryId": "<client-assigned-id>",
  "payload": { ... },
  "timestamp": 1234567890
}

Field	Description
`type`	One of the message types listed below.
`queryId`	Client-assigned identifier that correlates subscriptions with their result updates. Must be unique per subscription within a connection.
`payload`	Depends on `type` — see the table below.
`timestamp`	Unix milliseconds. Only populated for `ping` / `pong`.

Message Types

Type	Direction	Payload
`subscribe` (0)	Client → Server	`ObservableQuerySubscriptionRequest`
`unsubscribe` (1)	Client → Server	(none)
`queryResult` (2)	Server → Client	`QueryResult`
`unauthorized` (3)	Server → Client	(none)
`error` (4)	Server → Client	Error message string
`ping` (5)	Client → Server	(timestamp only)
`pong` (6)	Server → Client	(timestamp echoed from ping)

Subscribe Payload — `ObservableQuerySubscriptionRequest`

{
  "queryName": "MyApp.Authors.Listing.AllAuthors",
  "arguments": { "filter": "active" },
  "page": 0,
  "pageSize": 25,
  "sortBy": "name",
  "sortDirection": "asc"
}

Field	Required	Description
`queryName`	✅	Fully qualified name of the observable query method (e.g. `MyApp.Features.Authors.Listing.AllAuthors`).
`arguments`	☐	Query-string arguments forwarded to the query performer.
`page`	☐	Zero-based page index for paged queries.
`pageSize`	☐	Number of items per page.
`sortBy`	☐	Field name to sort by (case-insensitive).
`sortDirection`	☐	`asc` or `desc`.

WebSocket Transport

Connect to /.cratis/queries/ws and send subscribe messages to start receiving updates.

{
  "type": 0,
  "queryId": "authors-list",
  "payload": {
    "queryName": "MyApp.Authors.Listing.AllAuthors"
  }
}

The server responds with one or more queryResult messages whenever the underlying data changes:

{
  "type": 2,
  "queryId": "authors-list",
  "payload": {
    "isSuccess": true,
    "isAuthorized": true,
    "data": [ ... ],
    "validationResults": []
  }
}

Unsubscribe

{
  "type": 1,
  "queryId": "authors-list"
}

Keep-alive (Ping / Pong)

Send a ping with the current Unix timestamp; the server echoes it as a pong with the same timestamp for round-trip latency measurement.

{ "type": 5, "timestamp": 1740000000000 }
// Server responds:
{ "type": 6, "timestamp": 1740000000000 }

Unauthorized

If the current user is not authorized to access the requested query, the server sends an unauthorized message and no data stream is established:

{
  "type": 3,
  "queryId": "authors-list"
}

SSE Transport

Connect to /.cratis/queries/sse using the EventSource API. Pass the fully qualified query name in the query query-string parameter. All other query-string parameters are forwarded as query arguments.

GET /.cratis/queries/sse?query=MyApp.Authors.Listing.AllAuthors&filter=active

The server responds with the standard SSE content type (text/event-stream) and streams data: frames containing serialized ObservableQueryHubMessage envelopes whenever the underlying data changes.

Each SSE connection carries a single query subscription. To observe multiple queries simultaneously via SSE, open multiple EventSource connections — or use the WebSocket transport which multiplexes all subscriptions over one connection.

Paging and Sorting via SSE

Pass paging and sorting directly as query-string parameters:

GET /.cratis/queries/sse?query=MyApp.Authors.Listing.AllAuthors&page=0&pageSize=20&sortBy=name&sortDirection=asc

Authorization

Authorization is enforced for every subscription through the standard query pipeline, including all registered IQueryFilter implementations.

If the query performer has an [Authorize] attribute and the current user is not authenticated or lacks the required role, the subscription is rejected with an unauthorized message.
If the query allows anonymous access ([AllowAnonymous]), the subscription is accepted regardless of authentication state.
Authorization is re-evaluated on every new subscription, not cached for the lifetime of the connection.

Keep-alive

Both WebSocket and SSE transports send automatic keep-alive messages to prevent idle connections from being closed by proxies or firewalls.

The server sends a ping message only when no other message has been sent within the configured interval. If data is flowing normally (frequent queryResult messages), the keep-alive is suppressed — it fires only during periods of inactivity.

Configuration

Configure the keep-alive interval in ArcOptions.Query:

builder.Services.Configure<ArcOptions>(options =>
{
    options.Query.KeepAliveInterval = TimeSpan.FromSeconds(30); // default
});

Or inline when calling AddCratisArc():

builder.AddCratisArc(options =>
{
    options.Query.KeepAliveInterval = TimeSpan.FromSeconds(45);
});

Set KeepAliveInterval to TimeSpan.Zero or a negative value to disable keep-alive entirely.

Property	Type	Default	Description
`Query.KeepAliveInterval`	`TimeSpan`	30 seconds	How often to send a keep-alive ping when no data is flowing.