C# SDK を使った接続

C# は、Microsoft が提供する オブジェクト指向 プログラミング言語で、.NET Framework 上で動作します。

本記事では、C# プロジェクトで paho.mqtt.m2mqtt クライアントライブラリを使用し、クライアントから MQTT ブローカーへの接続、サブスクライブ、パブリッシュおよびメッセージ受信の方法を主に紹介します。

前提条件

C# クライアントを使って EMQX Cloud に接続する前に、以下の前提条件を満たしていることを確認してください。

• MQTT ブローカーのデプロイ
• .NET バージョンの確認
• MQTT クライアントの選択
• M2Mqtt のインストール

MQTT ブローカーのデプロイ

アプリケーションをEMQX Cloudに接続するには、デプロイメントを作成して設定する必要があります。

サーバレスデプロイメント

1. EMQX Cloudコンソールでサーバレスデプロイメントを作成します。

2. デプロイメントが作成されて稼働したら、デプロイメントの概要ページに移動し、以下を含むMQTT接続情報を確認します。

   • ブローカーアドレス

   • ポート番号（サーバレスではTLSポートのみ対応）

3. サーバレスデプロイメントはTLS接続が必須です。概要ページからCA証明書をダウンロードし、TLS用のポート8883を使用してください。

4. デプロイメントのアクセス制御 -> クライアント認証で、デフォルト認証（ユーザー名／パスワード）を設定します。

詳細はサーバレスポートガイドを参照してください。

Dedicated Flex または BYOC デプロイメント

1. EMQX CloudコンソールでDedicated FlexまたはBYOCのデプロイメントを作成できます。

2. 作成後、デプロイメントの概要ページに移動し、以下を含むMQTT接続情報を取得します。

   • ブローカーアドレス

   • MQTTおよびWebSocket用のTCPおよびTLSポート番号（TCPおよびTLS接続の両方に対応）

3. デプロイメントのアクセス制御 -> クライアント認証で、デフォルト認証（ユーザー名／パスワード）を設定します。

詳細なポート設定については、Dedicated & BYOCポートガイドをご覧ください。

.NET バージョンの確認

本プロジェクトは .NET 5.0 を使用して開発およびテストしています。以下のコマンドで .NET のバージョンを確認できます。

~ dotnet --version            
5.0.301

MQTT クライアントの選択

paho.mqtt.m2mqtt は、MQTT v3.1 および v3.1.1 の両方に対応した、すべての .NET プラットフォームで利用可能な MQTT クライアントです。

.NET CLI を使った M2Mqtt のインストール

プロジェクトのルートディレクトリで以下のコマンドを実行し、M2Mqtt をインストールします。

dotnet add package M2Mqtt --version 4.3.0

接続

コンソールのデプロイ概要で該当のアドレスとポート情報を確認してください。ベーシックエディションの場合、ポートは 1883 または 8883 ではないため、必ずポート番号を確認してください。

MQTT ブローカーへの接続

本記事では、EMQX が提供する 無料のパブリック MQTT ブローカー を使用します。このサービスは MQTT IoT クラウドプラットフォーム を基に作成されています。ブローカーの接続情報は以下の通りです。

• ブローカー: broker.emqx.io
• TCP ポート: 1883
• WebSocket ポート: 8083

M2Mqtt のインポート

using uPLibrary.Networking.M2Mqtt;

MQTT ブローカー接続のパラメータ設定

MQTT ブローカー接続のアドレス、ポート、トピックを設定します。同時に、C# の Guid.NewGuid() を呼び出して、MQTT クライアント ID としてランダムな uid を生成します。

string broker = "broker.emqx.io";
int port = 1883;
string topic = "Csharp/mqtt";
string clientId = Guid.NewGuid().ToString();
// ブローカーで認証が必要な場合は、ユーザー名とパスワードを設定
string username = "emqx";
string password = "public";

MQTT 接続メソッドの作成

静的クラスメソッド ConnectMQTT を作成し、MQTT クライアントを生成して指定したブローカーに接続します。クライアントの IsConnected プロパティで接続の成否を判定できます。最後にクライアントを返します。

static MqttClient ConnectMQTT(string broker, int port, string clientId, string username, string password)
{
    MqttClient client = new MqttClient(broker, port, false, MqttSslProtocols.None, null, null);
    client.Connect(clientId, username, password);
    if (client.IsConnected)
    {
        Console.WriteLine("Connected to MQTT Broker");
    }
    else
    {
        Console.WriteLine("Failed to connect");
    }
    return client;
}

メッセージのパブリッシュ

while ループを定義し、その中で MQTT クライアントの Publish メソッドを使って、指定したトピックに毎秒メッセージをパブリッシュします。

static void Publish(MqttClient client, string topic)
{
    int msg_count = 0;
    while (true)
    {
        System.Threading.Thread.Sleep(1*1000);
        string msg = "messages: " + msg_count.ToString();
        client.Publish(topic, System.Text.Encoding.UTF8.GetBytes(msg));
        Console.WriteLine("Send `{0}` to topic `{1}`", msg, topic);
        msg_count++;
    }
}

トピックのサブスクライブ

静的メソッド client_MqttMsgPublishReceived を作成します。このメソッドはクライアントが MQTT ブローカーからメッセージを受信した際に呼び出されます。受信したメッセージのトピックとペイロードをコンソールに出力します。

static void Subscribe(MqttClient client, string topic)
{
    client.MqttMsgPublishReceived += client_MqttMsgPublishReceived;
    client.Subscribe(new string[] { topic }, new byte[] { MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE });
}
static void client_MqttMsgPublishReceived(object sender, MqttMsgPublishEventArgs e)
{
    string payload = System.Text.Encoding.Default.GetString(e.Message);
    Console.WriteLine("Received `{0}` from `{1}` topic", payload, e.Topic.ToString());
}

完全なコード例

using System;
using uPLibrary.Networking.M2Mqtt;
using uPLibrary.Networking.M2Mqtt.Messages;

namespace csharpMQTT
{
    class Program
    {
        static MqttClient ConnectMQTT(string broker, int port, string clientId, string username, string password)
        {
            MqttClient client = new MqttClient(broker, port, false, MqttSslProtocols.None, null, null);
            client.Connect(clientId, username, password);
            if (client.IsConnected)
            {
                Console.WriteLine("Connected to MQTT Broker");
            }
            else
            {
                Console.WriteLine("Failed to connect");
            }
            return client;
        }

        static void Publish(MqttClient client, string topic)
        {
            int msg_count = 0;
            while (true)
            {
                System.Threading.Thread.Sleep(1*1000);
                string msg = "messages: " + msg_count.ToString();
                client.Publish(topic, System.Text.Encoding.UTF8.GetBytes(msg));
                Console.WriteLine("Send `{0}` to topic `{1}`", msg, topic);
                msg_count++;
            }
        }

        static void Subscribe(MqttClient client, string topic)
        {
            client.MqttMsgPublishReceived += client_MqttMsgPublishReceived;
            client.Subscribe(new string[] { topic }, new byte[] { MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE });
        }
        static void client_MqttMsgPublishReceived(object sender, MqttMsgPublishEventArgs e)
        {
            string payload = System.Text.Encoding.Default.GetString(e.Message);
            Console.WriteLine("Received `{0}` from `{1}` topic", payload, e.Topic.ToString());
        }

        static void Main(string[] args)
        {
            string broker = "broker.emqx.io";
            int port = 1883;
            string topic = "Csharp/mqtt";
            string clientId = Guid.NewGuid().ToString();
            string username = "emqx";
            string password = "public";
            MqttClient client = ConnectMQTT(broker, port, clientId, username, password);
            Subscribe(client, topic);
            Publish(client, topic);
        }
    }
}

テスト

コードを実行すると、コンソールに以下のように出力されます。