.NET MQTT Client Built for AWS Broker Using MQTTnet Library

After a lot of trial and error I’m excited to share a finished MQTT client class. This serves as lightweight IoT communication for a medical grade device, hence the emphasis on security. I wasn’t able to find a lot of examples online aside from the MQTTnet documentation itself, and even that had limited information.

public MqttClient(string mqttEndpoint, int mqttPort, X509Certificate2 caCert, X509Certificate2 deviceCert, string clientId)
{

    certChain = new X509Certificate2Collection
    {
        caCert,
        deviceCert
    };

    this.mqttEndpoint = mqttEndpoint;
    this.mqttPort = mqttPort;
    this.clientId = clientId;

    mqttFactory = new MqttFactory();
    mqttClient = mqttFactory.CreateMqttClient();

    //EVENT HANDLERS
    mqttClient.ApplicationMessageReceivedAsync += Client_MqttMsgPublishReceived;
    mqttClient.DisconnectedAsync += Client_MqttDisconnected;

}

The above constructor uses the client ID as well as the broker endpoint and port, creates a trust chain of certificates to verify the device, and then initiates the client.

public async Task Connect()
{
        var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer(mqttEndpoint)
            .WithClientId(clientId)
            .WithoutPacketFragmentation()
            .WithTlsOptions(
                o =>
                {
                    o.WithClientCertificates(certChain);
                    o.WithTrustChain(certChain);
                    o.UseTls();
                    o.WithSslProtocols(SslProtocols.Tls12);
                    o.Build();
                })
            .Build();

        using (var timeout = new CancellationTokenSource(5000))
        {
            await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token);

            Trace.WriteLine($"Connected to AWS IoT with client id: {clientId}.");
        }

}

We then apply the connection options to the client in the Connect() method above. Note that I specify the use of Tls 1.2 and that I also included WithoutPacketFragmentation() which is required when communicating with an AWS IoT broker.

public async Task SubscribeToGetDeviceToken()
{
    var toDeviceTokenTopic = IoTConfig.GetMqttClientIdToDeviceTokenTopic(clientId);

    var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
        .WithTopicFilter(
            f =>
            {
                f.WithTopic(toDeviceTokenTopic);
                f.WithAtLeastOnceQoS();
            })
        .Build();

    await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);

    Trace.WriteLine($"{clientId} subscribed to topic {toDeviceTokenTopic}");
}

Token retrieval is essential to communicate freely with the broker, and will even allow easy HTTP communication. First step is to subscribe to the specified topic as shown in the above SubscribeToGetDeviceToken() method.

public async Task PublishToGetToken()
{
    var fromDeviceTokenTopic = IoTConfig.GetMqttClientIdFromDeviceTokenTopic(clientId);

    //var message = "{}";
    var applicationMessage = new MqttApplicationMessageBuilder()
        .WithTopic(fromDeviceTokenTopic)
        .WithPayload("{}")
        .Build();

    await mqttClient.PublishAsync(applicationMessage, CancellationToken.None);

    Trace.WriteLine($"Published: topic: {fromDeviceTokenTopic} to get token");
}

The last step is to extract the token from the message received from the broker. If you scroll back up to the constructor you will notice a Client_MqttMsgPublishReceived method called when a response is received:

private Task Client_MqttMsgPublishReceived(MqttApplicationMessageReceivedEventArgs e)
{
    Trace.WriteLine($"Message recieved from topic {e.ApplicationMessage.Topic} with data: " + JObject.Parse(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment)));

    Trace.WriteLine(e.ApplicationMessage.Topic);

    if (e.ApplicationMessage.Topic != null)
    {
        Trace.WriteLine("Fetching JWT Token");
        accessToken = FindJWTTokenInMessage(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment));
        Trace.WriteLine($"Token: {accessToken}");
    }
    else
    {
        Trace.WriteLine("Unable to fetch JWT Token successfully");
    }

    return Task.FromResult<object>(null);
}

This last method uses the NewtonSoft JSON library to assist, and I modified it so I don’t expose the security of the application I’m working on to my personal website, so if the handling seems a little goofy that is why.

Hope you enjoyed this. I look forward to sharing more from this project!