Interface IObservableQueryDemultiplexer

Namespace: Cratis.Arc.Queries

Assembly: Cratis.Arc.Core.dll

Defines the observable query demultiplexer — a composite endpoint that accepts multiplexed observable query subscriptions over a single persistent connection and routes them to individual query pipelines.

public interface IObservableQueryDemultiplexer

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

Three fixed endpoints are provided under /.cratis/queries:

/.cratis/queries/ws — WebSocket; bidirectional subscribe/unsubscribe.
/.cratis/queries/sse — Server-Sent Events; multiplexed stream identified by a server-assigned connection ID.
/.cratis/queries/sse/subscribe and /.cratis/queries/sse/unsubscribe — POST endpoints for managing SSE subscriptions.

Both transports honor the query authorization pipeline; unauthorized subscriptions receive an Unauthorized response rather than data.

Methods

HandleSSEConnection(IHttpRequestContext)

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

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.

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.

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.

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.