Simulation/Assets/WorkSpace/LH/MQTTManager.cs

using UnityEngine;
using Best.MQTT;
using Best.MQTT.Packets;
using Best.MQTT.Packets.Builders;
using Newtonsoft.Json;
using System.IO;
using System;
using System.Collections.Generic;
using System.Text;
using System.Collections;
using Octopus.Simulator;

namespace Octopus.Simulator.Networks
{
    public class MQTTManager : MonoBehaviour
    {
        public static MQTTManager mqttManager;

        public Dictionary<string, MQTTClient> clientTable = new Dictionary<string, MQTTClient>();
        public BaseSimulationMessage basemessage;

        [SerializeField]
        string MQTTpath = "";
        MQTTClient client;

        [SerializeField]
        string _topic;
        [SerializeField]
        public string topic
        {
            get
            {
                return _topic;
            }
            set
            {
                _topic = value;
            }
        }

        public event Action onConnectedEvent;
        public event Action<DisconnectReasonCodes, string> onDisconnectedEvent;
        public event Action<string, string> onMessageReceived;
        public event Action<string> onErrorEvent;
        public event Action onCompletedTimerEvent;
        public 
        // Start is called once before the first execution of Update after the MonoBehaviour is created
        void Awake()
        {
            mqttManager = this;
            SetMqttConfig();
        }

        private void SetMqttConfig()
        {
            string path = Application.streamingAssetsPath + MQTTpath;
            string json;
            using (StreamReader reader = new StreamReader(path))
            {
                json = reader.ReadToEnd();
            }
            MQTTConfigList ConfigList = JsonConvert.DeserializeObject<MQTTConfigList>(json);
            foreach (var config in ConfigList.configs)
            {
                string clientName = config.name;
                string host = config.host;
                int port = config.port;

                Connect(clientName, host,port);
            }
        }

        public void SubscribeTopic(MQTTClient client, string topic)
        {
                client.AddTopicAlias(topic);

                client.CreateSubscriptionBuilder(topic)
                      .WithMessageCallback(OnMessage)
                      .WithMaximumQoS(QoSLevels.ExactlyOnceDelivery)
                      .BeginSubscribe();
        }

        public void UnsubscribeTopic(MQTTClient client,string topic)
        {
            client.CreateUnsubscribePacketBuilder(topic).BeginUnsubscribe();
        }

        public void Connect(string clientName, string host,int port)
        {
            client = new MQTTClientBuilder()
                .WithOptions(new ConnectionOptionsBuilder().WithTCP(host, port))
                .WithEventHandler(client=>OnConnected(client,clientName))
                .WithEventHandler(OnStateChange)
                .WithEventHandler(OnDisconnected)
                .WithEventHandler(OnError)
                .CreateClient();

            client.BeginConnect(ConnectPacketBuilderCallback);
        }

        public event Action<MQTTClient> onConnected;
        private void OnConnected(MQTTClient client,string clientName)
        {
            clientTable.Add(clientName, client);
            onConnected?.Invoke(client);

            SubscribeTopic(client, topic);
        }

        private ConnectPacketBuilder ConnectPacketBuilderCallback(MQTTClient client, ConnectPacketBuilder builder)
        {
            return builder;
        }

        private void OnError(MQTTClient client, string error)
        {
            Debug.Log($"OnError! :{error}");
        }

        private void OnMessage(MQTTClient client, SubscriptionTopic topic, string topicName, ApplicationMessage message)
        {
            var payload = Encoding.UTF8.GetString(message.Payload.Data, message.Payload.Offset, message.Payload.Count);
            Debug.Log(payload);
            basemessage = JsonConvert.DeserializeObject<BaseSimulationMessage>(payload);
            onMessageReceived?.Invoke(topicName, payload);
        }

        private void OnDisconnected(MQTTClient client, DisconnectReasonCodes reasonCode, string reasonMessage)
        {
            Debug.Log(reasonMessage);
            //throw new NotImplementedException();
        }

        private void OnStateChange(MQTTClient client, ClientStates oldState, ClientStates newState)
        {
            //throw new NotImplementedException();
        }

        void OnDestroy()
        {
            client?.CreateDisconnectPacketBuilder()
                .WithReasonCode(DisconnectReasonCodes.NormalDisconnection)
                //.WithReasonString("Bye")
                .BeginDisconnect();
        }
    }
}