SignalR: realtime messages for SignalR clients

If your app already uses SignalR, Inngest.SignalR brings the realtime messages of your functions to its clients. A client calls a hub method to join the topics of a channel, and every message that a function publishes there reaches it as a call of a client method. The app decides who may join what, the browser holds no Inngest token, and clients that join the same topics share one subscription to Inngest.

Set up

Install Inngest.SignalR from NuGet. It brings in Inngest and Inngest.AspNetCore:

dotnet add package Inngest.SignalR

Then add SignalR, the SDK and the bridge, and map the hub. Authorize is required, because nobody is let in by default:

Program.cs

using Inngest;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddSignalR();
builder.Services.AddInngest(options => options.AppId = "shop");
builder.Services.AddInngestSignalR(options =>
{
    // Check here that the signed-in user may see the channel.
    options.Authorize = request => ValueTask.FromResult<RealtimeSubscription?>(request.Requested);
});

var app = builder.Build();

app.MapInngestRealtimeHub("/hubs/inngest").RequireAuthorization();
app.Run();

MapInngestRealtimeHub maps InngestRealtimeHub, a hub with two methods: Subscribe(channel, topics) and Unsubscribe(channel, topics). Like any hub endpoint it follows the app’s authorization, so RequireAuthorization() decides who may connect, and Authorize decides what a connected user may join.

Decide who may join

Authorize runs each time a client asks to join. The request holds the client’s connection, with its user, the channel and topics it asked for, and a scope to resolve services from, such as a database context, which ends when the callback returns. The callback returns the subscription to join:

• request.Requested lets the client in as it asked.
• The same channel with fewer topics narrows the request.
• Another channel gives the client something it did not name, such as one derived from its user.
• null refuses, and so does a subscription with no topics. The client gets a HubException saying so. Throw a HubException yourself to say why; SignalR sends its message to the client.

using System.Security.Claims;

public interface IOrderAccess
{
    ValueTask<bool> CanSeeAsync(ClaimsPrincipal? user, string orderId, CancellationToken cancellationToken);
}

using Microsoft.AspNetCore.SignalR;

builder.Services.AddInngestSignalR(options =>
{
    options.Authorize = async request =>
    {
        var requested = request.Requested;
        if (!requested.Channel.StartsWith("order:", StringComparison.Ordinal))
        {
            return null;
        }

        var orders = request.Services.GetRequiredService<IOrderAccess>();
        if (!await orders.CanSeeAsync(request.Caller.User, requested.Channel["order:".Length..], request.CancellationToken))
        {
            throw new HubException("That order is not yours.");
        }

        // The tokens of a model's reply are for staff.
        return request.Caller.User?.IsInRole("staff") == true
            ? requested
            : requested with { Topics = [.. requested.Topics.Where(topic => topic != "tokens")] };
    };
});

The callback is asked when the client joins and not again while it stays, so a user whose access ends keeps what they joined until their connection ends. SignalR can end a connection when the authentication it was made with expires. Ask for that in the options of the connection, which MapInngestRealtimeHub passes on to MapHub:

app.MapInngestRealtimeHub("/hubs/inngest", options => options.CloseOnAuthenticationExpiration = true)
    .RequireAuthorization();

Join from a client

With the JavaScript client

The client listens to Realtime, where each message arrives as the JSON Inngest sends it, in a string, and joins with Subscribe. The format is that of the server-sent events relay: { kind, channel, topic, data, run_id, created_at }, and for the text of a stream { kind: "chunk", stream_id, data } between a datastream-start and a datastream-end.

import * as signalR from "@microsoft/signalr";

const connection = new signalR.HubConnectionBuilder()
  .withUrl("/hubs/inngest")
  .withAutomaticReconnect()
  .build();

connection.on("Realtime", (json) => {
  const message = JSON.parse(json);
  if (message.kind === "data") showStatus(message.topic, message.data);
  if (message.kind === "chunk") appendText(message.data);
});

const join = (channel, topics) => connection.invoke("Subscribe", channel, topics);
const joinOrder = () => join(`order:${orderId}`, ["status", "tokens"]);

// The client lost a subscription, and its connection is as it was. The server sends back the channel and topics it
// asked for. After "behind" or "send-failed" messages were missed: join again and read the current state. After
// "inngest-ended" Inngest refused or ended the subscription: try again later, not at once.
connection.on("RealtimeError", (channel, topics, reason) => {
  if (reason === "inngest-ended") setTimeout(() => join(channel, topics), 10000);
  else join(channel, topics).then(loadState);
});

// A reconnect is a new connection with a new id, and the server has forgotten the old one: join again.
connection.onreconnected(joinOrder);

await connection.start();
await joinOrder();

RealtimeError carries the channel, the topics and one of these reasons:

Reason	Meaning
behind	The client fell MaxQueuedMessagesPerConnection messages behind. What was queued for it is dropped.
send-failed	A message could not be sent to the connection.
inngest-ended	Inngest refused or ended the subscription.

Subscribe resolves once the client has joined, and rejects with an error that holds the message of the HubException when it may not. A client that leaves a page calls connection.invoke("Unsubscribe", channel, topics) with the arguments it joined with, and a connection that closes leaves everything at once. Joining what a client is in already, and leaving what it is not in, change nothing.

Subscribe returns as soon as the client has joined. It does not wait for the app’s subscription to Inngest to open, and nothing tells the client when it has: there is no ready signal. Inngest delivers only to subscribers that are connected and keeps nothing for later, so a function that starts publishing right after Subscribe returns can publish its first messages before the subscription is open, and those are lost. Where the first messages matter, have the page read the current state after it has joined, and again after a reconnect or a RealtimeError, rather than rely on the stream alone.

With the .NET client

using System.Text.Json;
using Microsoft.AspNetCore.SignalR.Client;

var connection = new HubConnectionBuilder()
    .WithUrl("https://shop.example.com/hubs/inngest")
    .WithAutomaticReconnect()
    .Build();

connection.On<string>("Realtime", json =>
{
    using var message = JsonDocument.Parse(json);
    Console.WriteLine(message.RootElement.GetProperty("kind").GetString());
});
connection.On<string, string[], string>("RealtimeError", (channel, topics, reason) =>
    connection.InvokeAsync("Subscribe", channel, topics));
connection.Reconnected += _ => connection.InvokeAsync("Subscribe", "order:o-1", new[] { "status" });

await connection.StartAsync();
await connection.InvokeAsync("Subscribe", "order:o-1", new[] { "status" });

The JSON and MessagePack protocols carry a message alike, because it is a string.

Join from a hub of your own

An app that has a hub calls the bridge from its methods instead of mapping InngestRealtimeHub. The hub injects InngestRealtimeBridge<THub>, which AddInngestSignalR<THub> registers. A typed topic names the channel and topic once for the function that publishes and the hub that joins:

using Inngest;
using Microsoft.AspNetCore.SignalR;

public sealed record OrderStatus(string Text, int Percent);

public static class Topics
{
    public static RealtimeTopic<OrderStatus> Status(string orderId) => new($"order:{orderId}", "status");
}

public sealed class OrdersHub(InngestRealtimeBridge<OrdersHub> bridge) : Hub
{
    public Task Watch(string orderId) =>
        bridge.SubscribeAsync(Context, Topics.Status(orderId), Context.ConnectionAborted).AsTask();

    public void Unwatch(string orderId) => bridge.Unsubscribe(Context, Topics.Status(orderId));
}

builder.Services.AddInngestSignalR<OrdersHub>(options =>
    options.Authorize = request => ValueTask.FromResult<RealtimeSubscription?>(request.Requested));

app.MapHub<OrdersHub>("/hubs/orders");

SubscribeAsync asks the same Authorize callback, returns the subscription that was joined, and throws the HubException the client sees. The hub needs no base class and no filter: the bridge leaves a connection’s subscriptions when SignalR aborts the connection, whichever hub it belongs to. Messages reach the client method Realtime unless MessageMethod says otherwise; ErrorMethod names the one for errors.

What the bridge does

• One subscription to Inngest for each channel and set of topics, whatever the order of the topics. It opens with the first client that joins, and closes with the last that leaves or disconnects. Subscriptions that overlap, such as [status] and [status, tokens] of one channel, are two subscriptions, and a client that joined both receives a status message once for each.
• Every connection has a queue of its own and a loop that sends it to the client, one message at a time and in order. A client that reads slowly delays only itself. One that falls MaxQueuedMessagesPerConnection (512) messages behind loses its subscriptions: the queue is emptied, and the client hears of each subscription in RealtimeError(channel, topics, "behind"), with the channel and topics it asked for, to join again with the same call. A send that throws does the same, with the reason send-failed. The connection stays open, for the app may use it for more than this, and the bridge never ends it: HubCallerContext.Abort tells a client not to reconnect, even one with automatic reconnect. The messages the client missed are gone, so it reads the current state when it joins again. Sends that a backplane or service stops accepting fill the queues of all the clients in the same way.
• A subscription Inngest refuses, such as one with a signing key it does not accept, is logged. Every client that had joined it hears of it in RealtimeError(channel, topics, "inngest-ended"), once for each subscription it had in it, and is out of it. Joining again tries anew, and a refusal that lasts is refused again, so wait before it.
• A subscription that drops is opened again with a new token, as SubscribeAsync does, and what is published meanwhile is missed. A client whose connection drops is a new connection when it returns, and joins again.
• When the app stops, the subscriptions to Inngest close at once and nobody joins any more.
• A connection may be in 32 subscriptions by default, counting each way it asked to join, for each can be a connection from the app to Inngest that nobody else shares. MaxSubscriptionsPerConnection changes it.

Scale out

SignalR serves a connection from one app instance. With several instances, a backplane lets one instance send to a connection, a user or a group that is on another: Redis, or Azure SignalR Service, which holds the client connections itself. See Microsoft’s guide. Every instance of the app subscribes to Inngest for the clients it has, so a bridge that sent each message to a group would deliver it again and again: a send to a group reaches its members on every instance, whichever sent it. With the same group name on three instances that all forward a message, a client would receive it three times.

So the bridge never sends to a group or to everyone. An instance sends each message to the connections that joined through it, by connection id, and the other instances do the same for theirs. Each client receives each message once for each of its subscriptions that includes the topic, however many instances there are, on Redis and on Azure SignalR Service. What that costs and saves:

• Redis. A send to a connection that is on the instance is written straight to it, without Redis, so the bridge’s messages put no load on Redis and do not wait for it. A group send is published to Redis, where every instance that has a member of the group delivers it.
• Azure SignalR Service. The instance sends the service one message for each client, and the service delivers it. A group send would be one message for all its members, which is cheaper for the instance when a channel has a great many clients.
• Instance-scoped group names (<group>:<instance id>) would also be exact, and send one message for all the clients of an instance. They need the backplane when a client joins or leaves: Redis subscribes a group’s channel for the first member on a server and drops it with the last, and Azure SignalR Service takes a round trip to the service for each join and leave. And a group send waits for every member’s write, on the default in-process manager and in the handler Redis runs on each server, so a client that reads slowly holds up the others of its group. The bridge sends to connections so that it can cut that one client off from its subscriptions and no other.
• Each instance opens its own subscription to Inngest for each channel its clients joined, so a channel that clients of three instances watch has three subscriptions, not one.
• A reconnect can land on another instance, with a new connection id. The client joins again, which opens a subscription on that instance if it has none. The instance of the old connection leaves its subscriptions when SignalR ends that connection.

If thousands of clients watch one channel, consider letting their browsers subscribe to Inngest themselves with a token and useRealtime: Inngest does the fan-out, and no instance of your app sends a message for each client.

Trimming and Native AOT

The package’s own code is trim and Native AOT compatible. SignalR’s own support is narrower:

• The ASP.NET Core documentation lists SignalR as not supported with Native AOT on .NET 8. .NET 9 added trimming and Native AOT for the SignalR server and client.
• Only the JSON protocol, with a System.Text.Json source-generated context. MessagePack is not supported.
• No strongly typed hubs with Native AOT, and the asynchronous return types of hub methods are limited to Task, Task<T>, ValueTask and ValueTask<T>.
• Every type that crosses the wire needs a place in the context, registered with SignalR’s JSON protocol.

In a trimmed or Native AOT app AddInngestSignalR registers the two types of its own hub, string and string[] (the text of a message and the topics of Subscribe), after your context. A hub of your own and its client calls add their types to the context as they would without the bridge:

using System.Text.Json.Serialization;

public sealed record OrderRef(string OrderId);

[JsonSerializable(typeof(OrderRef))]
internal sealed partial class HubJsonContext : JsonSerializerContext;

using Microsoft.AspNetCore.SignalR;

builder.Services.Configure<JsonHubProtocolOptions>(options =>
    options.PayloadSerializerOptions.TypeInfoResolverChain.Insert(0, HubJsonContext.Default));

See Native AOT for the SDK itself.

Compared with server-sent events

MapInngestRealtime relays the same messages as server-sent events, with no client library.

	MapInngestRealtime	Inngest.SignalR
Client	EventSource	@microsoft/signalr or the .NET client
What a connection receives	The topics of one channel, named by the URL	Any channels, joined and left by calls on one connection
Who decides	A callback on the request, and the endpoint’s authorization	A callback on each join, and the hub’s authorization
After a drop	The browser reconnects, and the callback runs again	The client reconnects and joins again
Subscriptions to Inngest	One for each channel and topics, shared	One for each channel and topics, shared, in each instance
A slow client	Disconnected after 512 messages	Loses its subscriptions after 512 messages and is told, and stays connected
With several instances	Nothing to set up	Each instance sends only to its own clients

SignalR suits an app that has it already and wants these messages on the connection the page uses for the rest. A page that only listens needs nothing but EventSource. A browser can also subscribe to Inngest itself with a token and useRealtime, as Realtime shows, which leaves the fan-out to Inngest.

Limits

• Each message is a string of the JSON Inngest sent, up to Inngest’s own limit of 512 KB. SignalR’s MaximumReceiveMessageSize applies to what clients send, such as the topics of Subscribe.
• Delivery follows Inngest’s: to the clients that have joined, from the moment the subscription to Inngest is open, and nothing is kept for later.
• A client that joined through one instance of the app is served by that instance until its connection ends.