chutian_modules/parkspace_manager/communication/communication_socket_base.cs


using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using NNanomsg.Protocols;
using System.Threading;
using Newtonsoft.Json.Linq;
using Google.Protobuf;
using tool;
using System.Runtime.InteropServices;
using System.Windows.Forms;

namespace communication_base
{
    class Communication_socket_base
    {
        //[DllImport("nnxx_socket_lib.dll")]
        //private static extern int nnxx_init();
        //[DllImport("nnxx_socket_lib.dll")]
        //private static extern int nnxx_uninit();
        //[DllImport("nnxx_socket_lib.dll")]
        //private static extern int nnxx_bind(string addr);

        //[DllImport("nnxx_socket_lib.dll")]
        //private static extern int nnxx_connect(string addr);

        //[DllImport("nnxx_socket_lib.dll")]
        //private static extern int nnxx_recv(byte[] addr, uint len, int flags = 0);

        //[DllImport("nnxx_socket_lib.dll")]
        //private static extern int nnxx_send(byte[] addr, uint len, int flags = 0);

        //[DllImport("nnxx_socket_lib.dll")]
        //private static extern void nnxx_close();

        [DllImport("nanomsg_nng_lib.dll")]
        private static extern void nanomsg_nng_init();
        [DllImport("nanomsg_nng_lib.dll")]
        private static extern void nanomsg_nng_uninit();
        [DllImport("nanomsg_nng_lib.dll")]
        private static extern int nanomsg_nng_bind(string addr);

        [DllImport("nanomsg_nng_lib.dll")]
        private static extern int nanomsg_nng_connect(string addr);

        [DllImport("nanomsg_nng_lib.dll")]
        private static extern int nanomsg_nng_recv(byte[] addr, uint len);

        [DllImport("nanomsg_nng_lib.dll")]
        private static extern int nanomsg_nng_send(byte[] addr, uint len);

        //通信状态
        public enum Communication_statu
        {
            COMMUNICATION_UNKNOW = 0,            //通信状态 未知
            COMMUNICATION_READY = 1,            //通信状态 正常
            COMMUNICATION_FAULT = 2,            //通信状态 错误
        };
        public Communication_socket_base()
        {
            m_communication_statu = Communication_statu.COMMUNICATION_UNKNOW;

            mp_receive_data_thread = null;
            mp_analysis_data_thread = null;
            mp_send_data_thread = null;
            mp_encapsulate_data_thread = null;

        }
        //初始化通信模块--默认从json文件读取
        public virtual void communication_init()
        {
            //m_socket = new BusSocket();
            //nnxx_init();
            nanomsg_nng_init();
            m_receive_data_queue = new Queue<ByteString>();
            m_send_data_queue = new Queue<ByteString>();

            //接受线程默认循环
            m_receive_data_condition = true;
            mp_receive_data_thread = new Thread(receive_data_thread);
            //解析线程默认等待, 需要接受线程去唤醒, 超时1ms, 超时后主动遍历m_receive_data_queue
            m_analysis_data_condition = true;
            mp_analysis_data_thread = new Thread(analysis_data_thread);
            //发送线程默认循环
            m_send_data_condition = true;
            mp_send_data_thread = new Thread(send_data_thread);
            //封装线程默认循环
            m_encapsulate_data_condition = true;
            mp_encapsulate_data_thread = new Thread(encapsulate_data_thread);

            m_event_exit = new EventWaitHandle(false, EventResetMode.ManualReset);

            m_analysis_cycle_time = 1000;//默认1000ms,就自动解析(接受list)
            m_encapsulate_cycle_time = 1000;//默认1000ms,就自动发送一次状态信息

            //绑定
            string bind_string = json_file_operation.Instance.read_json_string("communication_bind");
            communication_bind(bind_string);

            //连接
            JArray connect_string = json_file_operation.Instance.read_json_jarray("communication_connect");
            foreach (var connect in connect_string)
            {
                communication_connect(connect.ToString());
            }

            //启动通信, run thread
            communication_run();
        }
        //绑定
        public virtual void communication_bind(string bind_string)
        {
            //m_socket.Bind(bind_string);
            //nnxx_bind(bind_string);
            int results = nanomsg_nng_bind(bind_string);
            if (results == 0)
            {
                Console.WriteLine("communication_bind:" + bind_string + "  success !");
            }
            else
            {
                Console.WriteLine("communication_bind:" + bind_string + "  fail !!!!");
                MessageBox.Show("communication_bind:" + bind_string + "  fail !!!!");

            }
        }
        //连接
        public virtual void communication_connect(string connect_string)
        {
            //m_socket.Connect(connect_string);
            //nnxx_connect(connect_string);
            int results = nanomsg_nng_connect(connect_string);
            if (results == 0)
            {
                Console.WriteLine("communication_connect:" + connect_string + "  success !");
            }
            else
            {
                Console.WriteLine("communication_connect:" + connect_string + "  fail !!!!");
                MessageBox.Show("communication_connect:" + connect_string + "  fail !!!!");

            }
        }
        //启动通信
        public virtual void communication_run()
        {
            m_communication_statu = Communication_statu.COMMUNICATION_READY;
            //启动4个线程。
            mp_receive_data_thread.Start();
            mp_analysis_data_thread.Start();
            mp_send_data_thread.Start();
            mp_encapsulate_data_thread.Start();

        }
        //反初始化
        public virtual void communication_uninit()
        {
            //关闭几个线程
            m_receive_data_condition = false;
            m_analysis_data_condition = false;
            m_send_data_condition = false;
            m_encapsulate_data_condition = false;

            //回收4个线程的资源
            if (mp_receive_data_thread != null)
            {
                mp_receive_data_thread.Join();
            }
            if (mp_analysis_data_thread != null)
            {
                mp_analysis_data_thread.Join();
            }
            if (mp_send_data_thread != null)
            {
                mp_send_data_thread.Join();
            }
            if (mp_encapsulate_data_thread != null)
            {
                mp_encapsulate_data_thread.Join();
            }
            //清空list
            lock (m_receive_list_lock)
            {
                m_receive_data_queue.Clear();
            }
            lock (m_send_list_lock)
            {
                m_send_data_queue.Clear();
            }
            //nnxx_close();
            //nnxx_uninit();
            nanomsg_nng_uninit();
            m_communication_statu = Communication_statu.COMMUNICATION_UNKNOW;
        }


        //设置解析时间
        public void set_analysis_cycle_time(int analysis_cycle_time)
        {
            m_analysis_cycle_time = analysis_cycle_time;

        }
        //设置封装时间
        public void set_encapsulate_cycle_time(int encapsulate_cycle_time)
        {
            m_encapsulate_cycle_time = encapsulate_cycle_time;

        }

        //mp_receive_data_thread 接受线程执行函数，
        //receive_data_thread 内部线程负责接受消息
        protected void receive_data_thread()
        {
            Console.WriteLine(" Communication_socket_base::receive_data_thread start  ");

            //通信接受线程, 负责接受socket消息, 并存入 m_receive_data_queue
            while (m_receive_data_condition && !m_event_exit.WaitOne(1))
            {
                if (m_receive_data_condition)
                {
                    //非阻塞接受消息, 如果接收到消息, data!=null
                    //byte[] data = m_socket.ReceiveImmediate();
                    byte[] data = new byte[65536];
                    int length = nanomsg_nng_recv(data, 65536);
                    if (length > 0)
                    //if (data != null)
                    {

                        //截取实际长度
                        ByteString message = ByteString.CopyFrom(data.Skip(0).Take(length).ToArray());
                        //ByteString message = ByteString.CopyFrom(data.ToArray());

                        Message.Base_info base_Info = Message.Base_msg.Parser.ParseFrom(message).BaseInfo;
                        //Console.WriteLine(base_Info.ToString());
                        //检查消息是否有效, 主要检查消息类型和接受者, 判断这条消息是不是给我的.
                        if (check_msg(message))
                        {
                            lock (m_receive_list_lock)
                            {
                                //添加入队列
                                m_receive_data_queue.Enqueue(message);
                            }
                        }
                    }
                }
            }
            Console.WriteLine(" Communication_socket_base::receive_data_thread end  ");

        }

        //mp_analysis_data_thread 解析线程执行函数，
        //analysis_data_thread 内部线程负责解析消息
        protected void analysis_data_thread()
        {
            Console.WriteLine(" Communication_socket_base::analysis_data_thread start  ");
            //通信解析线程, 负责巡检m_receive_data_queue, 并解析和处理消息
            while (m_analysis_data_condition && !m_event_exit.WaitOne(1))
            {
                if (m_analysis_data_condition)
                {
                    //处理队列信息
                    lock (m_receive_list_lock)
                    {
                        if (m_receive_data_queue.Count > 0)
                        {
                            //取出消息
                            ByteString message = m_receive_data_queue.Dequeue();
                            //检查消息是否可以被处理
                            if (check_executer())
                            {
                                //处理消息
                                execute_msg(message);
                            }
                            else
                            {
                                //不能处理就放到末尾 等能处理时处理
                                m_receive_data_queue.Enqueue(message);
                            }
                        }
                        //没有消息直接跳过
                    }
                }
            }
            Console.WriteLine(" Communication_socket_base::analysis_data_thread end  ");

        }

        //mp_send_data_thread 发送线程执行函数，
        //send_data_thread 内部线程负责发送消息
        protected void send_data_thread()
        {
            Console.WriteLine(" Communication_socket_base::send_data_thread start  ");
            //通信发送线程, 负责巡检m_send_data_queue, 并发送消息
            while (m_send_data_condition && !m_event_exit.WaitOne(1))
            {
                if (m_send_data_condition)
                {
                    lock (m_send_list_lock)
                    {
                        if (m_send_data_queue.Count > 0)
                        {
                            ByteString message = m_send_data_queue.Dequeue();
                            //m_socket.Send(message.ToByteArray());
                            //nnxx_send(message.ToByteArray(), (uint)message.ToByteArray().Length, 1);
                            nanomsg_nng_send(message.ToByteArray(), (uint)message.ToByteArray().Length);
                        }
                    }
                }
            }

            Console.WriteLine(" Communication_socket_base::send_data_thread end  ");
            return;
        }


        //mp_encapsulate_data_thread 封装线程执行函数，
        //encapsulate_data_thread 内部线程负责封装消息
        protected void encapsulate_data_thread()
        {
            Console.WriteLine(" Communication_socket_base::encapsulate_data_thread start  ");
            //通信封装线程, 负责定时封装消息, 并存入 m_send_data_queue
            while (m_encapsulate_data_condition && !m_event_exit.WaitOne(m_encapsulate_cycle_time))
            {

                if (m_encapsulate_data_condition)
                {
                    encapsulate_send_data();
                }
            }
            Console.WriteLine(" Communication_socket_base::encapsulate_data_thread end  ");

        }
        //检查消息是否有效, 主要检查消息类型和接受者, 判断这条消息是不是给我的.
        //子类必须重载, 增加自己模块的判断逻辑
        public virtual bool check_msg(ByteString message)
        {
            // Message.Base_info base_Info = Message.Base_info.Parser.ParseJson(message);
            Message.Base_info base_Info = Message.Base_msg.Parser.ParseFrom(message).BaseInfo;
            if (base_Info.Receiver == Message.Communicator.EMain)
            {
                return true;
            }
            else
            {
                //无效的消息
                return true;
            }
        }
        //检查执行者的状态, 判断能否处理这条消息, 需要子类重载
        public virtual bool check_executer()
        {
            //检查对应模块的状态, 判断是否可以处理这条消息
            if (m_communication_statu == Communication_statu.COMMUNICATION_READY)
                return true;
            return false;
        }
        //处理消息, 需要子类重载
        public virtual void execute_msg(ByteString message)
        {
            //处理消息必须子类重载,本模块只负责通信
        }
        //封装消息, 需要子类重载
        public virtual void encapsulate_msg(ByteString message)
        {
            lock (m_send_list_lock)
            {
                m_send_data_queue.Enqueue(message);
            }
        }
        //定时封装发送消息, 一般为心跳和状态信息, 需要子类重载
        public virtual void encapsulate_send_data()
        {
            return;
            Message.Base_msg t_base_msg;
            Message.Store_command_request_msg msg;
            t_base_msg.BaseInfo.MsgType = Message.Message_type.EBaseMsg;
            t_base_msg.BaseInfo.TimeoutMs = 5000;
            t_base_msg.BaseInfo.Sender = Message.Communicator.EMain;
            t_base_msg.BaseInfo.Receiver = Message.Communicator.EMain;
            m_send_data_queue.Enqueue(t_base_msg.ToByteString());
        }



        //通用的网络编程接口, 默认使用总线模式, (网状结构)
          protected BusSocket m_socket;


        //数据锁
        //作用：将会锁住代码块的内容，并阻止其他线程进入该代码块，直到该代码块运行完成，释放该锁。
        //注意：定义的锁对象应该是 私有的，静态的，只读的，引用类型的对象，这样可以防止外部改变锁对象
        private static readonly object m_receive_list_lock = new object();
        private static readonly object m_send_list_lock = new object();

        //通信状态
        protected Communication_statu m_communication_statu;          //通信状态

        //接受模块,
        private Queue<ByteString> m_receive_data_queue;            //接受的队列容器
        protected Thread mp_receive_data_thread;         //接受的线程
        protected bool m_receive_data_condition;		//接受的条件变量
        protected Thread mp_analysis_data_thread;        //解析的线程
        protected bool m_analysis_data_condition;		//解析的条件变量
        protected int m_analysis_cycle_time;          //自动解析的时间周期

        //发送模块,
        private Queue<ByteString> m_send_data_queue;         //发送的队列容器
        protected Thread mp_send_data_thread;            //发送的线程
        protected bool m_send_data_condition;			//发送的条件变量
        protected Thread mp_encapsulate_data_thread;     //封装的线程
        protected bool m_encapsulate_data_condition;	//封装的条件变量
        protected int m_encapsulate_cycle_time;         //自动封装的时间周期

        protected EventWaitHandle m_event_exit;                   //防止CPU卡死

    }
}