.NET SDK for the HTTP server API of Centrifugo: every API method, including those of Centrifugo PRO; broadcasts, batches and automatic batching; connection, subscription and server API tokens with JWKS; the proxy handlers and the endpoints client SDKs call. For a real-time client over WebSocket, see centrifuge-csharp.
dotnet add package Centrifugal.Cent
dotnet add package Centrifugal.Cent.AspNetCore # serving handlers with ASP.NET Core, AddCentClient| Version | |
|---|---|
| .NET | 8 or newer; trimming and native AOT supported |
| Centrifugo | v6, any release |
| Map and shared poll methods | Centrifugo v6.8.0 or newer |
CentApiException.Temporary, code 111 (too many requests) |
Centrifugo v6.10.0 or newer |
Ed25519 tokens verified without JWKS, by client.token.ed25519_public_key |
Centrifugo v6.10.2 or newer |
| Methods marked "Centrifugo PRO only" | Centrifugo PRO v6 |
| Package | What | Depends on |
|---|---|---|
Centrifugal.Cent |
The client, tokens, JWKS, shared poll signatures, commands, and the HTTP handlers' framework-free core | nothing but .NET |
Centrifugal.Cent.AspNetCore |
MapCentrifugo to serve the handlers, AddCentClient for dependency injection |
ASP.NET Core |
csharpcent implements the Centrifugo server SDK specification, version 0.4, whose reference implementation is gocent.
using Centrifugal.Cent;
await using var client = new CentClient(new CentClientOptions
{
ApiEndpoint = "http://localhost:8000/api",
ApiKey = apiKey,
});
var result = await client.PublishAsync("news", new { text = "hello" });
Console.WriteLine(result.Offset);Create one client per process and share it: it is safe for concurrent use and
keeps its connections alive. Every API method is a method of the client taking
its request - PublishAsync, PresenceAsync, HistoryAsync... - with all the
method's fields. A request's fields the protos mark required are required
members, so the compiler says when one is missing; one set to null, "" or
0 - which go out as the field left unset - fails before sending with
CentInvalidRequestException naming it, nested ones included
(LimitStrategy.RateLimit.Policies[0].Rate is required). A field whose empty
value has a meaning is not checked: InvalidateUserTokensRequest.User, where
empty means anonymous users. Publish and broadcast also have short forms taking
the channel and data:
await client.PublishAsync(new PublishRequest { Channel = "news", Data = data, IdempotencyKey = key, SkipHistory = true });
await client.BroadcastAsync(["news", "sport"], data);In ASP.NET Core, register the client with AddCentClient, configured from
appsettings.json - it is closed when the application stops:
builder.Services.AddCentClient(builder.Configuration.GetSection("Centrifugo"));{
"Centrifugo": {
"ApiEndpoint": "http://localhost:8000/api",
"ApiKey": "...",
"RequestTimeout": "00:00:05",
"AutoBatch": { "MaxInFlight": 32 }
}
}A string field with a fixed set of values has a constant per value, generated
from the protos: ProviderValues.Fcm, PlatformValues.IOS, OpValues.Add,
StatusValues.Delivered, KeyModeValues.IfNew, CmpValues.Eq,
FilterNodeOpValues.And, and in the proxy EncodingValues.Binary,
ChannelEventTypeValues.Occupied, CapabilityMatchValues.Wildcard and the
rest. The fields stay strings: a value Centrifugo
adds is sent before the SDK names it.
await client.DeviceRegisterAsync(new DeviceRegisterRequest
{
Provider = ProviderValues.Fcm,
Platform = PlatformValues.Android,
Token = fcmToken,
User = "42",
});device_update's updates are sent whenever they are set, even empty: an empty
UserUpdate detaches the devices from their user, an empty TimezoneUpdate,
LocaleUpdate or MetaUpdate clears the setting, and one left null changes
nothing. Any other message in a request whose fields are all empty is left out,
as an empty field is.
await client.DeviceUpdateAsync(new DeviceUpdateRequest
{
Users = ["42"],
TimezoneUpdate = new DeviceTimezoneUpdate { Timezone = "" }, // clears it
});What configuration cannot hold - a bearer token function, say - goes in a
second argument: AddCentClient(section, (services, options) => ...). The
client's HttpClient comes from IHttpClientFactory, with the SDK's default
handler; AddCentClient returns its builder, for handlers of your own:
.AddHttpMessageHandler<MyHandler>().
A payload - Data, Info, Meta and the like - is any JSON value:
- an object of the application's, or an anonymous type, encoded as ASP.NET Core
and SignalR encode what goes to browsers - camelCase names (
JsonSerializerDefaults.Web); - a
JsonElementorJsonNode; - a string, number or boolean - a string is a JSON string, never JSON text;
RawJson: JSON text sent as the same value, number text kept - JSON the application already has, or encoded with options of its own.
new PublishRequest { Channel = "news", Data = new Message("hello") };
new PublishRequest { Channel = "news", Data = RawJson.Parse(jsonFromDatabase) };
new PublishRequest { Channel = "news", Data = RawJson.Serialize(message, MyJsonContext.Default.Message) }; // native AOTReplies give payloads as JsonElement?: null when absent, an element of kind
Null for JSON null. A payload which came back goes out again as the same JSON
value - republishing a publication from history, or answering a proxy with the
client's data, changes nothing. Decode<T>() reads one into an application's
type, as the SDK encodes it:
var history = await client.HistoryAsync(new HistoryRequest { Channel = "news", Limit = 10 });
var message = history.Publications[0].Data?.Decode<Message>();Binary data goes base64-encoded in the B64 fields (B64Data, B64Info):
bytes given as a JSON payload are refused, as are NaN, infinities and invalid
JSON - before anything is sent. JSON null as a payload is RawJson.Null.
JavaScript - where most Centrifugo payloads end up - reads JSON numbers as
doubles, which hold integers exactly only up to 2^53. Send integers which may
pass it - 64-bit IDs - as strings: new { id = id.ToString() }, or
[JsonNumberHandling(JsonNumberHandling.WriteAsString)] on the property.
Every call has one deadline: RequestTimeout (10 seconds by default), or the
timeout of a client from WithTimeout, which replaces it for that client's
calls, longer or shorter, sharing everything else. It bounds the whole call -
connecting, sending, reading the reply, a reply trickling in included - and its
end throws CentTimeoutException. A call's CancellationToken only shortens
it, throwing OperationCanceledException, and never extends it: a call made
with the token of an HTTP request which lives for minutes still gives up on a
stuck Centrifugo at RequestTimeout. To let calls take longer, raise the
timeout: it cannot be turned off. RequestTimeout zero is the default, and
Timeout.InfiniteTimeSpan, a negative timeout or one longer than 24 days is
refused - by CentClient's constructor, WithTimeout, AttemptTimeout and
the iterators' page timeout alike.
var history = await client.WithTimeout(TimeSpan.FromSeconds(30)).HistoryAsync(request, ct);HistoryAllAsync, DeviceListAllAsync, DeviceTopicListAllAsync,
UserTopicListAllAsync and MapReadStateAllAsync give every item of a paged
method as an IAsyncEnumerable, reading a page when the iteration reaches it:
leaving the loop reads no further page, and a failed call ends it with its
exception. The request's Limit is the page size. Each page's request is
bounded by RequestTimeout, or by a page timeout of its own -
client.DeviceListAllAsync(request, pageTimeout: TimeSpan.FromSeconds(30), ct) -
replacing it, longer or shorter. Nothing bounds the iteration as a whole but
the caller's CancellationToken: it lasts as long as the loop, and the timeout
of a client from WithTimeout does not apply to its pages. Items changing while
iterating - a device registered, an entry updated - may or may not be seen.
MapReadStateAllAsync reads every page after the first at the first one's
revision, unless the request sets RevisionOffset and RevisionEpoch: a map
state reset between pages fails with ErrorCodes.UnrecoverablePosition (112)
rather than mixing two states - read again from the start then.
The results holding a stream position - publish, history, map publish, map
remove, map read state and map read stream - give it as a StreamPosition,
result.Position, which is what HistoryRequest.Since takes. So do the items
of the history and map state iterators: HistoryAllAsync yields a
HistoryItem, the Publication with its Position - where to resume, or a
subscription's RecoverSince - and MapReadStateAllAsync a MapStateItem, the
Entry with State, the position the state was read at, which
MapReadStreamAsync follows it from. Reading what came after a publication:
var published = await client.PublishAsync("news", data, ct);
// ...
await foreach (var item in client.HistoryAllAsync(new HistoryRequest { Channel = "news", Limit = 100, Since = published.Position }, ct))
{
Handle(item.Publication.Data);
last = item.Position; // to resume from later
}HistoryAllAsync reads oldest first, so it refuses Reverse, and a Limit
below 1, before sending anything. A position whose epoch is no longer the
stream's fails with ErrorCodes.UnrecoverablePosition (112): the stream was
reset - Redis restarted, say - and reading from its start is the only way on.
A call returns its result only when it did everything asked, and throws a
CentException otherwise:
| Exception | When |
|---|---|
CentApiException |
Centrifugo refused the request or failed to carry it out: match Code against ErrorCodes, never the message |
CentHttpException |
The request did not reach the API method: StatusCode, the start of Body, Headers. CentUnauthorizedException for 401 |
CentTimeoutException |
The call's deadline ended |
CentNetworkException |
The connection was refused, reset or not made |
CentDecodeException |
The reply is not Centrifugo's - usually a wrong endpoint |
CentInvalidRequestException |
Not sent: the request cannot be, as it is |
CentInvalidConfigException |
The client refuses to work as configured - not created, or, before sending, an endpoint from ApiEndpointFunc failing ApiEndpoint's rules or an empty or malformed token from BearerTokenFunc: the message names the option and the fix |
CentFunctionException |
ApiEndpointFunc or BearerTokenFunc threw, before sending: Function names it, InnerException is what it threw, and Retry.IsRetryable judges that |
CentClosedException |
A call after CloseAsync |
BroadcastFailedChannelsException, BatchFailedCommandsException |
Some channels or commands failed, the others succeeded: below |
try
{
await client.PublishAsync(request, ct);
}
catch (CentApiException e) when (e.Code == ErrorCodes.UnknownChannel)
{
// the channel's namespace does not exist
}Error messages carry hints: a 404 says whether the endpoint is wrong or the
server is Centrifugo OSS answering a PRO method; a redirect says to check http
or https and the endpoint's path - Centrifugo's API never redirects a call, but
its router redirects a path not in canonical form, with a doubled slash say. With http_api.error_mode: transport, Centrifugo's errors are the same
CentApiExceptions.
Retry.IsRetryable(e) tells whether another attempt, with time left, may
succeed: a temporary Centrifugo error, a network failure, a timeout, an HTTP 429
or 5xx. A retryable error does not mean nothing was done: a publication may have
been published before its reply was lost. So retry a publication only with the
same IdempotencyKey:
var delay = Retry.ExponentialBackoff(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(5));
var request = new PublishRequest { Channel = "orders", Data = order, IdempotencyKey = $"order-{order.Id}-paid" };
for (var attempt = 1; ; attempt++)
{
try
{
await client.PublishAsync(request, ct);
break;
}
catch (CentException e) when (attempt < 3 && Retry.IsRetryable(e))
{
await Task.Delay(delay(attempt), ct);
}
}Idempotency keys:
- each publication or broadcast gets its own key, reused only to retry that same one; a broadcast needs one key, not one per channel;
- Centrifugo remembers a key per channel, five minutes by default: a second
publication under the same key is not published, so a key derived from an
entity must also name the event -
order-42-paid, notorder-42; - the memory, Redis and PostgreSQL brokers keep keys; the NATS broker ignores them, so a retry publishes again.
A broadcast publishes into many channels, with every channel's outcome in the
result, in order. When some channels fail it throws
BroadcastFailedChannelsException holding every outcome:
try
{
await client.BroadcastAsync(new BroadcastRequest { Channels = channels, Data = data }, ct);
}
catch (BroadcastFailedChannelsException e)
{
logger.LogWarning("broadcast failed in {Channels}", e.Failed.Select(c => c.Channel));
}e.Result.Channels holds every channel in the request's order, so the outcome
of the n-th channel you asked for is e.Result.Channels[n] - the way back from a
channel to the user or entity it was made for.
Never retry a whole broadcast: the other channels got the publication. Retry
only e.Result.RetryableChannels(), with the same idempotency key - or use
BroadcastWithRetryAsync, which does that, with one key on every attempt (the
request's, or one it makes and returns in the result):
var result = await client.BroadcastWithRetryAsync(request, new BroadcastRetryOptions
{
MaxAttempts = 3,
OnAttemptError = attempt =>
{
logger.LogWarning(attempt.Error, "attempt {N}, retrying {Channels}", attempt.Attempt, attempt.RetryChannels);
return null; // the default backoff; a TimeSpan to wait; throw to stop
},
}, ct);Each attempt has its own timeout: AttemptTimeout, replacing RequestTimeout
for it, longer or shorter, or RequestTimeout without one. The call's
CancellationToken, and the timeout of a client from WithTimeout, bound the
attempts and the waits between them together. In an automatic batch a call
keeps the deadline it would have alone, and the batch request lasts until the
latest of its calls' deadlines, at least RequestTimeout from when it is sent -
so it may outlast one call's.
A batch sends commands in one request. A method per API method adds one and
returns its Pending<T>, holding its outcome once the batch is sent:
var batch = client.NewBatch();
var publish = batch.Publish(new PublishRequest { Channel = "news", Data = data });
var presence = batch.Presence(new PresenceRequest { Channel = "chat" });
await batch.SendAsync(ct); // BatchFailedCommandsException when some failed
Console.WriteLine(publish.Result.Offset);A batch is sent once. By default Centrifugo runs its commands in order;
NewBatch(parallel: true) runs them concurrently. Centrifugo PRO sends a
sequential batch's publications to its broker together: each channel's take
effect in order, while publications into different channels may not, unless the
namespace sets publication_grouping_disabled. A PRO method in a batch sent to
Centrifugo OSS fails with code 104 (not found), with a hint.
For processes making many calls at once - a web application publishing on every
request, a worker draining a queue - AutoBatch sends concurrent calls together,
cutting the requests Centrifugo handles under load:
var client = new CentClient(new CentClientOptions
{
ApiEndpoint = endpoint,
ApiKey = apiKey,
AutoBatch = new AutoBatchOptions { MaxInFlight = 32 },
});Below MaxInFlight requests in flight every call is sent at once, alone: nothing
ever waits for a batch to fill. At the cap, calls queue, and when a request
finishes those queued go out as one batch (at most MaxBatchSize, 1000). Every
call gets exactly its own result or error, and calls made one after another keep
their order. Choose MaxInFlight at about half the calls the process makes at
once: above that it never batches. A proxy in front of Centrifugo may limit a
request's size where Centrifugo does not - nginx: 1 MB by default - failing a
batch with a 413: lower MaxBatchSize, or raise the proxy's limit.
ApiKey sends X-API-Key. Centrifugo PRO also takes API
tokens: JWTs signed with keys
Centrifugo reads from a JWKS endpoint. ApiTokenSource signs them, reusing each
for half its lifetime:
var tokens = new ApiTokenSource(apiKey, new ApiTokenClaims { Audience = "centrifugo-api", Subject = "billing" });
var client = new CentClient(new CentClientOptions { ApiEndpoint = endpoint, BearerTokenFunc = tokens.GetTokenAsync });Centrifugo PRO accepts a token signed with any key of its JWKS endpoint, and checks audience, issuer and scope only when configured to. So sign API tokens with keys of their own, publish them at a JWKS endpoint other than the client tokens' one, and configure Centrifugo PRO to require an audience or scope only API tokens carry: otherwise any user can call the server API with their connection token.
The default HTTP client keeps up to 256 connections to Centrifugo alive, all of
them while idle, closes idle ones after 50 seconds, drains reply bodies so
connections are reused, never follows redirects, and uses no proxy. It sets no
timeouts of its own: each call's deadline bounds it. AddCentClient makes it
with IHttpClientFactory, under the name Centrifugal.Cent, and returns its
builder; elsewhere, give HttpClient to change any of that. The SDK never
changes a client it is given: credentials and headers go on each request.
Mind a resilience handler which retries requests: it would publish again a
publication whose reply was lost - twice on the NATS broker, which ignores
idempotency keys - and its timeouts would cut the SDK's deadline short. Disable
its retries of POST, and retry with Retry and idempotency keys instead:
builder.Services.AddCentClient(builder.Configuration.GetSection("Centrifugo"))
.AddStandardResilienceHandler(o => o.Retry.DisableForUnsafeHttpMethods());Tracing and metrics come from .NET's HTTP instrumentation: OpenTelemetry's
AddHttpClientInstrumentation() traces every request, and System.Net.Http's
metrics count them. OnRequest reports each request - method, commands, failed
ones, status, duration, error - for logs and metrics of your own. It is called
synchronously and must not throw: what it throws is the call's error.
An automatic batch runs in an empty execution context, never a caller's. To give
it a tracing span linking its callers', use BatchContext:
options.BatchContext = callers =>
{
var links = new List<ActivityLink>();
foreach (var context in callers)
{
ExecutionContext.Run(context, _ =>
{
if (Activity.Current is { } a)
{
links.Add(new ActivityLink(a.Context));
}
}, null);
}
return source.StartActivity("centrifugo batch", ActivityKind.Client, default(ActivityContext), links: links);
};The client's methods are virtual, so code using it is tested with a mocking
library, or a class of your own deriving from CentClient:
var client = Substitute.For<CentClient>(); // NSubstitute; new Mock<CentClient>() in Moq
client.PublishAsync(Arg.Any<PublishRequest>(), Arg.Any<CancellationToken>())
.Returns(new PublishResult { Offset = 1 });
client.BroadcastAsync(Arg.Any<BroadcastRequest>(), Arg.Any<CancellationToken>())
.Returns<BroadcastResult>(_ => throw new BroadcastFailedChannelsException(new BroadcastResult(
[new ChannelResult("a", new PublishResult()), new ChannelResult("b", new CentApiException(102, "unknown channel"))])));The short forms call the request forms, so mocking PublishAsync(PublishRequest, ...)
answers PublishAsync(channel, data) too. Results, errors and the
partial-failure exceptions are all made with new:
new CentApiException(ErrorCodes.TooManyRequests, "too many requests", temporary: true);
new CentNetworkException("publish", new HttpRequestException("connection reset"));
new CentTimeoutException("publish", TimeSpan.FromSeconds(10));BroadcastWithRetryAsync sends its attempts itself, not through the virtual
BroadcastAsync: mock it directly, or test its retries against a fake
HttpMessageHandler, below. Batches are mocked the same
way, their commands returning Pending.FromResult(...) or Pending.FromException<T>(...):
var batch = Substitute.For<CentBatch>();
batch.Publish(Arg.Any<PublishRequest>()).Returns(Pending.FromResult(new PublishResult { Offset = 1 }));
client.NewBatch(Arg.Any<bool>()).Returns(batch);To test against Centrifugo's replies instead - batches included - give the
client an HttpClient whose handler answers as Centrifugo does: a method's reply is {"result": {...}} or
{"error": {"code": ..., "message": ...}}; a broadcast's,
{"result": {"responses": [...]}} with one response per channel; a batch's,
{"replies": [...]} with one reply per command, {"<method>": {...}} or
{"error": {...}}.
// Publishes succeed; a broadcast fails in the channels starting with "missing:".
sealed class FakeCentrifugo : HttpMessageHandler
{
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken ct)
{
var reply = """{"result":{"offset":1,"epoch":"e"}}""";
if (request.RequestUri!.AbsolutePath.EndsWith("/broadcast"))
{
// One response per requested channel, in order.
var body = JsonNode.Parse(await request.Content!.ReadAsStringAsync(ct))!;
var responses = body["channels"]!.AsArray().Select(c => c!.GetValue<string>().StartsWith("missing:")
? """{"error":{"code":102,"message":"unknown channel"}}"""
: """{"result":{"offset":1}}""");
reply = $$$"""{"result":{"responses":[{{{string.Join(",", responses)}}}]}}""";
}
return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(reply) };
}
}
var client = new CentClient(new CentClientOptions
{
ApiEndpoint = "http://centrifugo.test/api",
HttpClient = new HttpClient(new FakeCentrifugo()),
});CloseAsync refuses new calls with CentClosedException, waits for those in
progress - automatic batches whose callers left included - and then closes the
idle connections. Its CancellationToken ends the wait, leaving the calls to go
on; it may be called again, concurrently too, each call waiting the same way.
A broadcast retry stops at its next attempt, which fails with
CentClosedException. DisposeAsync closes the client too, and a client of
AddCentClient is closed when the application stops.
The proxy handlers, the endpoints for client SDKs and the JWKS endpoint are
CentHandlers, free of any framework: HandleAsync takes a HandlerRequest
and returns a HandlerResponse. Centrifugal.Cent.AspNetCore serves them:
app.MapCentrifugo("/centrifugo/connect", proxy.Connect(Connect));MapCentrifugo returns the endpoint's builder, for its conventions:
.RequireHost(...), authorization. A handler's function reaches ASP.NET Core's
HttpContext - the user, cookies, services - with
context.Request.GetHttpContext(). A request a handler fails is logged through
the application's logging, under the category Centrifugal.Cent: a malformed
one as a warning, a failure of its function as an error, with the exception.
Other frameworks call HandleAsync; a response's Error says why it failed.
ProxyHandlers makes a handler per type of Centrifugo's HTTP
proxy, from a function taking the
decoded request:
using Proxy = Centrifugal.Cent.Proxy;
var proxy = new Proxy.ProxyHandlers();
app.MapCentrifugo("/centrifugo/connect", proxy.Connect(async (request, context) =>
{
var http = context.Request.GetHttpContext();
var user = await sessions.UserAsync(http.Request.Cookies["session"], context.CancellationToken)
// 4501: closed for good - a code in [4500, 4999] tells the client not to reconnect.
?? throw new Proxy.ProxyDisconnectException(4501, "unauthorized");
return new Proxy.ConnectResult { User = user.Id };
}));
var subscribe = proxy.Subscribe(async (request, context) =>
{
if (!await permissions.CanReadAsync(request.User, request.Channel, context.CancellationToken))
{
throw new Proxy.ProxyErrorException(403, "permission denied");
}
return new Proxy.SubscribeResult();
});The function returns a result to accept the request; throws
ProxyErrorException to reject it with an error the client receives - a code in
[400, 1999]; or throws ProxyDisconnectException to close the client's
connection - a code in [4000, 4999], those in [4500, 4999] telling the client
not to reconnect, and a reason of at most 32 bytes. Refresh and sub refresh
answer only with a result, ending through its Expired. Any other exception
fails the request with HTTP status 500, logged - and reported to
ProxyHandlerOptions.OnError, when set - and so does a function returning
null: Centrifugo would take a response with none of result, error and
disconnect as accepting a subscribe, publish, rpc, map publish or map remove as
sent, as refusing a connect unless anonymous connections are allowed, and as a
refresh or sub refresh which expired. Payloads arrive
as JsonElement? - in the B64 fields in binary mode - and go back as they are.
The proxy's types live in Centrifugal.Cent.Proxy: some share names with the
API's (PublishRequest, SubscribeResult), so a file using both refers to them
as Proxy.PublishRequest, with using Proxy = Centrifugal.Cent.Proxy;.
Keep proxy endpoints reachable by Centrifugo only: whoever can call them can claim to be any user. The ways to do it:
- a network only Centrifugo reaches;
- a secret header Centrifugo adds to every proxy request - the proxy's
static_headersoption - which the application checks before the handler, in a middleware or an endpoint filter; - TLS where the traffic leaves a trusted network.
The handler's body limit and the deadline it passes on are its own share of the protection.
using Centrifugal.Cent.Tokens;
var key = SigningKey.Hmac(builder.Configuration["Centrifugo:TokenSecret"]!);
var token = Token.Connection(key, new ConnectionClaims
{
User = "42",
Expiry = Expiry.After(TimeSpan.FromHours(1)),
Info = new { name = "Alice" },
});
var subscription = Token.Subscription(key, new SubscriptionClaims { User = "42", Channel = "news", Expiry = Expiry.At(end) });Keys: SigningKey.Hmac (HS256), Rsa of 2048 bits or more (RS256), ECDsa on
P-256, P-384 or P-521 (ES256, ES384, ES512), FromPem for RSA and ECDSA private
keys, and Ed25519 (EdDSA). A smaller RSA key, or an ECDSA key on another curve,
is refused there. A token's Expiry is required - a time, or Expiry.Never
chosen explicitly. ConnectionExpiry and SubscriptionExpiry set when the connection
or subscription expires, if not at the token's expiry. A connection token's
Meta is not visible to other clients, but the user owning the token can decode
it: never put in it what that user must not see. A subscription token's User
is the connection's user, and its Client binds it to one connection, by client
ID - for a token made for a connection the application knows: Centrifugo refuses
it from any other. A connection token's Profile sets Centrifugo PRO's
compression profile.
.NET has no Ed25519, so an Ed25519 key takes the signing function of a library of your choice, and the SDK depends on none - with BouncyCastle:
// dotnet add package BouncyCastle.Cryptography
var privateKey = new Ed25519PrivateKeyParameters(seed);
var key = SigningKey.Ed25519(privateKey.GeneratePublicKey().GetEncoded(), data =>
{
var signer = new Ed25519Signer();
signer.Init(true, privateKey);
signer.BlockUpdate(data, 0, data.Length);
return signer.GenerateSignature();
}, keyId: "2026-10");Jwks.Endpoint(keys) serves public keys for client.token.jwks_public_endpoint
and PRO's http_api.jwks.endpoint. Rotate keys so: publish a new key next to
the old one; sign with it only once every instance of the application publishes
it - from an activation time set ahead, or in a second deploy; drop the old key
once its tokens have expired. Never give a key ID to another key: Centrifugo
caches keys by ID.
Make the key once, at startup, and use it in the handlers built then:
var endpoints = new EndpointHandlers();
app.MapCentrifugo("/centrifugo/connection_token", endpoints.ConnectionToken(context =>
{
var user = context.Request.GetHttpContext().User.Identity?.Name
?? throw new EndpointForbiddenException();
return Task.FromResult(Token.Connection(key, new ConnectionClaims { User = user, Expiry = Expiry.After(TimeSpan.FromMinutes(15)) }));
}));EndpointHandlers makes the endpoints the getToken and getSignature
callbacks of client SDKs call: ConnectionToken, SubscriptionToken and
SharedPollSignature. The function throws EndpointForbiddenException to
answer 403 Forbidden - the client stops asking, and reports it as its
UnauthorizedError; any other exception answers 500, and the client retries.
It is not the API client's CentUnauthorizedException, which means Centrifugo
refused this server's credentials.
var signature = SharedPoll.Sign(secret, user, "poll:prices", keys, Expiry.After(TimeSpan.FromMinutes(5)));The signature covers exactly keys, in order. Behind the signature endpoint,
keys the function leaves out are revoked: the client stops tracking them.
var command = Command.Publish(new PublishRequest { Channel = "news", Data = data });
await producer.ProduceAsync("centrifugo", new Message<string, string> { Key = "news", Value = command.ToJson() });A command is {"method": ..., "payload": {...}}, the message Centrifugo's
async consumers take - Kafka,
NATS JetStream, Redis Streams, a PostgreSQL outbox table, where command.Payload
goes into a jsonb column. It validates the request as a call would, so a
command Centrifugo could not run fails when written. A consumer keeps commands
in order only within a partition: a channel's commands keep their order only
when every writer derives their partition from the channel the same way.
Publish a model's output into a channel of its own, in a namespace with history and recovery, from one task - each publish waiting for the previous one - in chunks of what arrived every 50 ms or so, not a publication per token. See Centrifugo for AI apps, which also covers sizing history for a whole response.
- The examples: publishing, an ASP.NET Core backend, native AOT.
- Docs for agents and LLMs: centrifugal.dev/llms.txt, an index of the current (v6) documentation.
- Development: see CONTRIBUTING.md and AGENTS.md.
MIT license. Centrifugo PRO, which the methods marked "Centrifugo PRO only" call, is commercial software under its own license.