Subscriptions — push events to every open page

A component subscribes to an event the way it queries data: QueryClient.Subscribe<OrderPlaced>() in Render, and every OrderPlaced published afterwards — by a command handler, a background job, a domain event after a save, another server — lands in it and re-renders it. It is what tRPC calls a subscription, built on the CQRS notification Rask already has, so one record reaches both the code that reacts to it and the screens that show it.

Included in the Rask package, with CQRS and Rask.Query. In a separate front end, Rask.Cqrs.Client opens it on the server.

The whole of it

The event is an ordinary notification:

public sealed record OrderPlaced(Guid Id, string Customer, decimal Total) : INotification;

Publish it where it happens, through the dispatcher you already use:

public sealed class PlaceOrderHandler(IDispatcher dispatcher) : ICommandHandler<PlaceOrder, Guid>
{
    public async Task<Guid> HandleAsync(PlaceOrder command, CancellationToken ct)
    {
        var order = await Order.Create(command.Customer, command.Total);
        await dispatcher.PublishAsync(new OrderPlaced(order.Id, order.Customer, order.Total), ct);
        return order.Id;
    }
}

Subscribe in the component that shows it:

public sealed partial class NewOrders : Component
{
    protected override Component? Render()
    {
        var placed = QueryClient.Subscribe<OrderPlaced>().Keep(20);

        return Ul[placed.Items.Reverse().Select(o => Li.Key(o.Id)[$"{o.Customer} — {o.Total:C}"])];
    }
}

That is all. There is no subscription to dispose, no OnMount and no StateHasChanged:

• It lives as long as the component. The subscription closes when the component that read it unmounts — the visitor navigates away, the tab closes, the session ends.
• Each value re-renders the component, exactly as a query's result does when it lands.
• Publishing still runs the handlers. PublishAsync hands the notification to its INotificationHandlers and to every open subscription, so adding a screen never changes what the server does.

Try it: the buttons publish, and the two boards — which know nothing about the buttons or each other — each receive every order. The line above them watches one order and hears only its shipment.

using Rask.Cqrs;
using Rask.Query;

namespace Rask.Site.Features;

// Publishes through the dispatcher, the same call a command handler or a background job makes. It holds no reference to
// the boards below — they only subscribe to OrderPlaced — so a board anywhere else (or, on a server, in anyone else's
// open tab) would receive the same order.
public sealed partial class SubscriptionDemo(IDispatcher dispatcher) : Component
{
    private static readonly string[] Items = ["anvil", "rocket skates", "giant magnet", "earthquake pills"];

    private int _placed;

    // One order's events, and following it: the lambda runs at every read, and returns null — waiting — until an order
    // has been placed.
    private Subscription<OrderShipped> Shipped =>
        field ??= QueryClient.Subscribe<OrderShipped>(() => _placed > 0 ? new WatchOrder(_placed) : null);

    protected override Component? Render() =>
        Div.Id("subscription-demo").Class("flex flex-col gap-3")[
            Div.Class("flex gap-2 flex-wrap")[
                UiButton.Tone(UiTone.Primary).Id("subscription-place").OnClick(PlaceOrderAsync)["Place an order"],
                UiButton.Id("subscription-ship").Disabled(_placed == 0).OnClick(ShipOrderAsync)["Ship it"]
            ],
            P.Id("subscription-tracked").Class("mb-0")[Tracking()],
            Div.Class("grid gap-3 sm:grid-cols-2")[
                OrderBoard.Title("Warehouse"),
                OrderBoard.Title("Accounting")
            ]
        ];

    private string Tracking() => _placed switch
    {
        0 => "Place an order to track it.",
        _ when Shipped.Data?.Number == _placed => $"Order #{_placed} has shipped.",
        _ => $"Order #{_placed} is waiting to ship.",
    };

    private Task PlaceOrderAsync() =>
        dispatcher.PublishAsync(new OrderPlaced(++_placed, Items[(_placed - 1) % Items.Length]));

    private Task ShipOrderAsync() => dispatcher.PublishAsync(new OrderShipped(_placed));
}

Where to declare one

In the same three places as a query, with the same rules:

// in Render — the same call is the same subscription every render, re-pointed when what it watches changes
var shipped = QueryClient.Subscribe(new WatchOrder(Id));

// in a property or the constructor, from a lambda — re-run at every read; null waits
Subscription<OrderShipped> Shipped =>
    field ??= QueryClient.Subscribe<OrderShipped>(() => Selected is { } id ? new WatchOrder(id) : null);

A lambda that returns null means the input is not there yet: nothing is opened, and IsLoading is false, until it returns a record. A render that stops asking for a subscription sets it aside, and its next read opens it again. Two records that are equal are the same subscription, so building one per render costs nothing.

What a component reads

var shipped = QueryClient.Subscribe(new WatchOrder(Id));

return shipped.IsLoading ? Spinner()
     : Div[Badge[shipped.Data?.Status ?? order.Data?.Status], shipped.IsReconnecting ? Small["reconnecting…"] : null];

It starts with the last value published for what it watches, so a page opened after the fact still shows it. On a server render that value is in Data before the first paint.

An event about one thing

Most events are about one record: this order shipped, this user's export is ready. The event stays plain; what to watch is its own record — a subscription, the fourth message shape beside a query, a command and a notification:

public sealed record OrderShipped(Guid OrderId, string Status) : INotification;

public sealed record WatchOrder(Guid OrderId) : ISubscription<OrderShipped>
{
    public bool Matches(OrderShipped e) => e.OrderId == OrderId;
}

var shipped = QueryClient.Subscribe(new WatchOrder(Id));   // only this order's

Matches says which notifications are its own. It runs for each published notification of the type, on the publisher's thread, so it reads the notification and nothing else: no database, no service, no await. Anything it can ask — "orders over £100", "either of these two rooms" — is a subscription, not just an id.

Who may open it is a watch policy, and that is where the database goes:

public sealed class WatchingOrders : IWatchPolicy<WatchOrder>
{
    public async Task<bool> CanWatchAsync(WatchOrder watch, CancellationToken ct) =>
        (await Order.Find(watch.OrderId)).CustomerId == Current.UserId || Current.Principal?.IsInRole("Admin") == true;
}

Write it anywhere in the project; the generator registers it, like a handler. It is asked once, when the subscription opens, in the subscriber's own scope — so Current.UserId is the person asking and a scoped DbContext is the one a handler would get. One class may implement several policies where the rule is the same.

It fails closed. A subscription with no policy lets nobody open it, so a forgotten policy never shows one customer another's order. The subscription settles on Error with an UnauthorizedAccessException before anything arrives — in this process and from a browser alike.

A subscription record is a message like any other: it is compared structurally, and it crosses the wire as its generated JSON, so it carries whatever its properties carry.

Watching a type. An event that is about nothing in particular — every order placed, on an admin board — needs no record at all: QueryClient.Subscribe<OrderPlaced>() watches the type itself.

Patching a query on screen

.Into edits a query's cached result with each value, so the list on screen changes in place with no round trip — the same shape as an optimistic edit:

var orders = QueryClient.Query(new GetOrders(Page));

QueryClient.Subscribe<OrderShipped>()
    .Into(orders, (list, e) => [.. list.Select(o => o.Id == e.OrderId ? o with { Status = e.Status } : o)]);

Write the patch so that applying it twice changes nothing — replace by id rather than prepend — because the query's own refetch may already include the value. After a dropped connection comes back, each query a subscription patches is refetched once, so whatever was published meanwhile is not lost.

A stream that is a function

Data that does not arrive as a notification — a price feed, a third-party stream — is a function returning an IAsyncEnumerable<T>, opened for an input and reopened when a render passes a different one:

var price = QueryClient.Subscribe(Symbol, (symbol, ct) => Prices.Stream(symbol, ct));

It runs where the component runs: on the server for a Server-host page, in the browser for a WebAssembly one. When it runs out, Status is Ended; when it throws, the subscription reopens it.

Publishing from anywhere

Everything that publishes a notification reaches subscribers, because they are the same notification:

• A command handler or any code with IDispatcher — dispatcher.PublishAsync(new OrderShipped(id, "Shipped")).
• A background job — a job's handler publishes the same way, so progress and "your report is ready" reach the page that is waiting for them.
• A domain event raised by an aggregate is published after its save commits (Rask.Data), and one relayed by the outbox is published when it is relayed.

Outside a component

IDispatcher.SubscribeAsync is the same subscription as an IAsyncEnumerable<T>, for code with no component — a hosted service, a test:

await foreach (var shipped in dispatcher.SubscribeAsync(new WatchOrder(orderId), ct))
    logger.LogInformation("{Order} is {Status}", shipped.OrderId, shipped.Status);

dispatcher.SubscribeAsync<OrderPlaced>(ct) is the same thing for an event watched by type.

It starts with the last one published, asks the policy first, and ends when the token is cancelled.

In a WebAssembly front end

In a wasm-hosted app the page runs in the browser, and the events happen on the server. With AddRaskCqrsClient(), a subscription opens on the server: a long-lived GET answered with server-sent events, over the same origin and cookie as every other message. So a browser page hears what any visitor's command published, and an event the page itself publishes travels to the server and comes back once, like everyone else's.

The server's MapRaskCqrs() serves it at GET /_rask/cqrs/request/events/{name}?m={json} — under the same prefix as the messages, and the subscription record travels in ?m= exactly as a query's message does, so the same CSRF header, the same authentication and any rate limit you attach to the group apply. Who may subscribe from outside is decided on the server, and closed unless opened:

The last row is deliberate: an app's auth events and domain events are notifications too, and none of them should be one browser request away. A handler's [Authorize] still decides who may publish a notification from the browser; the record's own decides who may subscribe to it. In an app that turned the endpoint's authentication off entirely (Rask:Cqrs:Server:RequireAuthenticatedUser false — for an app with no accounts), a bare [Authorize] has nobody to require, so it only opens the notification; name a role or a policy to mean more than that.

Admitted once, at the open. The policy runs when the stream opens, so a stream already running keeps delivering until it drops — signing out elsewhere does not cut it mid-flight, and the next reconnect is refused. Where a revocation must take effect immediately, publish a change the page reacts to.

A reverse proxy that buffers responses would hold every event back, so the stream is sent with X-Accel-Buffering: no (nginx) and response buffering off. A comment every fifteen seconds keeps a quiet stream open through idle timeouts.

How values are delivered

• Latest first. A new subscription gets the most recent notification it matches, then every one after. Only a type somebody has subscribed to is remembered, so a domain event nobody watches is never held, and at most 4,096 are kept across the whole feed, oldest out first — so a subscription opened long after a quiet event may find nothing to replay.
• In order. Values arrive in the order they were published, and a replay racing a new publish never lands after it.
• Never holding the publisher up. PublishAsync hands the notification over and returns; it does not wait for any page to render.
• Bounded behind a slow reader. A subscriber more than 256 values behind loses the oldest of them. A subscription shows the latest state, so the middle of a burst costs nothing on screen.
• Reconnecting by itself. A dropped connection is reopened with backoff, from half a second up to thirty; IsLive is false meanwhile and Data keeps the last value. A refusal — a policy saying no, a record the server rejects — is final, and is not retried.

Every number above is a setting, read from Rask:Cqrs first and overridable in code: ReplayCapacity (4096), SubscriptionBuffer (256), SubscriptionReconnectDelay (00:00:00.500) and SubscriptionReconnectCeiling (00:00:30); the event stream's keep-alive is Rask:Cqrs:Server:EventKeepAlive (00:00:15).

What it is not

• Not a queue. Beyond the one replayed value, delivery is at most once, to the subscriptions open when it is published. When a page needs the current state rather than the latest change, query it, and patch the query with .Into.
• Not beyond its own process — below.

One process, for now

Every publish and every subscription in a process meet in memory, and that is the whole of it: one app on one server needs no configuration, no broker and no backplane.

What that means for more than one process — the two containers of a deploy, or several servers behind a load balancer — is that each hears only what it published itself. A page on the other one catches up when its queries refetch, not the moment the event happens. Carrying events between processes is on the roadmap; until then, keep anything that must reach every visitor in the database the pages already read.

Coming from IBroadcast

IBroadcast and Topic<T> are gone; a notification does what a topic did, and more.

→ Related: Rask.Query for queries and commands · CQRS for notifications and their handlers · scaling for running more than one server · composition for parent–child communication on one page