Using MQTT

WARNING

Wolverine requires the V5 version of MQTT for its broker support

The Wolverine 1.9 release added a new transport option for the MQTT standard common in IoT Messaging.

Installing

To use MQTT as a transport with Wolverine, first install the WolverineFx.Mqtt library via nuget to your project. Behind the scenes, this package uses the MQTTnet managed library for accessing MQTT brokers and also for its own testing.

dotnet add package WolverineFx.Mqtt

In its most simplistic usage you enable the MQTT transport through calling the WolverineOptions.UseMqtt() extension method and defining which MQTT topics you want to publish or subscribe to with the normal subscriber rules as shown in this sample:

var builder = Host.CreateApplicationBuilder();

builder.UseWolverine(opts =>
{
    // Connect to the MQTT broker
    opts.UseMqtt(mqtt =>
    {
        var mqttServer = builder.Configuration["mqtt_server"];

        mqtt
            .WithMaxPendingMessages(3)
            .WithClientOptions(client => { client.WithTcpServer(mqttServer); });
    });

    // Listen to an MQTT topic, and this could also be a wildcard
    // pattern
    opts.ListenToMqttTopic("app/incoming")
        // In the case of receiving JSON data, but
        // not identifying metadata, tell Wolverine
        // to assume the incoming message is this type
        .DefaultIncomingMessage<Message1>()

        // The default is AtLeastOnce
        .QualityOfService(MqttQualityOfServiceLevel.AtMostOnce);

    // Publish messages to an outbound topic
    opts.PublishAllMessages()
        .ToMqttTopic("app/outgoing");
});

using var host = builder.Build();
await host.StartAsync();

INFO

The MQTT transport at this time only supports endpoints that are either Buffered or Durable.

WARNING

The MQTT transport does not really support the "Requeue" error handling policy in Wolverine. "Requeue" in this case becomes effectively an inline "Retry"

Connecting to Multiple MQTT Brokers

If a single Wolverine application needs to talk to more than one MQTT broker, register the additional broker(s) with AddNamedMqttBroker using a BrokerName, then pin publishing or listening to a specific broker with the *OnNamedBroker overloads:

opts.UseMqtt(mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("primary-broker")));

// An additional, independent MQTT broker identified by name
opts.AddNamedMqttBroker(new BrokerName("secondary"),
    mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("secondary-broker")));

// Publish a message type to a topic on a named broker
opts.PublishMessage<OrderPlaced>()
    .ToMqttTopicOnNamedBroker(new BrokerName("secondary"), "orders");

// Listen to a topic on a named broker
opts.ListenToMqttTopicOnNamedBroker(new BrokerName("secondary"), "orders");

INFO

The Wolverine Uri scheme for any endpoint on a named broker is the broker name itself, so in the example above you would see endpoint URIs like secondary://topic/orders. The default broker keeps the canonical mqtt:// scheme, which keeps the two brokers' endpoints from colliding.

Each named broker is a completely separate IManagedMqttClient with its own per-node response topic, so request/reply works independently over each broker.

Connecting to multiple named brokers is distinct from broker-per-tenant multi-tenancy: a named broker is a statically-addressed second connection that you target explicitly, whereas per-tenant connections are selected at runtime from each message's tenant id.

Broadcast to User Defined Topics

As long as the MQTT transport is enabled in your application, you can explicitly publish messages to any named topic through this usage:

public static async Task broadcast(IMessageBus bus)
{
    var paymentMade = new PaymentMade(200, "EUR");
    await bus.BroadcastToTopicAsync("region/europe/incoming", paymentMade);
}

^{snippet source | anchor}

Publishing to Derived Topic Names

INFO

The Wolverine is open to extending the options for determining the topic name from the message type, but is waiting for feedback from the community before trying to build anything else around this.

As a way of routing messages to MQTT topics, you also have this option:

var builder = Host.CreateApplicationBuilder();

builder.UseWolverine(opts =>
{
    // Connect to the MQTT broker
    opts.UseMqtt(mqtt =>
    {
        var mqttServer = builder.Configuration["mqtt_server"];

        mqtt
            .WithMaxPendingMessages(3)
            .WithClientOptions(client => { client.WithTcpServer(mqttServer); });
    });

    // Publish messages to MQTT topics based on
    // the message type
    opts.PublishAllMessages()
        .ToMqttTopics()
        .QualityOfService(MqttQualityOfServiceLevel.AtMostOnce);
});

using var host = builder.Build();
await host.StartAsync();

^{snippet source | anchor}

In this approach, all messages will be routed to MQTT topics. The topic name for each message type would be derived from either Wolverine's message type name rules or by using the [Topic("topic name")] attribute as shown below:

[Topic("one")]
public class TopicMessage1;

^{snippet source | anchor}

[Topic("color.blue")]
public class FirstMessage
{
    public Guid Id { get; set; } = Guid.NewGuid();
}

^{snippet source | anchor}

Publishing by Topic Rules

You can publish messages to MQTT topics based on user defined logic to determine the actual topic name.

As an example, say you have a marker interfaces for your messages like this:

public interface ITenantMessage
{
    string TenantId { get; }
}

^{snippet source | anchor}

To publish any message implementing that interface to an MQTT topic, you could specify the topic name logic like this:

var builder = Host.CreateApplicationBuilder();

builder.UseWolverine(opts =>
{
    // Connect to the MQTT broker
    opts.UseMqtt(mqtt =>
    {
        var mqttServer = builder.Configuration["mqtt_server"];

        mqtt
            .WithMaxPendingMessages(3)
            .WithClientOptions(client => { client.WithTcpServer(mqttServer); });
    });

    // Publish any message that implements ITenantMessage to
    // MQTT with a topic derived from the message
    opts.PublishMessagesToMqttTopic<ITenantMessage>(m => $"{m.GetType().Name.ToLower()}/{m.TenantId}")

        // Specify or configure sending through Wolverine for all
        // MQTT topic broadcasting
        .QualityOfService(MqttQualityOfServiceLevel.ExactlyOnce)
        .BufferedInMemory();
});

using var host = builder.Build();
await host.StartAsync();

^{snippet source | anchor}

Listening by Topic Filter

Wolverine supports topic filters for listening. The syntax is still just the same ListenToMqttTopic(filter) as shown in this snippet from the Wolverine.MQTT test suite:

_receiver = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseMqttWithLocalBroker(port);
        opts.ListenToMqttTopic("incoming/one", "group1").RetainMessages();
    }).StartAsync();

^{snippet source | anchor}

_receiver = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseMqttWithLocalBroker(port);
        opts.ListenToMqttTopic("incoming/#").RetainMessages();
    }).StartAsync();

^{snippet source | anchor}

In the case of receiving any message that matches the topic filter according to the MQTT topic filter rules, that message will be handled by the listening endpoint defined for that filter.

Integrating with Non-Wolverine

It's quite likely that in using Wolverine with an MQTT broker that you will be communicating with non-Wolverine systems or devices on the other end, so you can't depend on the Wolverine metadata being sent in MQTT UserProperties data. Not to worry, you've got options.

In the case of the external system sending you JSON, but nothing else, if you can design the system such that there's only one type of message coming into a certain MQTT topic, you can just tell Wolverine to listen for that topic and what that message type would be so that Wolverine is able to deserialize the message and relay that to the correct message handler like so:

var builder = Host.CreateApplicationBuilder();
builder.UseWolverine(opts =>
{
    // Connect to the MQTT broker
    opts.UseMqtt(mqtt =>
    {
        var mqttServer = builder.Configuration["mqtt_server"];

        mqtt
            .WithMaxPendingMessages(3)
            .WithClientOptions(client => { client.WithTcpServer(mqttServer); });
    });

    // Listen to an MQTT topic, and this could also be a wildcard
    // pattern
    opts.ListenToMqttTopic("app/payments/made")
        // In the case of receiving JSON data, but
        // not identifying metadata, tell Wolverine
        // to assume the incoming message is this type
        .DefaultIncomingMessage<PaymentMade>();
});

using var host = builder.Build();
await host.StartAsync();

^{snippet source | anchor}

For more complex interoperability, you can implement the IMqttEnvelopeMapper interface in Wolverine to map between incoming and outgoing MQTT messages and the Wolverine Envelope structure. Here's an example:

public class MyMqttEnvelopeMapper : IMqttEnvelopeMapper
{
    public void MapEnvelopeToOutgoing(Envelope envelope, MqttApplicationMessage outgoing)
    {
        // This is the only absolutely mandatory item
        outgoing.PayloadSegment = envelope.Data!;

        // Maybe enrich this more?
        outgoing.ContentType = envelope.ContentType;
    }

    public void MapIncomingToEnvelope(Envelope envelope, MqttApplicationMessage incoming)
    {
        // These are the absolute minimums necessary for Wolverine to function
        envelope.MessageType = typeof(PaymentMade).ToMessageTypeName();
        envelope.Data = incoming.PayloadSegment.Array;

        // Optional items
        envelope.DeliverWithin = 5.Seconds(); // throw away the message if it
        // is not successfully processed
        // within 5 seconds
    }
}

^{snippet source | anchor}

And apply that to an MQTT topic like so:

var builder = Host.CreateApplicationBuilder();

builder.UseWolverine(opts =>
{
    // Connect to the MQTT broker
    opts.UseMqtt(mqtt =>
    {
        var mqttServer = builder.Configuration["mqtt_server"];
        
        mqtt
            .WithMaxPendingMessages(3)
            .WithClientOptions(client => { client.WithTcpServer(mqttServer); });
    });

    // Publish messages to MQTT topics based on
    // the message type
    opts.PublishAllMessages()
        .ToMqttTopics()

        // Tell Wolverine to map envelopes to MQTT messages
        // with our custom strategy
        .UseInterop(new MyMqttEnvelopeMapper())
        .QualityOfService(MqttQualityOfServiceLevel.AtMostOnce);

});

using var host = builder.Build();
await host.StartAsync();

^{snippet source | anchor}

Clearing Out Retained Messages

MQTT brokers allow you to publish retained messages to a topic, meaning that the last message will always be retained by the broker and sent to any new subscribers. That's a little bit problematic if your Wolverine application happens to be restarted, because that last retained message may easily be resent to your Wolverine application when you restart.

Not to fear, the MQTT protocol allows you to "clear" out a topic by sending it a zero byte message, and Wolverine has a couple shortcuts for doing just that by returning a cascading message to "zero out" the topic a message was received on or a named topic like this:

public static AckMqttTopic Handle(ZeroMessage message)
{
    // "Zero out" the topic that the original message was received from
    return new AckMqttTopic();
}

public static ClearMqttTopic Handle(TriggerZero message)
{
    // "Zero out" the designated topic
    return new ClearMqttTopic("red");
}

^{snippet source | anchor}

Authentication via OAuth2

Wolverine supports MQTT v5 OAuth2/JWT authentication by supplying a token callback and refresh interval when you configure the transport. The callback returns raw token bytes (use UTF-8 encoding if your token is a string). When configured, Wolverine sets the MQTT authentication method to OAUTH2-JWT, sends the initial token with the connect packet, and re-authenticates on the configured refresh period while the client is connected.

INFO

You don't need to configure AuthenticationMethod and AuthenticationData by yourself. These are overriden when the MqttJwtAuthenticationOptions parameter is set.

Minimal configuration example:

var builder = Host.CreateApplicationBuilder();

builder.UseWolverine(opts =>
{
    opts.UseMqtt(
        mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("broker")),
        new MqttJwtAuthenticationOptions(
            async () => Encoding.UTF8.GetBytes(await GetJwtTokenAsync()),
            30.Minutes()));
});

Broker per Tenant

Named brokers (above) are a static topology: you pin specific endpoints to a specific broker by name at configuration time. Broker-per-tenant is different — it is runtime routing. You declare one shared topic topology, and each tenant is served by its own dedicated MQTT connection (a distinct broker). Which connection a message goes to (and which connection an inbound message came from) is decided at runtime by the message's tenant id, typically set through DeliveryOptions.TenantId:

opts.UseMqtt(mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("shared-broker")))

    // How should Wolverine route a message whose TenantId is null or unknown?
    // FallbackToDefault (the default) uses the shared connection; TenantIdRequired
    // throws; IgnoreUnknownTenants silently drops it.
    .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault)

    // Each tenant gets its OWN dedicated MQTT connection, but shares the
    // topic topology declared below.
    .AddTenant("tenant-west",
        mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("west-broker")))

    .AddTenant("tenant-east",
        mqtt => mqtt.WithClientOptions(client => client.WithTcpServer("east-broker")));

// One shared topology; messages are routed to the right connection at runtime by
// Envelope.TenantId (e.g. new DeliveryOptions { TenantId = "tenant-west" }).
opts.PublishMessage<OrderPlaced>().ToMqttTopic("orders");
opts.ListenToMqttTopic("orders");

To route a specific message to a tenant's connection, stamp the tenant id on the send:

await bus.SendAsync(new OrderPlaced("blue"), new DeliveryOptions { TenantId = "tenant-west" });

Wolverine wraps the outbound endpoint in a TenantedSender that dispatches on Envelope.TenantId, and builds a compound listener that runs one subscription per tenant connection — each inbound envelope is stamped with the tenant id of the connection it was consumed from. This mirrors the RabbitMQ and Azure Service Bus broker-per-tenant support.

Unique ClientId per tenant (mandatory)

MQTT brokers forcibly disconnect a second connection that shares a ClientId. Wolverine therefore always gives each tenant connection a unique ClientId derived from the tenant id (<clientId>-tenant-<tenantId>), even if you pre-set one on the tenant options. This is required so tenant connections never kick each other.

Shared topology, isolated broker

Unlike a shared-broker/topic-prefix multi-tenancy model, MQTT tenants use the same topic string on each tenant's own broker — isolation is purely a matter of which broker the tenant's connection points at. One consequence: retained messages are per-broker, so a retained message on one tenant's broker is not visible to any other tenant.

Named broker vs. broker-per-tenant

Use a named broker when a fixed set of endpoints should always talk to a specific broker. Use broker-per-tenant when the same logical endpoints should be transparently routed to a different connection per tenant based on the runtime tenant id. They are independent features and can be combined.

Interoperability

TIP

Also see the more generic Wolverine Guide on Interoperability

The Wolverine MQTT transport supports pluggable interoperability strategies through the Wolverine.MQTT.IMqttEnvelopeMapper interface to map from Wolverine's Envelope structure and MQTT's MqttApplicationMessage structure.

Here's a simple example:

public class MyMqttEnvelopeMapper : IMqttEnvelopeMapper
{
    public void MapEnvelopeToOutgoing(Envelope envelope, MqttApplicationMessage outgoing)
    {
        // This is the only absolutely mandatory item
        outgoing.PayloadSegment = envelope.Data!;

        // Maybe enrich this more?
        outgoing.ContentType = envelope.ContentType;
    }

    public void MapIncomingToEnvelope(Envelope envelope, MqttApplicationMessage incoming)
    {
        // These are the absolute minimums necessary for Wolverine to function
        envelope.MessageType = typeof(PaymentMade).ToMessageTypeName();
        envelope.Data = incoming.PayloadSegment.Array;

        // Optional items
        envelope.DeliverWithin = 5.Seconds(); // throw away the message if it
        // is not successfully processed
        // within 5 seconds
    }
}

^{snippet source | anchor}

You will need to apply that mapper to each endpoint like so:

var builder = Host.CreateApplicationBuilder();

builder.UseWolverine(opts =>
{
    // Connect to the MQTT broker
    opts.UseMqtt(mqtt =>
    {
        var mqttServer = builder.Configuration["mqtt_server"];
        
        mqtt
            .WithMaxPendingMessages(3)
            .WithClientOptions(client => { client.WithTcpServer(mqttServer); });
    });

    // Publish messages to MQTT topics based on
    // the message type
    opts.PublishAllMessages()
        .ToMqttTopics()

        // Tell Wolverine to map envelopes to MQTT messages
        // with our custom strategy
        .UseInterop(new MyMqttEnvelopeMapper())
        .QualityOfService(MqttQualityOfServiceLevel.AtMostOnce);

});

using var host = builder.Build();
await host.StartAsync();

^{snippet source | anchor}

URI reference

The MqttEndpointUri helper class builds canonical endpoint URIs:

URI form	Helper call
mqtt://topic/{name}	MqttEndpointUri.Topic("name")

using Wolverine.MQTT;

var uri = MqttEndpointUri.Topic("sensor/temperature");