MQTTプロトコルを用いたメッセージングシステムの実装例を示す。サーバー側にはMQTTnet 3.1.2を採用し、クライアントは最新バージョン4.1.0を使用する。

ブローカー実装

MQTTブローカーを構築するための基本実装:

using MQTTnet;
using MQTTnet.Server;
using System.Text;

namespace MqttBroker
{
    public class BrokerService
    {
        private IMqttServer _broker;
        private const int DefaultPort = 1883;
        private const string AuthUser = "admin";
        private const string AuthPass = "securePass";

        public async Task InitializeBroker()
        {
            var config = new MqttServerOptionsBuilder()
                .WithDefaultEndpointPort(DefaultPort)
                .WithConnectionValidator(ctx =>
                {
                    if (ctx.Username != AuthUser || ctx.Password != AuthPass)
                    {
                        ctx.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                        return;
                    }
                    ctx.ReasonCode = MqttConnectReasonCode.Success;
                })
                .Build();

            _broker = new MqttFactory().CreateMqttServer();
            _broker.ClientConnectedAsync += OnClientConnect;
            _broker.ClientDisconnectedAsync += OnClientDisconnect;
            _broker.ApplicationMessageReceivedAsync += OnMessageReceive;

            await _broker.StartAsync(config);
        }

        private Task OnClientConnect(MqttServerClientConnectedEventArgs e)
        {
            Console.WriteLine($"クライアント {e.ClientId} 接続");
            return Task.CompletedTask;
        }

        private Task OnMessageReceive(MqttApplicationMessageReceivedEventArgs e)
        {
            var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
            Console.WriteLine($"受信トピック: {e.ApplicationMessage.Topic} | 内容: {payload}");
            return Task.CompletedTask;
        }
    }
}

メッセージ購読クライアント

特定トピックを監視するクライアント実装:

using MQTTnet;
using MQTTnet.Client;

namespace MqttSubscriber
{
    public class SubscriberClient
    {
        private IMqttClient _client;

        public async Task Connect(string brokerIp, string topicFilter)
        {
            var options = new MqttClientOptionsBuilder()
                .WithTcpServer(brokerIp)
                .WithCredentials("admin", "securePass")
                .Build();

            _client = new MqttFactory().CreateMqttClient();
            _client.ApplicationMessageReceivedAsync += OnNotificationReceived;
            
            await _client.ConnectAsync(options);
            await _client.SubscribeAsync(topicFilter);
        }

        private Task OnNotificationReceived(MqttApplicationMessageReceivedEventArgs e)
        {
            Console.WriteLine($"新着メッセージ: {e.ApplicationMessage.Topic}");
            return Task.CompletedTask;
        }
    }
}

メッセージ送信クライアント

トピックへメッセージを送信する実装:

using MQTTnet;
using MQTTnet.Client;

namespace MqttPublisher
{
    public class PublisherClient
    {
        private IMqttClient _client;

        public async Task SendMessage(string brokerIp, string targetTopic, string content)
        {
            var options = new MqttClientOptionsBuilder()
                .WithTcpServer(brokerIp)
                .Build();

            _client = new MqttFactory().CreateMqttClient();
            await _client.ConnectAsync(options);

            var message = new MqttApplicationMessageBuilder()
                .WithTopic(targetTopic)
                .WithPayload(content)
                .Build();

            await _client.PublishAsync(message);
        }
    }
}

動作検証手順

1. ブローカーアプリケーションを起動
2. 購読クライアントを実行しトピックを監視
3. 送信クライアントからメッセージを発行