MQTTnet を Unity で使う

cluster では現在 [M2Mqtt](https://github.com/eclipse/paho.mqtt.m2mqtt) を使っているのですが、Async や Rx のサポートがあるライブラリ使いたいお気持ちが発生したので [chkr1011/MQTTnet](https://github.com/chkr1011/MQTTnet) を触ってみたいと思います。

（この記事は[Cluster,Inc. Advent Calendar 2018 - Qiita](https://qiita.com/advent-calendar/2018/cluster) 9日目の記事のはずでした）

# 準備

## インポート

Unity（今回は 2018.3.0f2 を使用）で新規プロジェクトを作成します。
[NuGet Gallery](https://www.nuget.org/packages/MQTTnet/) から MQTTnet 2.8.5 のパッケージをダウンロードし、展開（.nupkg -> .zip にリネームして展開しました）して .NET 4.7.2 向けの DLL ファイルを、プロジェクトにインポートします。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/20631/7424615c-cdb9-5f3e-5f05-d2d58e125523.png)

# 接続

`MqttFactory` から作成したクライアントに、`MqttClientOptionsBuilder` から作成した接続オプションを渡して接続します。
`IMqttClientOptions` に限らず、MQTTnet はオブジェクト作成が概ね Builder パターンで提供されており、M2Mqtt と比べてやたら引数の多いコンストラクタやメソッドが出てこないので精神衛生上よいです。

```csharp

var factory = new MqttFactory();
var mqttClient = factory.CreateMqttClient();

var options = new MqttClientOptionsBuilder()
    .WithTcpServer("broker.hivemq.com")
    .Build();

mqttClient.Connected += (s, e) => Debug.Log("接続したときの処理");

mqttClient.Disconnected += (s, e) => Debug.Log("切断されたときの処理");
      
await mqttClient.ConnectAsync(options);
```

`IMqttClient.Disconnected` は `IMqttClient.DisconnectAsync()` を呼んだ場合も、意図しない切断が発生したときも両方呼ばれます。
意図しない切断の場合は `MqttClientDisconnectedEventArgs.Exception` に何かしら入っているので、そこを見てよしなにやれば再接続処理などもできます。

```csharp
mqttClient.Disconnected += async (s, e) =>
{
    if (e.Exception == null)
    {
        Debug.Log("意図した切断です");
        return;
    }

    Debug.Log("意図しない切断です。5秒後に再接続を試みます");

    await Task.Delay(TimeSpan.FromSeconds(5));

    try
    {
        await mqttClient.ConnectAsync(options);
    }
    catch
    {
        Debug.Log("再接続に失敗しました");
    }
};
```

# メッセージの送受信

## 受信

受信したメッセージは `IMqttClient.ApplicationMessageReceived` でハンドリングします。

```csharp
mqttClient.ApplicationMessageReceived += (s, e) =>
{
    var stringBuilder = new StringBuilder();
    stringBuilder.AppendLine("メッセージを受信しました");
    stringBuilder.AppendLine($"Topic = {e.ApplicationMessage.Topic}");
    stringBuilder.AppendLine($"Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
    stringBuilder.AppendLine($"QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
    stringBuilder.AppendLine($"Retain = {e.ApplicationMessage.Retain}");
    Debug.Log(stringBuilder);
};
```

## 購読

`TopicFilter` をクライアントに渡して購読します。

```csharp
wait mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("johnson65/helloworld").Build());
};
```

## 送信

`MqttApplicationMessage` をクライアントに渡して送信します。

```csharp
var message = new MqttApplicationMessageBuilder()
    .WithTopic("johnson65/helloworld")
    .WithPayload("Hello World")
    .WithExactlyOnceQoS()
    .Build();

await mqttClient.PublishAsync(message);
```

# まとめ

来年はこれ使いたいです。

最終的なコード

```csharp

using System;
using System.Text;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Client;
using UnityEngine;

public class Main : MonoBehaviour
{
    IMqttClient mqttClient;

    async void Start()
    {
        var factory = new MqttFactory();
        mqttClient = factory.CreateMqttClient();

        var options = new MqttClientOptionsBuilder()
            .WithTcpServer("broker.hivemq.com")
            .Build();

        mqttClient.Connected += (s, e) => Debug.Log("接続したときの処理");
        
        mqttClient.Disconnected += async (s, e) =>
        {
            Debug.Log("切断したときの処理");
            
            if (e.Exception == null)
            {
                Debug.Log("意図した切断です");
                return;
            }

            Debug.Log("意図しない切断です。５秒後に再接続を試みます");

            await Task.Delay(TimeSpan.FromSeconds(5));

            try
            {
                await mqttClient.ConnectAsync(options);
            }
            catch
            {
                Debug.Log("再接続に失敗しました");
            }
        };
        
        mqttClient.ApplicationMessageReceived += (s, e) =>
        {
            var stringBuilder = new StringBuilder();
            stringBuilder.AppendLine("メッセージを受信しました");
            stringBuilder.AppendLine($"Topic = {e.ApplicationMessage.Topic}");
            stringBuilder.AppendLine($"Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
            stringBuilder.AppendLine($"QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
            stringBuilder.AppendLine($"Retain = {e.ApplicationMessage.Retain}");
            Debug.Log(stringBuilder);
        };

        await mqttClient.ConnectAsync(options);

        await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("johnson65/helloworld").Build());

        var message = new MqttApplicationMessageBuilder()
            .WithTopic("johnson65/helloworld")
            .WithPayload("Hello World")
            .WithExactlyOnceQoS()
            .Build();
        
        await mqttClient.PublishAsync(message);
        
        await mqttClient.DisconnectAsync();
    }
}
```

実行結果

![image.png](https://qiita-image-store.s3.amazonaws.com/0/20631/0522f7d9-7d78-b3a6-5955-a188d2bcbf91.png)

（`TopicFilter` で QoS を指定しなかったから QoS ダウングレードで AtMostOnce になっているな :thinking: ）