Class MqttClient

Namespace

Pamoja.Mqtt

Assembly

Pamoja.Mqtt.dll

An MQTT client transport, the ergonomic facade over the native pamoja core.

public sealed class MqttClient : ILink, IAsyncEnumerable<MqttMessage>, IAsyncDisposable, IDisposable

Inheritance

object

MqttClient

Implements

ILink

IAsyncEnumerable<MqttMessage>

IAsyncDisposable

IDisposable

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

Examples

await using var client = new MqttClient(new MqttClientOptions
{
    ClientId = "sensor-1",
    Host = "localhost",
    Port = 1883,
});
await client.ConnectAsync();
await client.SubscribeAsync("sensors/+/temperature");
await client.PublishAsync("sensors/1/temperature", "21.5");
await foreach (var message in client)
{
    Console.WriteLine($"{message.Topic}: {message.Payload.Length} bytes");
}

Remarks

The native C ABI is synchronous, so every operation runs on the thread pool and is awaited here; failures surface as PamojaException. Construct the client with broker settings, ConnectAsync(), then PublishAsync(string, ReadOnlyMemory<byte>, MqttPublishOptions), SubscribeAsync(string), and read inbound messages with RecvAsync() or by iterating the client with await foreach.

A receive waits apart from the rest of the client, so one task can wait for commands while another publishes readings on the same client.

Constructors

MqttClient(MqttClientOptions)

Creates a disconnected client from the given options.

public MqttClient(MqttClientOptions options)

Parameters

options MqttClientOptions

The broker connection settings.

Exceptions

ArgumentNullException

options is null.

PamojaException

The native client could not be created.

ArgumentOutOfRangeException

A QoS is not one of the Qos values.

ArgumentException

A password comes without a username, or a client certificate without its key.

Methods

ConnectAsync()

Connects to the broker and starts the background event loop.

public Task ConnectAsync()

Returns

Task

A task that completes once connected.

Exceptions

PamojaException

The connection could not be established.

DisconnectAsync()

Closes the connection and stops the background event loop.

public Task DisconnectAsync()

Returns

Task

A task that completes once the client has disconnected.

Exceptions

PamojaException

The disconnect failed.

Dispose()

Releases the native client.

public void Dispose()

DisposeAsync()

Disconnects (best-effort) and releases the native client.

public ValueTask DisposeAsync()

Returns

ValueTask

A task that completes once the client has been released.

GetAsyncEnumerator(CancellationToken)

Iterates incoming messages, so the client can be used with await foreach.

public IAsyncEnumerator<MqttMessage> GetAsyncEnumerator(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Stops iteration when canceled.

Returns

IAsyncEnumerator<MqttMessage>

An async enumerator over incoming messages.

IsConnectedAsync()

Reports whether the client currently holds an active connection.

public Task<bool> IsConnectedAsync()

Returns

Task<bool>

A task resolving to the connection state.

Messages(CancellationToken)

Yields messages from subscribed topics until the connection ends.

public IAsyncEnumerable<MqttMessage> Messages(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Stops iteration when canceled.

Returns

IAsyncEnumerable<MqttMessage>

An async stream over incoming messages.

Remarks

The client is free between messages, so the body of an await foreach can publish, and canceling stops the wait within a quarter of a second. A connection that ends on its own throws its reason out of the loop.

PublishAsync(string, ReadOnlyMemory<byte>, MqttPublishOptions)

Publishes a payload to a topic.

public Task PublishAsync(string topic, ReadOnlyMemory<byte> payload, MqttPublishOptions options = default)

Parameters

topic string

The destination topic.

payload ReadOnlyMemory<byte>

The message body.

options MqttPublishOptions

The quality of service and retain flag for this message.

Returns

Task

A task that completes once the payload is queued for the broker, before the broker acknowledges it; PublishConfirmedAsync(string, ReadOnlyMemory<byte>, MqttPublishOptions) waits for that.

Exceptions

PamojaException

The payload could not be sent.

PublishAsync(string, string, MqttPublishOptions)

Publishes a UTF-8 string payload to a topic.

public Task PublishAsync(string topic, string payload, MqttPublishOptions options = default)

Parameters

topic string

The destination topic.

payload string

The message body, encoded as UTF-8.

options MqttPublishOptions

The quality of service and retain flag for this message.

Returns

Task

A task that completes once the payload is queued for the broker.

Exceptions

PamojaException

The payload could not be sent.

PublishConfirmedAsync(string, ReadOnlyMemory<byte>, MqttPublishOptions)

Publishes a payload to a topic and waits for the broker to acknowledge it.

public Task PublishConfirmedAsync(string topic, ReadOnlyMemory<byte> payload, MqttPublishOptions options = default)

Parameters

topic string

The destination topic.

payload ReadOnlyMemory<byte>

The message body.

options MqttPublishOptions

The quality of service and retain flag for this message.

Returns

Task

A task that completes once the broker holds the message.

Remarks

The broker answers with a PUBACK at AtLeastOnce and a PUBCOMP at ExactlyOnce; at AtMostOnce MQTT acknowledges nothing, so the task completes once the connection has taken the message.

Exceptions

PamojaException

The payload could not be sent, or the connection ended before the acknowledgment, when the message may or may not have arrived.

PublishConfirmedAsync(string, string, MqttPublishOptions)

Publishes a UTF-8 string payload and waits for the broker to acknowledge it.

public Task PublishConfirmedAsync(string topic, string payload, MqttPublishOptions options = default)

Parameters

topic string

The destination topic.

payload string

The message body, encoded as UTF-8.

options MqttPublishOptions

The quality of service and retain flag for this message.

Returns

Task

A task that completes once the broker holds the message.

Exceptions

PamojaException

The payload could not be sent or was not acknowledged.

RecvAsync()

Awaits the next message from any subscribed topic.

public Task<MqttMessage?> RecvAsync()

Returns

Task<MqttMessage>

The next message, or null once the connection has ended.

Exceptions

PamojaException

The client is not connected, or, once, saying why, the connection ended on its own because the broker went away or another client connected with the same id.

RecvAsync(TimeSpan)

Waits a limited time for the next message from any subscribed topic.

public Task<MqttMessage?> RecvAsync(TimeSpan timeout)

Parameters

timeout TimeSpan

How long to wait.

Returns

Task<MqttMessage>

The next message, or null once the connection has ended.

Remarks

When the time runs out nothing is lost: a message arriving afterwards waits for the next receive.

Exceptions

TimeoutException

No message arrived in time.

PamojaException

The client is not connected, or the connection ended on its own.

SendAsync(string, ReadOnlyMemory<byte>)

Publishes a payload to a topic at the client's default quality of service.

public Task SendAsync(string topic, ReadOnlyMemory<byte> payload)

Parameters

topic string

The destination topic.

payload ReadOnlyMemory<byte>

The message body.

Returns

Task

A task that completes once the payload is queued for the broker.

Remarks

This is the SendAsync every link has, so a profile's node or a rule engine publishes over an MQTT client the way it publishes over any other link.

Exceptions

PamojaException

The payload could not be sent.

SendAsync(string, string)

Publishes a UTF-8 string payload to a topic at the client's default quality of service.

public Task SendAsync(string topic, string text)

Parameters

topic string

The destination topic.

text string

The message body, encoded as UTF-8.

Returns

Task

A task that completes once the payload is queued for the broker.

Exceptions

PamojaException

The payload could not be sent.

SubscribeAsync(string)

Subscribes to a topic filter.

public Task SubscribeAsync(string topic)

Parameters

topic string

The topic or wildcard filter to subscribe to.

Returns

Task

A task that completes once the subscription is registered.

Exceptions

PamojaException

The subscription was rejected.