Class ObservableQueryDemultiplexer

Namespace: Cratis.Arc.Queries

Assembly: Cratis.Arc.Core.dll

Represents an implementation of IObservableQueryDemultiplexer providing composite observable query streaming over a fixed WebSocket endpoint (/.cratis/queries/ws) and a fixed Server-Sent Events endpoint (/.cratis/queries/sse).

[Singleton]
public class ObservableQueryDemultiplexer : IObservableQueryDemultiplexer

Inheritance: object

ObservableQueryDemultiplexer

Implements: IObservableQueryDemultiplexer

Inherited Members: object.GetType()

object.MemberwiseClone()

object.ToString()

object.Equals(object)

object.Equals(object, object)

object.ReferenceEquals(object, object)

object.GetHashCode()

Extension Methods: MethodCalls.CallGenericMethod<TOut, T>(T, Expression<Func<T, Func<TOut>>>, params Type[])

MethodCalls.CallGenericMethod<TOut, T, T1>(T, Expression<Func<T, Func<T1, TOut>>>, T1, params Type[])

MethodCalls.CallGenericMethod<TOut, T, T1, T2>(T, Expression<Func<T, Func<T1, T2, TOut>>>, T1, T2, params Type[])

MethodCalls.CallGenericMethod<TOut, T, T1, T2, T3>(T, Expression<Func<T, Func<T1, T2, T3, TOut>>>, T1, T2, T3, params Type[])

ConverterExtensions.ConvertTo(object, Type)

ConceptExtensions.GetConceptValue(object)

ConceptExtensions.IsConcept(object)

JsonValueExtensions.ToJsonValue(object?)

ObservableExtensions.InvokeAndWrapWithSubject<TResult>(object, Func<CancellationToken, IObservable<TResult>>)

ObservableExtensions.InvokeAndWrapWithTransformSubject<TResult, TSource>(object, Func<CancellationToken, IObservable<TSource>>, Func<TSource, TResult>)

Remarks

Authorization is honored for every subscription through the query pipeline filters. If the current user is not authorized to perform a query, an Unauthorized message is sent instead of data.

Both transports support multiple concurrent subscriptions over a single connection. The WebSocket transport uses bidirectional frames; the SSE transport uses separate POST endpoints for subscribe/unsubscribe, correlated via a server-assigned connection identifier.

Because the SSE transport spreads one connection across several HTTP requests, the caller that opened the stream is captured as an Cratis.Arc.Queries.ObservableQueryCaller and every subscribe and unsubscribe is checked against it. A control request from a different caller is answered with 404, identically to one naming a connection that does not exist, so a connection identifier on its own grants nothing.

Each subject-backed subscription gets its own IServiceScope, created at subscribe time from serviceProvider and disposed when the subscription ends. Chronicle's client services are registered scoped, resolving the current tenant's namespace the first time they are used within a scope and caching it for the scope's lifetime — resolving them from the root container instead would cache whichever tenant asked first and reuse it for every subscription thereafter. Emissions arrive on the subject's own producer thread, where the AsyncLocal tenant context does not flow, so Current is restored to the subscribing connection immediately before each interception call.

The subscription-time authorization verdict gates obtaining the stream, not the stream itself. An application that needs the verdict re-checked while a subscription is running implements an IGuardObservableQueryEmission; it is then consulted for every emission, and can withhold a single emission or terminate the subscription. With no guard implemented nothing is dispatched and emissions take exactly the path they always did.

Constructors

ObservableQueryDemultiplexer(IQueryPipeline, IQueryContextManager, IHttpRequestContextAccessor, IHostApplicationLifetime, IReadModelInterceptors, IServiceProvider, IOptions<ArcOptions>, IQueryHealthTracker, IObservableQueryEmissionGuards, ILogger<ObservableQueryDemultiplexer>)

public ObservableQueryDemultiplexer(IQueryPipeline queryPipeline, IQueryContextManager queryContextManager, IHttpRequestContextAccessor httpRequestContextAccessor, IHostApplicationLifetime hostApplicationLifetime, IReadModelInterceptors readModelInterceptors, IServiceProvider serviceProvider, IOptions<ArcOptions> arcOptions, IQueryHealthTracker healthTracker, IObservableQueryEmissionGuards emissionGuards, ILogger<ObservableQueryDemultiplexer> logger)

Parameters

queryPipeline IQueryPipeline: The IQueryPipeline used to perform and authorize queries.
queryContextManager IQueryContextManager: The IQueryContextManager for managing query contexts.
httpRequestContextAccessor IHttpRequestContextAccessor: The IHttpRequestContextAccessor used to propagate the caller's identity into the authorization pipeline, and restored around each emission so tenant resolution sees the subscribing connection.
hostApplicationLifetime IHostApplicationLifetime: The IHostApplicationLifetime used to cancel connections on shutdown.
readModelInterceptors IReadModelInterceptors: The IReadModelInterceptors used to intercept (e.g. decrypt compliance/PII properties on) each emitted read model before it is sent to the client.
serviceProvider IServiceProvider: The IServiceProvider used to create a per-subscription IServiceScope for resolving interceptors — see remarks.
arcOptions IOptions<ArcOptions>: The ArcOptions used for JSON serialization.
healthTracker IQueryHealthTracker: The IQueryHealthTracker used to track subscription health.
emissionGuards IObservableQueryEmissionGuards: The IObservableQueryEmissionGuards consulted per emission when an application opts in with an IGuardObservableQueryEmission.
logger ILogger<ObservableQueryDemultiplexer>: The logger.

Remarks

Authorization is honored for every subscription through the query pipeline filters. If the current user is not authorized to perform a query, an Unauthorized message is sent instead of data.

Methods

HandleSSEConnection(IHttpRequestContext)

Handles an incoming SSE connection on the /.cratis/queries/sse endpoint.

public Task HandleSSEConnection(IHttpRequestContext context)

Parameters

context IHttpRequestContext: The IHttpRequestContext for the request.

Returns

Task: A Task that completes when the client disconnects.

Remarks

Establishes a persistent SSE stream. The server sends a Connected message containing the connection identifier. The client then uses the POST endpoints to subscribe/unsubscribe using this identifier. The stream carries all subscription results and keep-alive pings.

HandleSSESubscribe(IHttpRequestContext)

Handles a POST request to subscribe to a query over an established SSE connection.

public Task HandleSSESubscribe(IHttpRequestContext context)

Parameters

context IHttpRequestContext: The IHttpRequestContext for the request.

Returns

Task: A Task representing the asynchronous operation.

HandleSSEUnsubscribe(IHttpRequestContext)

Handles a POST request to unsubscribe from a query over an established SSE connection.

public Task HandleSSEUnsubscribe(IHttpRequestContext context)

Parameters

context IHttpRequestContext: The IHttpRequestContext for the request.

Returns

Task: A Task representing the asynchronous operation.

HandleWebSocketConnection(IHttpRequestContext)

Handles an incoming WebSocket connection on the /.cratis/queries/ws endpoint.

public Task HandleWebSocketConnection(IHttpRequestContext context)

Parameters

context IHttpRequestContext: The IHttpRequestContext for the request.

Returns

Task: A Task that completes when the WebSocket connection closes.

Remarks

The method reads ObservableQueryHubMessage frames from the socket. Clients subscribe to individual queries by sending Subscribe messages and can unsubscribe at any time using Unsubscribe. Ping/pong keep-alive frames are also supported.