Fluent API

Every operation in Pulse has a direct form (options records, explicit packets) and a fluent form. The fluent layer is a thin set of builders over the same APIs — same semantics, same guarantees, no reflection, fully Native AOT safe. Use whichever reads better; mix freely.

Building a client

For use without dependency injection, PulseMqttClientBuilder replaces hand-assembling the options records:

await using var client = await new PulseMqttClientBuilder()
    .WithTcp("broker.example.com", 8883, useTls: true)
    .WithClientId("service-1")
    .WithCredentials("device-42", "secret")
    .WithKeepAlive(TimeSpan.FromSeconds(30))
    .WithCleanStart(false)
    .WithSerializer(new JsonMqttSerializer(AppJsonContext.Default))
    .WithBackoff(TimeSpan.FromMilliseconds(500), TimeSpan.FromSeconds(30))
    .WithOfflineQueue(capacity: 2048, OverflowPolicy.DropOldest)
    .BuildAndConnectAsync(ct);

Everything configurable on ResilientMqttClientOptions has a method: the swap points (WithReconnectStrategy, WithReconnectDecision, WithLifecycle, WithSessionStore, WithMessageStore), WithTransport for WebSocket or the in-process test broker, WithLogger, WithTimeProvider for fake-clock tests, WithRawOptions for handshake timeouts, and WithConnect as the full-CONNECT escape hatch (wills, session expiry, enhanced auth).

Build() returns the client disconnected; BuildAndConnectAsync(ct) connects it (reconnects continue in the background, as always). Validation is explicit: no transport or no identity fails with a message naming the missing call, and mixing WithConnect with the individual identity methods is rejected rather than silently merged.

With dependency injection

Registered clients already have a fluent surface — AddPulseMqttClient(...) returns a builder with the same swap methods. See Dependency injection.

Publishing

var outcome = await client.Publish("sensors/boiler-1/telemetry")
    .AtLeastOnce()                          // or .ExactlyOnce(), .WithQualityOfService(...)
    .WithRetain()
    .WithMessageExpiry(TimeSpan.FromMinutes(5))
    .WithUserProperty("tenant", "acme")
    .WithPayload(reading)                   // typed: serializes and stamps content type
    .SendAsync(ct);

WithPayload takes a typed value (through the configured serializer), a string (UTF-8, with the payload format stamped), or raw bytes. WithContentType, WithResponseTopic, and WithCorrelationData cover the remaining MQTT 5 properties. SendAsync returns the same PublishOutcome as PublishAsync — Delivered, Queued, or DroppedOffline, never silent.

Routing

For endpoint-style routing, OnAsync subscribes the broker filter and registers the local route in one call:

await using var route = await client.OnAsync<TelemetryReading>(
    "sensors/{deviceId}/temp",
    MqttQualityOfService.AtLeastOnce,
    (reading, message, ct) =>
        Handle(reading, message.Values["deviceId"]),
    ct);

Async disposal unregisters the local route and unsubscribes the broker filter.

When one route should own both broker subscription and local dispatch, the route builder has asynchronous terminals:

await using var route = await client.Route("sensors/{deviceId}/temp")
    .AtLeastOnce()
    .WithConcurrency(4)
    .HandleAsync<TelemetryReading>((reading, message, ct) =>
        Handle(reading, message.Values["deviceId"]), ct);

When broker acknowledgement must wait for application work, make the route delivery mode manual before the terminal:

await using var route = await client.Route("orders/{id}")
    .AtLeastOnce()
    .ManualAcknowledgement()
    .HandleAsync(async (message, ct) =>
    {
        await PersistAsync(message.Message, ct);
        await message.AcknowledgeAsync(ct);
    }, ct);

ManualAcknowledgement().StreamAsync(ct) returns a subscribed acknowledged stream and disposes both the local route and broker subscription with the stream.

When you want explicit subscription ownership, use the local route terminals:

var route = client.Route("sensors/{deviceId}/temp");
await client.SubscribeAsync([route.ToTopicFilter(MqttQualityOfService.AtLeastOnce)], ct);

using var registration = route
    .WithQueue(capacity: 128, RouteOverflow.DropOldest)
    .WithConcurrency(4)
    .Handle<TelemetryReading>((reading, message, ct) =>
        Handle(reading, message.Values["deviceId"]));

Local terminals are Handle (raw handler), Handle<T> (typed), and Stream for the await foreach form. Everything from Routing — broker subscription, bounded queues, overflow, acknowledgement mode, and fault isolation — applies unchanged.

Request and response

var reply = await client.Request("devices/boiler-1/status")
    .WithTimeout(TimeSpan.FromSeconds(5))
    .WithQualityOfService(MqttQualityOfService.AtLeastOnce)
    .SendAsync<StatusRequest, StatusReply>(new StatusRequest("dashboard"), ct);

The raw terminal pairs with WithPayload/WithContentType when payloads are untyped:

MqttPublishPacket raw = await client.Request("devices/boiler-1/status")
    .WithPayload(requestBytes)
    .SendAsync(ct);

Correlation, the private reply subscription, and timeouts behave exactly as in Request and response.

Design notes

• Builders are plain mutable classes returning this — no expression trees, no reflection, nothing for the trimmer to warn about.
• Each terminal delegates to the corresponding client method, so behavior, diagnostics, and outcomes are identical between the fluent and direct forms.
• The route and request builders reuse the options records (MqttRouteOptions, MqttRequestOptions); subscription options stay on MqttTopicFilter. There is no second configuration model to learn — the DSL is shorthand, not a dialect.