cluster 에서는 현재 M2Mqtt 를 사용하고 있습니다만, Async 나 Rx 의 서포트가 있는 라이브러리 사용하고 싶은 기분이 발생했으므로 chkr1011/MQTTnet 를 접해 보고 싶습니다.

(이 기사는 Cluster,Inc. Advent Calendar 2018 - Qiita 9일째의 기사의 것이었습니다)

준비

가져오기

Unity(이번에는 2018.3.0f2 사용)에서 새 프로젝트를 만듭니다.
NuGet Gallery에서 MQTTnet 2.8.5 패키지를 다운로드하고 배포(.nupkg -> .zip으로 이름을 바꾸고 확장)한 다음 .NET 4.7.2용 DLL 파일을 프로젝트로 가져옵니다.

연결

MqttFactory에서 만든 클라이언트에 MqttClientOptionsBuilder에서 만든 연결 옵션을 전달하여 연결합니다.IMqttClientOptions 에 한정하지 않고, MQTTnet 는 오브젝트 작성이 대체로 Builder 패턴으로 제공되고 있어 M2Mqtt 에 비해 하면 인수가 많은 생성자나 메소드가 나오지 않기 때문에 정신 위생상 좋다.

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 에 무엇인가 들어가 있으므로, 거기를 보고 좋게 하면 재접속 처리등도 할 수 있습니다.

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로 처리됩니다.

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를 클라이언트에 전달하고 구독합니다.

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

전송

MqttApplicationMessage를 클라이언트에 전달하여 보냅니다.

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

await mqttClient.PublishAsync(message);

요약

내년에는 이것을 사용하고 싶습니다.

최종 코드

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();
    }
}

실행 결과



(TopicFilter에서 QoS를 지정하지 않았기 때문에 QoS 다운 그레이드에서 AtMostOnce가되지 않았습니다 )