Getting started

Install

dotnet add package Pulse.Mqtt.Client
dotnet add package Pulse.Mqtt.DependencyInjection
dotnet add package Pulse.Mqtt.Serialization.Json

Pulse.Mqtt.Core comes in transitively. See Package add-ons to choose durable storage, compact binary serializers, Dataflow pipelines, alternate transport, the in-process test broker, and optional analyzers. Each package also has its own package page.

Register a client

var builder = Host.CreateApplicationBuilder(args);

builder.Services
    .AddPulseMqttClient("devices", options =>
    {
        options.Host = "broker.example.com";
        options.Port = 1883;
        options.ClientId = "my-service";
        options.KeepAliveSeconds = 30;
    })
    .UseSerializer(_ => new JsonMqttSerializer(AppJsonContext.Default));

The client starts with the host, connects in the background, reconnects on drops, and re-subscribes on its own. Resolve it anywhere:

var client = provider.GetRequiredService<IPulseMqttClientFactory>().GetClient("devices");

Manual control

Prefer to connect and disconnect the client yourself? Set options.ConnectWithHost = false and call ConnectAsync/DisconnectAsync whenever you want — see Lifecycle and state.

No host? Construct directly:

var factory = new TcpTransportFactory(new TcpTransportOptions { Host = "broker.example.com" });
await using var client = new ResilientMqttClient(factory, new ResilientMqttClientOptions
{
    Connect = new MqttConnectPacket { ClientId = "my-service" },
    Serializer = new JsonMqttSerializer(AppJsonContext.Default),
});
await client.ConnectAsync(ct);

Publish

// Typed, awaited to the broker acknowledgement at QoS 1.
var outcome = await client.PublishAsync(
    "sensors/boiler-1/telemetry",
    new TelemetryReading("C", 21.5, DateTimeOffset.UtcNow),
    MqttQualityOfService.AtLeastOnce);

// outcome.Disposition: Delivered, Queued (offline), or DroppedOffline — never silent.

Subscribe

await using var route = await client.OnAsync<TelemetryReading>(
    "sensors/{deviceId}/telemetry",
    MqttQualityOfService.AtLeastOnce,
    (reading, message, token) =>
    {
        Console.WriteLine($"{message.Values["deviceId"]}: {reading.Value}{reading.Unit}");
        return ValueTask.CompletedTask;
    },
    token);

OnAsync tells the broker to deliver sensors/+/telemetry, captures {deviceId}, and dispatches locally. Async disposal unregisters the local route and unsubscribes the broker filter. For advanced queue settings or separate subscription ownership, use the explicit SubscribeAsync plus RegisterRoute APIs.

Request and response

// One side asks…
var reply = await client.RequestAsync<StatusRequest, StatusReply>(
    "devices/boiler-1/status", new StatusRequest("dashboard"));

// …the other answers.
var statusTemplate = MqttRouteTemplate.Parse("devices/{deviceId}/status");
await client.SubscribeAsync([statusTemplate.ToTopicFilter(MqttQualityOfService.AtLeastOnce)], token);

using var responder = client.RegisterRequestHandler<StatusRequest, StatusReply>(
    statusTemplate,
    (request, message, token) =>
        ValueTask.FromResult(new StatusReply(message.Values["deviceId"], "online")));

Watch the connection

await client.WaitUntilConnectedAsync(TimeSpan.FromSeconds(10), token);   // readiness gate

await foreach (var change in client.WatchState(token))
{
    logger.LogInformation("MQTT: {Previous} -> {Current}", change.Previous, change.Current);
}

Run the sample

samples/Pulse.Mqtt.Sample exercises everything above. It needs no infrastructure — with no arguments it runs against the in-process test broker:

dotnet run --project samples/Pulse.Mqtt.Sample
dotnet run --project samples/Pulse.Mqtt.Sample -- --host localhost --port 1883

For ASP.NET Core hosting, health checks, keyed DI, diagnostics snapshots, and capability-aware MQTT 5 feature branching, run the Minimal API sample:

dotnet run --project samples/Pulse.Mqtt.AspNetCoreSample
dotnet run --project samples/Pulse.Mqtt.AspNetCoreSample -- --Mqtt:Host localhost --Mqtt:Port 1883

For worker pipelines with bounded Dataflow stages, explicit subscriptions, graceful shutdown, and the same capability-aware MQTT 5 branching, run the worker sample:

dotnet run --project samples/Pulse.Mqtt.WorkerSample
dotnet run --project samples/Pulse.Mqtt.WorkerSample -- --Mqtt:Host localhost --Mqtt:Port 1883

Next steps

• Connecting — TLS, WebSocket, credentials, protocol versions.
• Publishing — QoS levels, outcomes, retained messages, MQTT 5 properties.
• Resilience — reconnect policy, offline queue, sticky faults.
• Testing — millisecond tests with the in-process broker.