chutian_modules/parkspace_manager/communication/Communicator.cs

using NNanomsg;
using NNanomsg.Protocols;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Google.Protobuf;
using Message;

namespace parkspace_manager
{
    class Communicator
    {
        // 各类消息委托，收到对应消息即调用对应委托/回调
        public delegate void parkspaceStatusDelegate(Parkspace_allocation_data_msg parkspace_status);
        public delegate void parkspaceForceUpdateResponseDelegate(Parkspace_force_update_response_msg response);
        // 对应委托/回调句柄
        private parkspaceStatusDelegate m_parkspaceStatusCallback;
        private parkspaceForceUpdateResponseDelegate m_parkspaceForceUpdateResponseCallback;
        // 对应委托/回调句柄设置函数
        public void SetParkspaceStatusDelegate(parkspaceStatusDelegate callback)
        {
            m_parkspaceStatusCallback = callback;
        }
        public void SetParkspaceForceUpdateResponseDelegate(parkspaceForceUpdateResponseDelegate callback)
        {
            m_parkspaceForceUpdateResponseCallback = callback;
        }

        /// <summary>
        /// 消息超时时间
        /// </summary>
        public int m_timeout_milli = 2000;
        /// <summary>
        /// 单例锁对象
        /// </summary>
        private readonly static object lockObj = new object();
        /// <summary>
        /// 单例
        /// </summary>
        private static Communicator instance = null;
        /// <summary>
        /// 接收解析锁
        /// </summary>
        private object m_receive_lock;
        /// <summary>
        /// 发送锁
        /// </summary>
        private object m_send_lock;
        /// <summary>
        /// 车位状态访问锁
        /// </summary>
        private object m_parkspace_status_access_lock;
        /// <summary>
        /// 车位消息队列
        /// </summary>
        private Parkspace_allocation_data_msg m_parkspace_current_status;
        /// <summary>
        /// 当前时间
        /// </summary>
        private DateTime m_last_parkspace_status_time;
        /// <summary>
        /// 发送队列
        /// </summary>
        private Queue<ByteString> m_send_queue;
        /// <summary>
        /// 接收队列
        /// </summary>
        private Queue<ByteString> m_receive_queue;
        /// <summary>
        /// 发送线程
        /// </summary>
        private Thread m_thread_send;
        /// <summary>
        /// 接收线程
        /// </summary>
        private Thread m_thread_receive;
        /// <summary>
        /// 解析接收string到protobuf消息
        /// </summary>
        private Thread m_thread_decode_receive;
        /// <summary>
        /// nanomsg 通信句柄
        /// </summary>
        private BusSocket m_socket;
        /// <summary>
        /// nnxx生成id队列
        /// </summary>
        private Queue<NanomsgEndpoint> nanomsgEndpoints_queue;
        /// <summary>
        /// 实例退出标记
        /// </summary>
        public bool mb_exit;
        /// <summary>
        /// 初始化标记
        /// </summary>
        public bool mb_initialized;

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="server_ip"></param>
        /// <param name="server_port"></param>
        private Communicator()
        {
            mb_exit = false;
            mb_initialized = false;
            m_receive_lock = new object();
            m_send_lock = new object();
            m_parkspace_status_access_lock = new object();
            m_receive_queue = new Queue<ByteString>();
            m_send_queue = new Queue<ByteString>();
            m_parkspace_current_status = new Parkspace_allocation_data_msg();
            nanomsgEndpoints_queue = new Queue<NanomsgEndpoint>();
        }

        /// <summary>
        /// 析构函数
        /// </summary>
        ~Communicator()
        {
            mb_exit = true;
            if (m_thread_send != null)
            {
                m_thread_send.Abort();
            }
            if (m_thread_receive != null)
            {
                m_thread_receive.Abort();
            }
            if (m_thread_decode_receive != null)
                m_thread_decode_receive.Abort();
        }

        /// <summary>
        /// 单例访问
        /// </summary>
        public static Communicator Instance
        {
            get
            {
                if (instance == null)
                {
                    lock (lockObj)
                    {
                        if (instance == null)
                        {
                            instance = new Communicator();
                        }
                    }
                }
                return instance;
            }
        }

        /// <summary>
        /// 初始化
        /// </summary>
        /// <returns></returns>
        public bool Init()
        {
            try
            {
                if (!mb_initialized)
                {
                    mb_exit = false;
                    mb_initialized = true;
                    m_socket = new BusSocket();
                    m_thread_receive = new Thread(new ParameterizedThreadStart(Receive_thread_function));
                    m_thread_send = new Thread(new ParameterizedThreadStart(Send_thread_function));
                    m_thread_decode_receive = new Thread(new ParameterizedThreadStart(Decode_thread_function));
                    m_thread_receive.Start(this);
                    m_thread_send.Start(this);
                    m_thread_decode_receive.Start(this);
                    // 设置车位状态回调
                    SetParkspaceStatusDelegate(ParkspaceStatusUpdate);

                    // 设置车位刷新线程
                    Task.Factory.StartNew(()=> {
                        while (!mb_exit)
                        {
                            Parkspace_refresh_request_msg t_refresh_req = new Parkspace_refresh_request_msg();
                            t_refresh_req.BaseInfo = new Base_info();
                            t_refresh_req.BaseInfo.MsgType = Message_type.EParkspaceAllocationDataMsg;
                            t_refresh_req.BaseInfo.TimeoutMs = 5000;
                            t_refresh_req.BaseInfo.Sender = Message.Communicator.EEmpty;
                            t_refresh_req.BaseInfo.Receiver = Message.Communicator.EParkspace;
                            Send_msg(t_refresh_req.ToByteString());
                            Thread.Sleep(1000 * 5);
                        }
                    });
                    return true;
                }
                else
                {
                    return false;
                }
            }
            catch (Exception ex) { Console.WriteLine(ex.StackTrace); return false; }
        }

        /// <summary>
        /// 反初始化
        /// </summary>
        /// <returns></returns>
        public bool Uninit()
        {
            mb_exit = true;
            if (m_thread_receive != null)
                m_thread_receive.Join();
            if (m_thread_send != null)
                m_thread_send.Join();
            if (m_thread_decode_receive != null)
                m_thread_decode_receive.Join();
            if(m_socket!=null)
            {
                while(nanomsgEndpoints_queue.Count>0)
                {
                    NanomsgEndpoint nnep = nanomsgEndpoints_queue.Dequeue();
                    m_socket.Shutdown(nnep);
                }
                m_socket.Dispose();
            }
            mb_initialized = false;
            return true;
        }

        /// <summary>
        /// 连接
        /// </summary>
        /// <param name="server_address"></param>
        /// <returns></returns>
        public bool Connect(string server_address)
        {
            if (m_socket == null)
                return false;
            NanomsgEndpoint end_point = m_socket.Connect(server_address);
            nanomsgEndpoints_queue.Enqueue(end_point);
            return true;
        }

        /// <summary>
        /// 绑定本地端口监听
        /// </summary>
        /// <param name="self_address"></param>
        /// <returns></returns>
        public bool Bind(string self_address)
        {
            if (m_socket == null)
                return false;
            NanomsgEndpoint end_point = m_socket.Bind(self_address);
            nanomsgEndpoints_queue.Enqueue(end_point);
            return true;
        }

        /// <summary>
        /// 车位更新回调内部定义
        /// </summary>
        /// <param name="parkspace_status"></param>
        private void ParkspaceStatusUpdate(Parkspace_allocation_data_msg parkspace_status)
        {
            lock (m_parkspace_status_access_lock)
            {
                m_parkspace_current_status = parkspace_status;
                //comm.m_parkspace_current_status.MergeFrom(parkspace_status);
                m_last_parkspace_status_time = DateTime.Now;
            }
        }

        /// <summary>
        /// 获取当前车位状态
        /// </summary>
        public bool Get_parkspace_status(ref Parkspace_allocation_data_msg msg)
        {
            lock (m_parkspace_status_access_lock)
            {
                msg = new Parkspace_allocation_data_msg();
                msg.MergeFrom(m_parkspace_current_status);
            }
            if ((DateTime.Now - m_last_parkspace_status_time).Milliseconds > m_timeout_milli)
            {
                return false;
            }
            else
            {
                return true;
            }
        }

        /// <summary>
        /// 发送手动更新车位消息
        /// </summary>
        /// <returns></returns>
        public bool Send_msg(ByteString bs)
        {
            lock(m_send_lock)
            {
                m_send_queue.Enqueue(bs);
            }
            return true;
        }

        /// <summary>
        /// 接收线程函数
        /// </summary>
        /// <param name="handle"></param>
        private static void Receive_thread_function(object handle)
        {
            if (handle == null)
                return;
            Communicator comm = (Communicator)handle;
            while (!comm.mb_exit)
            {
                try
                {
                    if (!comm.mb_initialized || comm.m_socket == null)
                        continue;
                    byte[] data = comm.m_socket.ReceiveImmediate();
                    if (data != null && data.Length > 0 && comm.m_receive_queue != null)
                    {
                        lock (comm.m_receive_lock)
                        {
                            //ByteString tmp = ByteString.CopyFrom(data);
                            //Base_msg base_msg = Base_msg.Parser.ParseFrom(tmp);
                            //Console.WriteLine(base_msg.ToString());
                            //Console.WriteLine("msg received");
                            comm.m_receive_queue.Enqueue(ByteString.CopyFrom(data));
                        }
                    }
                }
                catch (Exception ex) { Console.WriteLine(ex.StackTrace); }
                Thread.Sleep(50);
            }
            Console.WriteLine("receive thread exit");
        }

        /// <summary>
        /// 发送线程函数
        /// </summary>
        /// <param name="handle"></param>
        private static void Send_thread_function(object handle)
        {
            if (handle == null)
                return;
            Communicator comm = (Communicator)handle;
            while (!comm.mb_exit)
            {
                try
                {
                    if (!comm.mb_initialized || comm.m_socket == null)
                        continue;
                    lock (comm.m_send_lock)
                    {
                        while (comm.m_send_queue.Count > 0)
                        {
                            ByteString bs = comm.m_send_queue.Dequeue();
                            comm.m_socket.Send(bs.ToByteArray());
                        }
                    }
                    //Console.WriteLine("msg sent");
                }
                catch (Exception ex) { Console.WriteLine(ex.StackTrace); }
                Thread.Sleep(50);
            }
            Console.WriteLine("send thread exit");
        }

        /// <summary>
        /// 解析线程函数
        /// </summary>
        /// <param name="handle"></param>
        private static void Decode_thread_function(object handle)
        {
            if (handle == null)
                return;
            Communicator comm = (Communicator)handle;
            while (!comm.mb_exit)
            {
                try
                {
                    DateTime current_time = DateTime.Now;

                    if (!comm.mb_initialized)
                        continue;
                    ByteString msg_str = null;
                    lock (comm.m_receive_lock)
                    {
                        if (comm.m_receive_queue.Count > 0)
                        {
                            msg_str = comm.m_receive_queue.Dequeue();
                        }
                    }
                    // 解析
                    comm.Decode_msg(msg_str);
                    //Console.WriteLine("msg parsed");
                }
                catch (Exception ex) { Console.WriteLine(ex.StackTrace); }
                Thread.Sleep(50);
            }
            Console.WriteLine("decode thread exit");
        }

        /// <summary>
        /// 解析string到protobuf消息
        /// </summary>
        private void Decode_msg(ByteString msg)
        {
            if (msg == null)
                return;
            Base_msg base_msg = Base_msg.Parser.ParseFrom(msg);
            //Console.WriteLine(base_msg.ToString());
            switch(base_msg.BaseInfo.MsgType)
            {
                case Message_type.EParkspaceAllocationDataMsg:
                    Parkspace_allocation_data_msg parkspace_status_msg = Parkspace_allocation_data_msg.Parser.ParseFrom(msg);
                    m_parkspaceStatusCallback?.Invoke(parkspace_status_msg);
                    break;
                case Message_type.EParkspaceForceUpdateResponseMsg:
                    Console.WriteLine("update response");
                    Parkspace_force_update_response_msg parkspace_force_update_msg = Parkspace_force_update_response_msg.Parser.ParseFrom(msg);
                    Console.WriteLine(parkspace_force_update_msg.ToString());
                    m_parkspaceForceUpdateResponseCallback?.Invoke(parkspace_force_update_msg);
                    break;
                default:
                    //Console.WriteLine("unrecognized message received");
                    break;
            }
        }

    }
}