GHZDaPing/Assets/ShuZiDaPing/MQTTClient.cs

using LitJson;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Runtime.InteropServices;
using System.Text;
using UnityEngine;
using UnityEngine.Android;
using UnityEngine.UI;
using uPLibrary.Networking.M2Mqtt.Messages;
using static QTTManager;

public class MQTTClient : MonoBehaviour
{

    [DllImport("__Internal")]
    private static extern void Connect(string host, string port, string clientId, string username, string password, string destination);
    [DllImport("__Internal")]
    private static extern void Subscribe(string topic);
    [DllImport("__Internal")]
    private static extern void Send(string topic, string payload);
    [DllImport("__Internal")]
    private static extern void Unsubscribe(string topic);
    [DllImport("__Internal")]
    private static extern void Disconnect();
    public static MQTTClient Instance;
    //public string front = "client/";
    // public string id = "";
    //public string username = "u@unity3";// 需要根据服务器设置
    //public string password = null;// 需要根据服务器设置
    public string rid = "mqttx_b4c02ddc"; //其他人的ID
    //public string account;
    //public string roomId;
    //public string email;
    //public string phone;
    //public string roleId;
    //public string resourcePool;
    //public string expertType;
    //public string _username;

    public event Action<JsonData> OnCoordinate;
    public static byte[] bytes;
   // public static string certification;
    // public static byt ta;
    QTTManager qt;
    // Start is called before the first frame update

    private int ReconnectionCount = 0;

    public MqttState mqttState = MqttState.start;
    private void Awake()
    {
        Instance = this;
    }
    void Start()
    {
        // 请求文件读取和写入权限
        if (!Permission.HasUserAuthorizedPermission(Permission.ExternalStorageRead) ||
            !Permission.HasUserAuthorizedPermission(Permission.ExternalStorageWrite))
        {
            Permission.RequestUserPermission(Permission.ExternalStorageRead);
            Permission.RequestUserPermission(Permission.ExternalStorageWrite);
        }
        bytes = (Resources.Load("emqxsl-ca") as TextAsset).bytes;

        Debug.Log(" DGJ  ===> emqxsl-ca.bytes " + bytes.Length);
    }
    Queue<MqttMsgPublishEventArgs> rlist = new Queue<MqttMsgPublishEventArgs>();

    Queue<ShowError> errorList = new Queue<ShowError>();
    private void OnReceived(MqttMsgPublishEventArgs obj)
    {
        rlist.Enqueue(obj);


    }

    private void OnConnecting()
    {
        Debug.Log("MQtt  连接中");
    }

    private void onSucceed()
    {
     //   ErrorPopup.Instance.UpdateEmqxState(NetState.Connect);
        ReconnectionCount = 0;
        if (mqttState == MqttState.reconnection)
        {         
         //   ErrorPopup.Instance.ShowNetError("网络重新连接", 3);
            errorList.Enqueue(new ShowError("信令 重连成功", 3));

        }
        mqttState = MqttState.successfu;
        // Subscribe(MQTTManager.Instance.front + "online/" + MQTTManager.Instance.phone);

        //Subscribe(MQTTManager.Instance.front + "personnel/" + MQTTManager.Instance.resourcePool);
        if( !string.IsNullOrEmpty( MQTTManager.Instance.roomId))
        Subscribe(MQTTManager.Instance.front + "room/" + MQTTManager.Instance.roomId);
        Subscribe(MQTTManager.Instance.front + "personnel/" + MQTTManager.Instance.resourcePoolId);
        Subscribe(MQTTManager.Instance.front + "online/" + MQTTManager.Instance.id);
        Subscribe(MQTTManager.Instance.front + "sensor/VOICE"); //噪声
        Subscribe(MQTTManager.Instance.front + "sensor/TEMPERATURE_HUMIDITY"); //温湿度传感器
        Subscribe(MQTTManager.Instance.front + "sensor/VIBRATE"); //振动传感器
        //if (!string.IsNullOrEmpty( MQTTManager.Instance.topiceNavigation))
        //{
        //    Subscribe(MQTTManager.Instance.topiceNavigation);
        //    Subscribe(MQTTManager.Instance.topiceSync);
        //    Subscribe(MQTTManager.Instance.topiceClose);
        //}

        // Subscribe((MQTTManager.Instance.front + "younuo"));
     //   MQTTManager.Instance.isCreateRoom = true;
       
    }



    private void onFaild()
    {
        Debug.Log("MQtt  连接失败");
    }

    // Update is called once per frame
    void Update()
    {
        if (rlist.Count > 0)
        {
            for (int i = 0; i < rlist.Count; i++)
            {
                OnUnityReceived(rlist.Dequeue());
            }
        }

        if (errorList.Count > 0)
        {
            ShowError error = errorList.Dequeue();
        //    ErrorPopup.Instance.ShowEmqxError(error.error, error.timer);
        }
    }

    public Text SENSO;
    public Text disp;
    public Text speed;
    public Text temperature;
    public Text humidity;
    //MQTT接收到的数据
    public void OnUnityReceived(MqttMsgPublishEventArgs obj)
    {
        string msg = Encoding.UTF8.GetString(obj.Message);
        Debug.Log("uid => " + obj.Topic + ":\n" + msg);
        JsonData data = JsonMapper.ToObject(msg);
        switch (data["device"]["deviceCode"].ToString())
        {
            case "VOICE_SENSO_JSTZ01"://噪声
                SENSO.text = data["value"].ToString()+ data["unit"].ToString();
                break;
            case "VIBRATE_SENSOR_JSTZ01"://震动
                disp.text = data["disp"]["value"].ToString() + data["disp"]["unit"].ToString();
                speed.text = data["speed"]["value"].ToString() + data["speed"]["unit"].ToString(); 
                break;
            case "TEMPERATURE_HUMIDITY_SENSO_JSTZ01"://温湿度
                temperature.text = data["temperature"]["value"].ToString() + data["temperature"]["unit"].ToString();
                humidity.text =data["humidity"]["value"].ToString() + data["humidity"]["unit"].ToString();
                break;
        }
     //   UILogManager.Instance.text3.text = msg;
      //  MQTTManager.Instance.Received(msg);
        MQTTManager.Instance.Received(obj);

    }

    private void OnDestroy()
    {
       
    }

    private void OnApplicationQuit()
    {
        DisConnect();
    }
    //连接
    public void Connect(string host)
    {
        Debug.Log(" DGJ  ===> emqxsl-ca.bytes 1 " + bytes.Length);

        Debug.Log(" DGJ  ===> username " + MQTTManager.Instance.username);
        Debug.Log(" DGJ  ===> clientId " + MQTTManager.Instance.clientId);
        Debug.Log(" DGJ  ===> certification " + MQTTManager.certification);
        //if (DeviceType.type == "Phone")
        //{
        //    id =UserInfo.Instance.name + "_Phone";
        //    rid = front+UserInfo.Instance.name + "_Glasses";

        //}
        //else
        //{
        //    id = UserInfo.Instance.name + "_Glasses";
        //    rid = front+UserInfo.Instance.name + "_Phone";
        //}

        //  Debug.Log("DGJ ==>MQTT " + id + account + "  " + username + "   " + certification + "  " + HttpActionLang.Instance.mqttSocket);

        Connect(host, "1883", MQTTManager.Instance.clientId, MQTTManager.Instance.username,"", MQTTManager.certification);
        return;
        try
        {

            qt = new QTTManager(MQTTManager.Instance.clientId, MQTTManager.Instance.username, MQTTManager.certification, host, "1883");
            Debug.Log("DGJ ===> MQTT " + qt!=null);
            qt.ConnectionFailed += onFaild;
            qt.ConnectionSucceeded += onSucceed;
            qt.OnConnecting += OnConnecting;
            qt.OnReceived += OnReceived;
            qt.OnClose += OnClose;
            
            qt.Connect(bytes);
            StartCoroutine(Reconnection());
        }
        catch (Exception e)
        {

            Debug.LogError("MQTT Connect Error  "+ e.Message);
        }
        
        //  StartSendCameraPos();
    }

    private void OnClose(EventArgs obj)
    {
        Debug.Log("断开连接");

        mqttState = MqttState.disconnect;

      //  ErrorPopup.Instance.UpdateEmqxState(NetState.ReConnection);
        errorList.Enqueue(new ShowError("信令 断开连接 重连中...  ", -1));

     
    }

    private IEnumerator Reconnection()
    {
        while (true)
        {
            yield return new WaitForSeconds(5);
            
            if (qt != null && !qt.IsConnect())
            {
               
                mqttState = MqttState.reconnection;

                if (ReconnectionCount >= 3)
                {
                    Debug.LogError(" DGJ RTC服务重连失败，请退出应用重试 ");
                    DisConnect();
                  //  ErrorPopup.Instance.UpdateEmqxState(NetState.DisConnect);
                    errorList.Enqueue(new ShowError("信令重连失败，请检测网络状态", -1));
                  //  SignalClient.showErrorList.Enqueue(new ShowError("RTC服务重连失败，请退出应用重试", -1));
                }
                else
                {
                    DisConnect();
                    Debug.Log(" DGJ ===>  Reconnection ");
                    ReconnectionCount++;
                    Connect(MQTTManager.host);
                }
               
            }
        }
    }



    //断开连接
    public void DisConnect()
    {
        if (qt != null && qt.IsConnect())
            qt.DisConnect();

    }
    /*
    /// <summary>
    /// 订阅频道
    /// </summary>
    /// <param name="channelID"></param>
    public void Subscribe(string channelID)
    {
        Debug.Log(" DGJ   =====> Subscribe " + channelID);
        if (qt != null && qt.IsConnect())
        {
            ushort s = qt.Subscribe(
                        new string[]
                        {
                              channelID
                              //front+"room/"+roomId

                        },
                        new byte[]
                        {
                                MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE,
                                //MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE

                        });

            
            Debug.Log(" DGJ   =====> Subscribe2");


          
        }
        else
            Debug.LogError(" MQTT 未连接 ！！！");

     


    }
    
    public void UnSubscribe(string channelID)
    {
        if (qt != null && qt.IsConnect())
        {
          //  ushort s = qt.Unsubscribe(new string[] { channelID });
        }
    }*/

    public void publish(byte[] bs)
    {
        if (qt != null && qt.IsConnect())
        {
           // Debug.Log(id + account + "  DGJ  publish =====>  " + front + "room/" + roomId + "   " + bs.Length);
          //  qt.Publish(front + "room/" + roomId, bs, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, false);
        }
        else
        {
            Debug.LogError(" MQTT  未连接 ");
        }

    }

    public void publish(string channelID, byte[] bs)
    {
        if (qt != null && qt.IsConnect())
        {
            Debug.Log(MQTTManager.Instance.clientId + MQTTManager.Instance.account + "  DGJ  publish =====>  " + channelID + "   " + bs.Length);
            qt.Publish(channelID, bs, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, false);
        }
        else
        {
            Debug.LogError(" MQTT  未连接 ");
        }

    }
}

public enum MqttState
{
    start,
    successfu,
    reconnection,
    disconnect
}

public class ShowError
{
    public string error;
    public float timer;

    public ShowError( string error,float timer)
    {
        this.error = error;
        this.timer = timer;
    }
}