using System;using System.Collections.Generic;using System.Configuration;using System.Linq;using System.Net;using System.Net.Sockets;using System.Text;using System.Threading;using System.Threading.Tasks;using Jinher.AMP.SNS.Chat.Client;using Jinher.AMP.SNS.Chat.Deploy.CustomDTO;using Jinher.AMP.SNS.Chat.Packet;using Jinher.AMP.SNS.Chat.Utility;namespace Jinher.AMP.SNS.Chat.SocketManager{    /// <summary>    /// 消息服务基类    /// </summary>    public abstract class MessageCenter    {        /// <summary>        /// 表示是否连接上        /// </summary>        public bool IsConnected { get; protected set; }        /// <summary>        /// 表示是否进行第一次握手协议        /// </summary>        public bool IsHandStake { get; protected set; }        /// <summary>        /// 表示是否注册APP        /// </summary>        public bool IsRegisterApp { get; protected set; }        private string _ip = string.Empty;        private int _port = 0;        private object lockObject = new object();        List<byte[]> byteList = new List<byte[]>();        //通知        private SocketAsyncEventArgsPool pool = null;        private BufferManager m_bufferManager = null;        /// <summary>        /// 线程休眠时间（毫秒）        /// </summary>        private int threadSleepTime = 5000;        //连接对象        private Socket _socket = null;        private int maxNumber = 10;        protected Action<List<MessageTemplate>> MyAction = null;        protected Action<WebMessageDTO> receiveAction = null;        public MessageCenter()        {            //初始化，获取Host和Port            _ip = System.Configuration.ConfigurationManager.AppSettings["serverip"];            _port = int.Parse(System.Configuration.ConfigurationManager.AppSettings["port"]);            maxNumber = int.Parse(System.Configuration.ConfigurationManager.AppSettings["maxNumber"] ?? "10");            string timer = System.Configuration.ConfigurationManager.AppSettings["threadSleepTime"] ?? "5000";            threadSleepTime = int.Parse(timer);            //初始化连接对象            InitSocket();        }        /// <summary>        /// 带委托的构造函数        /// </summary>        /// <param name="action"></param>        public MessageCenter(Action<List<MessageTemplate>> action)            : this()        {            MyAction = action;        }        /// <summary>        /// 启动消息中心        /// </summary>        public void Run()        {            try            {                LogHelper.WriteLog("启动服务");                //第一次连接服务器                Connect();                //开始接受                AccpetTo();                //开始处理接受的数据                ProcessAccpet();                //保持连接                KeepConnect();                //推送消息队列                //SendAsync();                Task.Factory.StartNew(new Action(() =>                {                    client = ChatClient.Instance(ConfigurationManager.AppSettings["notificationUri"]);                    client.OnMsgReceiveed = null;                    client.OnMsgReceiveed += client_OnMsgReceiveed;                }));            }            catch (SocketException exception)            {                //socket出错                LogHelper.WriteLog("socket创建连接出错,尝试重新启动", exception);                Thread.Sleep(5000);                this.Run();            }            catch (Exception ex)            {                //未知错误                LogHelper.WriteLog("启动时发生未知错误！", ex);                throw;            }            finally            {                //重置连接                //ResetConnect();            }        }        #region 连接服务器        /// <summary>        /// 连接服务器        /// </summary>        protected void Connect()        {            try            {                if (_socket == null)                {                    InitSocket();                }                LogHelper.WriteLog("开始建立连接");                _socket.Connect(IPAddress.Parse(_ip), _port);                LogHelper.WriteLog("连接已经建立");            }            catch (Exception ex)            {                LogHelper.WriteLog("出错了", ex);                LogHelper.WriteLog("建立连接失败，等待重新建立连接");                Thread.Sleep(threadSleepTime);                Connect();                return;            }            //先进行第一次握手协议            HandShakeCmdOp();            //注册app            RegisterApp();            IsConnected = true;        }        /// <summary>        /// 先进行第一次握手协议        /// </summary>        private void HandShakeCmdOp()        {            if (!IsHandStake)            {                LogHelper.WriteLog("开始第一次握手");                _socket.Send(pmsMessage.HandShakePacket());                IsHandStake = true;                LogHelper.WriteLog("成功握手");            }        }        /// <summary>        /// 注册app        /// </summary>        private void RegisterApp()        {            //            一级命令：XNS_ROUTER            //二级命令：REGISTER_SOCIAL_APP            //并且规定AppId：99999999            LogHelper.WriteLog("开始注册APP");            _socket.Send(pmsMessage.RegisterPacket());            IsRegisterApp = true;            LogHelper.WriteLog("注册成功");        }        /// <summary>        /// 初始化连接对象        /// </summary>        private void InitSocket()        {            _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);        }        #endregion        #region 发送消息        /// <summary>        /// 发送消息（异步）        /// </summary>        /// <param name="buffer"></param>        /// <param name="count">表示失败重新发送次数</param>        /// <returns></returns>        protected void SendAsync(byte[] buffer, int count = 0)        {            //先将该消息尝试发送，如果发送失败，则连接后继续发送            //byteList.Add(buffer);            //启动任务，发送消息            Task.Factory.StartNew(new Action(() =>            {                try                {                    if (_socket != null && IsConnected && _socket.Connected && !_isResetConnecting)                    {                        lock (lockObject)                        {                            if (_socket != null && IsConnected && _socket.Connected && !_isResetConnecting)                            {                                SendByInit(buffer);                            }                            else                            {                                ResetConnect();                                SendAsync(buffer);                            }                        }                    }                    else                    {                        ResetConnect();                        SendAsync(buffer);                    }                }                catch (Exception)                {                    //重发一次                    if (count == 0)                    {                        //表示失败重新发送                        SendAsync(buffer, count + 1);                    }                    else                    {                        ResetConnect();                        SendAsync(buffer);                    }                }                finally                {                }            }));        }        /// <summary>        /// 发送数据()        /// </summary>        /// <param name="buffer"></param>        private void SendByInit(byte[] buffer)        {            _socket.BeginSend(buffer, 0, buffer.Length, SocketFlags.None, new AsyncCallback(SendComplated), _socket);        }        /// <summary>        /// 发送消息回调函数        /// </summary>        /// <param name="ar"></param>        private void SendComplated(IAsyncResult async)        {            try            {                Socket skt = async.AsyncState as Socket;                if (skt.Connected)                {                    skt.EndSend(async);                }                LogHelper.WriteLog("发送成功");            }            catch (SocketException ex)            {                //日志文件            }        }        #endregion        #region 接受消息(方式一)        /// <summary>        /// 接受消息方式一        /// </summary>        public void AccpetOne()        {            LogHelper.WriteLog("开始建立接受消息（方式一）");            pool = new SocketAsyncEventArgsPool(10);            // 预先分配一个对象池            SocketAsyncEventArgs readWriteEventArg;            for (int i = 0; i < maxNumber; i++)            {                m_bufferManager = new BufferManager(1024 * 1024);                m_bufferManager.InitBuffer();                //初始化 SocketAsyncEventArgs                readWriteEventArg = new SocketAsyncEventArgs();                readWriteEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(Receive_Completed);                readWriteEventArg.UserToken = new AsyncUserToken() { Manager = m_bufferManager };                // 从缓冲池分配一个字节缓冲区SocketAsyncEventArg对象                m_bufferManager.SetBuffer(readWriteEventArg);                // add SocketAsyncEventArg to the pool                pool.Push(readWriteEventArg);            }            //取出  监视            Receive();        }        private void Receive()        {            if (IsConnected && IsRegisterApp && IsHandStake)            {                SocketAsyncEventArgs readEventArgs = null;                readEventArgs = pool.Pop();                if (readEventArgs != null)                {                    ((AsyncUserToken)readEventArgs.UserToken).Socket = _socket;                    _socket.ReceiveAsync(readEventArgs);                }            }        }        private void Receive_Completed(object sender, SocketAsyncEventArgs e)        {            Receive();            ReceiveAsync(e);        }        private void ReceiveAsync(SocketAsyncEventArgs e)        {            AsyncUserToken token = (AsyncUserToken)e.UserToken;            if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)            {                //echo the data received back to the client                e.SetBuffer(0, e.BytesTransferred);                //string txt = Encoding.Default.GetString(e.Buffer, 0, e.BytesTransferred);                //Console.WriteLine(tmp);                //通知                #region 收到消息平台发送来的消息                //client.Send();                if (e.BytesTransferred > 0)                {                    byte[] temp = new byte[e.BytesTransferred];                    Array.Copy(e.Buffer, 0, temp, 0, e.BytesTransferred);                    ProcessAccpet(temp, e.BytesTransferred);                }                #endregion                BufferManager bufferManager = ((AsyncUserToken)(e.UserToken)).Manager;                if (bufferManager == null)                {                    bufferManager = new BufferManager(1024 * 1024);                    e.UserToken = new AsyncUserToken() { Manager = bufferManager };                }                bufferManager.FreeBuffer(e);                bufferManager.SetBuffer(e);                //回收                pool.Push(e);            }        }        #endregion        #region 接受消息（方式二）        List<byte> testList = new List<byte>();        private void AccpetTo()        {            LogHelper.WriteLog("开始建立接受消息（方式二）");            Task.Factory.StartNew(new Action(() =>            {                try                {                    byte[] bytes = new byte[1024 * 1024];                    while (true)                    {                        int r = _socket.Receive(bytes);                        //通知                        #region 收到消息平台发送来的消息                        if (r > 0)                        {                            byte[] temp = new byte[r];                            Array.Copy(bytes, 0, temp, 0, r);                            lock ("我要给集合增加数据")                            {                                testList.AddRange(temp);                            }                            // ProcessAccpet(temp, r);                        }                        #endregion                        Array.Clear(bytes, 0, r);                    }                }                catch (SocketException ex)                {                    LogHelper.WriteLog("socket出错", ex);                    ResetConnect();                }                catch (Exception ex)                {                    LogHelper.WriteLog("出错", ex);                }            }));        }        #endregion        #region 收到的消息进行处理        /// <summary>        /// 消息处理中转        /// </summary>        /// <param name="p"></param>        /// <param name="r"></param>        private async void ProcessAccpet(byte[] p, int r)        {            //if (r > 10)            //{            //    LogHelper.WriteLog("收到正常消息");            //    var result = pmsMessage.ReversePacketMessage(p, r);            //    //发送给客户端            //    ClientSend(result);            //    //发送给测试端            //    if (MyAction != null)            //    {            //        MyAction(result.ToList());            //    }            //}            //else            //{            //    LogHelper.WriteLog(string.Format("收到心跳包"));            //}        }        //标示处理任务是否正在运行        private bool IsProcessAccpetRuning = false;        private void ProcessAccpet()        {            if (IsProcessAccpetRuning)            {//表示任务正在执行                return;            }            else            {                IsProcessAccpetRuning = true;            }            Task.Factory.StartNew(new Action(() =>            {                //IsProcessAccpetRuning = false;                //Thread.Sleep(100);                //IsProcessAccpetRuning = true;                while (IsProcessAccpetRuning)                {                    if (testList.Count > 0)                    {                        List<byte> tempList = new List<byte>();                        lock ("我要给集合增加数据")                        {                            tempList = testList.ToList();                            testList.Clear();                        }                        //启动一个任务来发送数据                        Task.Factory.StartNew(new Action(() =>                        {                            var result = pmsMessage.ReversePacketMessage(tempList, tempList.Count);                            //发送给客户端                            ClientSend(result);                            //发送给测试端                            if (MyAction != null && result != null && result.Count > 0)                            {                                MyAction(result.ToList());                            }                        }));                    }                    else                    {                        Thread.Sleep(50);                    }                }                LogHelper.WriteLog("新的处理请求，停止处理接受");            }));        }        #endregion        #region 保持连接        /// <summary>        /// 重置连接        /// </summary>        public void ResetConnect()        {            if (!_isResetConnecting)            {                lock (lockObject)                {                    if (!_isResetConnecting)                    {                        _isResetConnecting = true;                        LogHelper.WriteLog("=========重新连接==============");                        lock (lockObject)                        {                            this.Dispose();                            InitSocket();                        }                        Thread.Sleep(10000);                        Run();                        _isResetConnecting = false;                    }                }            }        }        private Timer timer = null;        /// <summary>        /// 保持连接        /// </summary>        private void KeepConnect()        {            timer = new Timer(new TimerCallback((o) =>            {                if (IsConnected)                {                    //发送心跳包                    SendAsync(pmsMessage.GetHeardbeat());                    LogHelper.WriteLog("已经发送心跳包");                }            }), null, new TimeSpan(0, 0, 0, 5), new TimeSpan(0, 0, 0, 10));        }        #endregion        #region 通知相关        private ChatClient client = null;// ChatClient.Instance(ConfigurationManager.AppSettings["notificationUri"]);        PacketConvert pmsMessage = new PacketConvert();        private bool _isResetConnecting = false;        /// <summary>        /// 请求发送消息        /// </summary>        /// <param name="msg"></param>        protected void client_OnMsgReceiveed(WebMessageDTO msg)        {            //请求发送消息            LogHelper.WriteLog(string.Format("开始请求发送消息:消息id=>{0} **********消息内容=>:{1}", msg.MsgId, msg.Msg != null ? msg.Msg.Message : ""));            byte[] bytes = pmsMessage.PacketMessage(msg);            SendAsync(bytes);            //发送给测试程序            if (receiveAction != null)            {                receiveAction(msg);            }        }        private void ClientSend(IList<MessageTemplate> message)        {            if (message != null && client != null)                client.SendAsync(message);        }        #endregion        #region 资源清理        /// <summary>        /// 清理资源        /// </summary>        public void Dispose()        {            IsRegisterApp = false;            IsHandStake = false;            IsConnected = false;            if (_socket.Connected == true)                _socket.Shutdown(SocketShutdown.Both);            _socket.Dispose();            _socket = null;            pool = null;            m_bufferManager = null;        }        #endregion    }}