C# MQTT using MQTTnet

前言

最近需要透過 MQTT 來收取 IOT Sensor 的資料，
所以到 MQTT.org 看一下有那些軟體可以當 MQTT Broker 及 MQTT Client。
有人使用 mosquitto，也有人使用 RabbitMQ with MQTT Plugin。
而 .NET 則可以使用 MQTTnet。
以下本文將使用 MQTTnet 來測試 Sensor 送資料到 Broker ，
再由 Backend ControlCenter 將訂閱的 Topic 訊息讀取出來，
最後再讀取來自 mosquitto 的訊息。

實作

建立 MQBroker, MQControlCenter 及 MQSensor 3 個 Console 專案，

MQBroker 專案

1.加入 MQTTnet 套件
要建立 MQTT Server，所以加入 MQTTnet 套件

2.建立 MQTT Server，並啟動它
Program.cs 中建立 MQTT Server，並在 InterceptingPublishAsync 事件 Log 送進來的訊息

using MQTTnet.Server;
using MQTTnet;
using System.Text;
using static System.Console;

// 預設 port 是 1883
var options = new MqttServerOptionsBuilder()
    //Set endpoint to localhost
    .WithDefaultEndpoint();

// Create a new mqtt server
var server = new MqttFactory().CreateMqttServer(options.Build());
//Add Interceptor for logging incoming messages
server.InterceptingPublishAsync += Server_InterceptingPublishAsync;

WriteLine("Start MQTTnet Server ...");
// Start the server
await server.StartAsync();
WriteLine("Press any key to stop Server ...");
ReadKey();

Task Server_InterceptingPublishAsync(InterceptingPublishEventArgs arg)
{
    // Convert Payload to string
    var payload = arg.ApplicationMessage?.Payload == null ? null : Encoding.UTF8.GetString(arg.ApplicationMessage?.Payload);


    WriteLine(
        " TimeStamp: {0} -- Message: ClientId = {1}, Topic = {2}, Payload = {3}, QoS = {4}, Retain-Flag = {5}",

        DateTime.Now,
        arg.ClientId,
        arg.ApplicationMessage?.Topic,
        payload,
        arg.ApplicationMessage?.QualityOfServiceLevel,
        arg.ApplicationMessage?.Retain);
    return Task.CompletedTask;
}

MQControlCenter 專案

1.加入 MQTTnet.Extensions.ManagedClient 套件
使用 MQTT Client，所以加入 MQTTnet.Extensions.ManagedClient 套件

2.建立 MQTT Client 並設定要訂閱的 Topic
在 Program.cs 中建立 MQTT Client 並設定要訂閱的 Topic Filter，
可使用萬用字元 + or # ，

在 ApplicationMessageReceivedAsync 事件中取得 Payload。

• 註:因為要取得多個 Sensor 的資料，所以 Topic 使用 **home/temperature/#**，所以可以取得 Topic 開頭為 home/temperature/ 的訊息

using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet;
using System.Text.Json;
using static System.Console;
using System.Text;
using MQTTnet.Packets;

var _mqttClient = new MqttFactory().CreateManagedMqttClient();

var clientId = "controlCenter";
var server = "localhost";
var builder = new MqttClientOptionsBuilder()
                .WithClientId(clientId)
                .WithTcpServer(server);

var options = new ManagedMqttClientOptionsBuilder()
                    .WithAutoReconnectDelay(TimeSpan.FromSeconds(60))
                    .WithClientOptions(builder.Build())
                    .Build();
// Set up handlers
_mqttClient.ConnectedAsync += _mqttClient_ConnectedAsync;
_mqttClient.DisconnectedAsync += _mqttClient_DisconnectedAsync;
_mqttClient.ConnectingFailedAsync += _mqttClient_ConnectingFailedAsync;
_mqttClient.ApplicationMessageReceivedAsync += _mqttClient_ApplicationMessageReceivedAsync;

var topicFilters = new List<MqttTopicFilter> { new MqttTopicFilter { Topic= "home/temperature/#" } };
await _mqttClient.SubscribeAsync(topicFilters);

WriteLine($"{clientId} 連線到 MQTT Broker ({server}) ....");
await _mqttClient.StartAsync(options);
WriteLine("Press any key to stop receive message ...");
ReadKey();

Task _mqttClient_ConnectedAsync(MqttClientConnectedEventArgs arg)
{
    WriteLine("Connected");
    return Task.CompletedTask;
};
Task _mqttClient_DisconnectedAsync(MqttClientDisconnectedEventArgs arg)
{
    WriteLine("Disconnected");
    return Task.CompletedTask;
};
Task _mqttClient_ConnectingFailedAsync(ConnectingFailedEventArgs arg)
{
    WriteLine("Connection failed check network or broker!");
    return Task.CompletedTask;
}

Task _mqttClient_ApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
{
    var topic = arg?.ApplicationMessage?.Topic;
    var payloadText = Encoding.UTF8.GetString(
            arg?.ApplicationMessage?.Payload ?? Array.Empty<byte>());

    WriteLine($"Received: Topic:{topic}, Payload:{payloadText}");
    return Task.CompletedTask;
}

MQSensor 專案

1.加入 MQTTnet.Extensions.ManagedClient 套件
使用 MQTT Client，所以加入 MQTTnet.Extensions.ManagedClient 套件

2.建立 MQTT Client 並設定要發送的 Topic 及 Payload
在 Program.cs 中建立 MQTT Client 並設定要發送的 Topic 及 Payload

using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet;
using System.Text.Json;
using static System.Console;

var _mqttClient = new MqttFactory().CreateManagedMqttClient();

var clientId = "sensor1";
var server = "localhost";
var builder = new MqttClientOptionsBuilder()
                .WithClientId(clientId)
                .WithTcpServer(server);
var options = new ManagedMqttClientOptionsBuilder()
                    .WithAutoReconnectDelay(TimeSpan.FromSeconds(60))
                    .WithClientOptions(builder.Build())
                    .Build();
// Set up handlers
_mqttClient.ConnectedAsync += _mqttClient_ConnectedAsync;
_mqttClient.DisconnectedAsync += _mqttClient_DisconnectedAsync;
_mqttClient.ConnectingFailedAsync += _mqttClient_ConnectingFailedAsync;

WriteLine($"{clientId} 連線到 MQTT Broker ({server}) ....");
await _mqttClient.StartAsync(options);

while (true)
{
    WriteLine("請輸入Topic, Like(home/temperature/sensor-1):");
    var topic = ReadLine();
    if(string.IsNullOrWhiteSpace(topic))
        topic=$"home/temperature/{clientId}";
    WriteLine("請輸入要傳送的訊息:");
    var message = ReadLine();
    if (!string.IsNullOrWhiteSpace(message))
    {
        var json = JsonSerializer.Serialize(new { message, sent = DateTime.UtcNow });
        await _mqttClient.EnqueueAsync(topic, json);

        // Wait until the queue is fully processed.
        SpinWait.SpinUntil(() => _mqttClient.PendingApplicationMessagesCount == 0, 10000);
        WriteLine($"Pending messages = {_mqttClient.PendingApplicationMessagesCount}");
    }
    else
    {
        WriteLine("訊息不可為空，請重新輸入!");
    }
}
Task _mqttClient_ConnectedAsync(MqttClientConnectedEventArgs arg)
{
    WriteLine("Connected");
    return Task.CompletedTask;
};
Task _mqttClient_DisconnectedAsync(MqttClientDisconnectedEventArgs arg)
{
    WriteLine("Disconnected");
    return Task.CompletedTask;
};
Task _mqttClient_ConnectingFailedAsync(ConnectingFailedEventArgs arg)
{
    WriteLine("Connection failed check network or broker!");
    return Task.CompletedTask;
}

測試

程式寫好，分別將 MQBroker.exe, MQControlCenter.exe 及 MQSensor.exe 執行起來，
在 MQSensor.exe Console 中輸入 Topic 及 內容，
可以發現在 MQBroker.exe 會 Log 訊息，
在 MQControlCenter.exe 也可以接收到訂閱的訊息(sensor-1, sensor-2 及 sensor-3)。

測試接收 mosquitto 訊息

MQControlCenter.exe 也可以接收 mosquitto 的訊息。
先到 mosquitto Download 下載安裝檔，
安裝完成後，直接開啟 mosquitto.exe 會啟動 MQTT Broker Server，
透過 mosquitto_pub 來發佈訊息 mosquitto_pub -t "home/temperature/sensor1" -m "hello world"，
透過 MQControlCenter.exe 則可以收到訂閱的訊息，如下，

• 註: 第三個發佈訊息因為 Topic 不符合訂閱的 Topic Filter (home/temperature/#)，所以不會被收到

參考資料

MQTT
MQTTnet
C# MQTT broker using MQTTnet version 4
MQTT Client with MQTTnet 4 and C#
mosquitto
RabbitMQ with MQTT Plugin