Connecting

A connection is described by two things: a transport (where the bytes go) and a CONNECT packet (who you are to the broker). Everything else — retries, keep-alive, re-subscription — is handled for you.

TCP

The default. With dependency injection, Host, Port, and UseTls configure it:

services.AddPulseMqttClient("devices", options =>
{
    options.Host = "broker.example.com";
    options.Port = 1883;
    options.ClientId = "my-service";
});

Directly, build the factory yourself:

var factory = new TcpTransportFactory(new TcpTransportOptions
{
    Host = "broker.example.com",
    Port = 1883,
});

Sockets run with NoDelay and each packet is written as a single buffer — there is no fragmentation penalty on proxied or Nagle-affected paths.

TLS

options.Port = 8883;
options.UseTls = true;

For client certificates, SNI overrides, or custom validation, configure the transport options:

var factory = new TcpTransportFactory(new TcpTransportOptions
{
    Host = "broker.example.com",
    Port = 8883,
    UseTls = true,
    TlsTargetHost = "broker.internal",            // SNI, when it differs from Host
    ClientCertificates = certificates,            // mutual TLS
    ServerCertificateValidation = (s, c, ch, e) => /* pinning, custom roots */ true,
});

WebSocket

From the Pulse.Mqtt.Transport.WebSocket package:

.UseTransportFactory(_ => new WebSocketTransportFactory(new WebSocketTransportOptions
{
    Uri = new Uri("wss://broker.example.com/mqtt"),
    // SubProtocol defaults to "mqtt".
}))

Through a reverse proxy

Headers and Proxy are first-class — handy for a broker behind a reverse proxy or gateway that expects an Authorization token or a routing header, reached through a corporate proxy:

new WebSocketTransportOptions
{
    Uri = new Uri("wss://gateway.example.com/mqtt"),
    Headers = new Dictionary<string, string>
    {
        ["Authorization"] = $"Bearer {token}",   // gateway authenticates the upgrade
        ["X-Tenant"] = "acme",                    // routing header the proxy keys on
    },
    Proxy = new WebProxy("http://corp-proxy:8080"),
}

ConfigureClient is still there for anything these do not cover (client certificates, cookies, custom keep-alive); it runs last, so it can override Headers and Proxy.

QUIC

From the Pulse.Mqtt.Transport.Quic package (.NET 10, brokers with a QUIC listener such as EMQX):

.UseTransportFactory(_ => new QuicTransportFactory(new QuicTransportOptions
{
    Host = "broker.example.com",
    // Port defaults to 14567, ALPN to "mqtt"; TLS 1.3 is always on.
}))

QUIC needs OS and msquic support — check QuicTransportFactory.IsSupported and fall back to TCP or WebSocket when it is false. Details and platform notes: QUIC transport.

Credentials and identity

options.ClientId = "my-service";        // required
options.Username = "device-42";
options.Password = "secret";
options.CleanStart = true;              // false resumes a broker-side session

Direct construction exposes the full CONNECT packet — will messages, session expiry, receive maximum, user properties:

var connect = new MqttConnectPacket
{
    ClientId = "my-service",
    KeepAliveSeconds = 30,
    CleanStart = false,
    Username = "device-42",
    Password = Encoding.UTF8.GetBytes("secret"),
};

The same packet template is used for every reconnection.

Enhanced authentication (MQTT 5)

For challenge/response schemes — SCRAM, OAuth-style token exchanges, Kerberos — implement one small interface and hand it to the client:

public sealed class ScramAuthenticator : IMqttAuthenticator
{
    public string Method => "SCRAM-SHA-256";

    public ValueTask<ReadOnlyMemory<byte>?> NextDataAsync(
        ReadOnlyMemory<byte>? challenge, CancellationToken ct)
    {
        // null challenge → produce the initial data for CONNECT (or a re-auth start);
        // otherwise → answer the broker's challenge.
        return ValueTask.FromResult<ReadOnlyMemory<byte>?>(ComputeNextStep(challenge));
    }
}

new ResilientMqttClientOptions
{
    Connect = connect,
    Raw = new RawMqttClientOptions { Authenticator = new ScramAuthenticator() },
}

The client carries the method and initial data on CONNECT, answers every broker AUTH challenge through the authenticator until the broker concludes with a CONNACK, and supports client-initiated re-authentication on a live connection:

await client.ReAuthenticateAsync(token);   // e.g. when a token rotates

A throwing authenticator fails the attempt like any connect failure; with no authenticator configured, no AUTH is ever sent and a broker that starts an exchange anyway is a protocol error.

MQTT 5 only

Enhanced authentication and live re-authentication use the MQTT 5 AUTH exchange. Configuring RawMqttClientOptions.Authenticator with MqttProtocolVersion.V311, or calling ReAuthenticateAsync on an MQTT 3.1.1 session, throws NotSupportedException. MQTT 3.1.1 supports only the CONNECT username/password credential fields.

Protocol version

MQTT 5.0 is the default. For brokers that only speak 3.1.1:

options.ProtocolVersion = MqttProtocolVersion.V311;

The codec implements both protocol versions. MQTT 5-only features are not available on a 3.1.1 session: properties cannot travel on the wire, packet codecs reject MQTT 5-only properties instead of silently dropping them, request/response helpers and enhanced authentication fail fast, and broker-negotiated limits such as receive maximum or topic aliases are absent. See MQTT protocol compatibility for the full feature matrix.

Keep-alive

KeepAliveSeconds (default 60, 0 disables) drives a PINGREQ loop when the connection is idle. A missing PINGRESP within RawMqttClientOptions.PingResponseTimeout faults the connection, which triggers the reconnect cycle. Brokers may override the interval via the CONNACK's server keep-alive; Pulse honors it.

Handshake timeouts

All on ResilientMqttClientOptions.Raw:

Setting	Default	Meaning
ConnAckTimeout	30 s	How long to wait for the broker's CONNACK
PingResponseTimeout	30 s	How long to wait for a PINGRESP
AcknowledgementTimeout	30 s	How long to wait for publish/subscribe acknowledgements
InboundMessageCapacity	256	Bound of the received-message queue

Custom transports

Anything that moves bytes can carry MQTT: implement IMqttTransportFactory returning an IMqttTransport (a PipeReader/PipeWriter pair). The in-process test broker is exactly that — see Extending.