subsmgserver.cs

联通的SGIP发送代码

        {
            StateObject obj = (StateObject)state;
            _userConnection.Logout(obj.LoginName, obj.WorkSocket);
            //释放等待unbind_resp命令超时的定时器
            lock (_waitUnbindRespTimer)
            {
                if (_waitUnbindRespTimer.ContainsKey(obj))
                {
                    _waitUnbindRespTimer[obj].Dispose();
                    _waitUnbindRespTimer.Remove(obj);
                }
            }
            obj.Dispose();
        }

        /// <summary>
        /// 发送unbind命令给客户端，在异步事件中等待unbind_resp命令，如果超时也将关闭socket
        /// </summary>
        /// <param name="obj"></param>
        private void ForceDisconnectClient(StateObject obj)
        {
            try
            {
                //开始计时，如果未收到unbind_resp应答，则自动调用ProcessUnbind_respCommand中断连接
                lock (_waitUnbindRespTimer)
                {
                    if (_waitUnbindRespTimer.ContainsKey(obj))
                    {
                        _waitUnbindRespTimer[obj].Dispose();
                    }
                    _waitUnbindRespTimer[obj] = new Timer(new TimerCallback(ProcessUnbind_respCommand), obj, _answerTimeOut, Timeout.Infinite);
                }
                obj.WorkSocket.Send(Command.CreateCommand(new Unbind()).GetBytes());
            }catch(Exception e){
                System.Diagnostics.Debug.Write(e);
                ProcessUnbind_respCommand(obj);
            }
        }

        /// <summary>
        /// 客户端长时间没有发送消息，主动断开之
        /// </summary>
        /// <param name="state"></param>
        private void OnClientWaitTooLong(object state)
        {
            //主动断开Socket连接
            ForceDisconnectClient((StateObject)state);
        }

        /// <summary>
        /// 一个连接断开了，则注销这个连接的登录，并释放资源
        /// </summary>
        /// <param name="obj"></param>
        private void LongoutAndDispose(StateObject obj)
        {
            _userConnection.Logout(obj.LoginName, obj.WorkSocket);
            obj.Dispose();
        }

        private void ReceiveClientData(StateObject obj)
        {
            //队列满了，暂停2秒; 考虑：是否继续接收，并返回服务器忙应答?
            while (IsSocketQueueFull(obj))
            {
                Console.WriteLine(this.GetType().FullName + ":socket队列已满，要暂停2秒了!");
                System.Diagnostics.Debug.WriteLine(this.GetType().FullName + ":socket队列已满，要暂停2秒了!");
                Thread.Sleep(2000);
            }
            obj.Timer.Change(_idleTimeOut, Timeout.Infinite); //开始记时
            try
            {
                SgipHelper.ClearCommandBuffer(obj.Buffer);
                obj.WorkSocket.BeginReceive(obj.Buffer, 0, obj.Buffer.Length, SocketFlags.None, new AsyncCallback(OnReceiveDataCallBack), obj);
            }
            catch (Exception e)
            {
                //如果在调用obj.WorkSocket的BeginReceive方法之前远程主机强制断开了，则会发生异常
                System.Diagnostics.Debug.WriteLine(e);
                LongoutAndDispose(obj);
            }
        }

        private void OnReceiveDataCallBack(IAsyncResult ar)
        {
            StateObject obj = (StateObject)ar.AsyncState;
            int nBytesRead = 0;
            try
            {
                nBytesRead = obj.WorkSocket.EndReceive(ar);
            }
            catch (Exception e)
            {
                System.Diagnostics.Debug.WriteLine(e);
            }

            //连接已断开
            if (nBytesRead <= 0)
            {
                LongoutAndDispose(obj);
                return;
            }

            obj.Timer.Change(Timeout.Infinite, Timeout.Infinite); //停止记时
            Command cmd = null;
            try
            {
                cmd = Command.ToCommand(obj.Buffer, 0);
            }catch (Exception e){
                System.Diagnostics.Debug.WriteLine(e);
            }

            //收到非法的命令
            if (cmd == null)
            {
                ReceiveClientData(obj);
                return;
            }

            switch (cmd.Head.CommandID)
            {
                case Command.SGIP_UNBIND:
                    ProcessUnbindCommand(obj, cmd);
                    return;
                case Command.SGIP_UNBIND_RESP:
                    ProcessUnbind_respCommand(obj);
                    return;
                case Command.SGIP_BIND:   //重复登录
                    SendBind_Resp(obj.WorkSocket, Command.ERR_RepeatLogin, cmd);
                    break;
                default:
                    try
                    {
                        ProcessOtherCommand(obj, cmd); //由继承类来处理其它命令
                    }catch (Exception e){
                        System.Diagnostics.Debug.WriteLine(e);
                    }
                    break;
            }
            ReceiveClientData(obj);
        }
    }

    public class SubmitListenServer : ListenServer
    {
        private SocketCommandQueue _submitQueue;
        private Transmit _transmit;

        public SubmitListenServer() : base(SgipConfig.Submit.IP, SgipConfig.Submit.Port, 
            SgipConfig.Submit.MaxConnection, SgipConfig.Submit.AnswerTimeOut, SgipConfig.Submit.IdleTimeOut)
        {
            _submitQueue = new SocketCommandQueue(SgipConfig.Submit.MaxWaitAnswerCommand);
            
            Bind bindBody = new Bind();
            bindBody.LoginName = SgipConfig.SP.UserName;
            bindBody.LoginPassword = SgipConfig.SP.Password;
            bindBody.LoginType = 1;
            _transmit = new Transmit(SgipConfig.SMG.IP, SgipConfig.SMG.Port, SgipConfig.SMG.MaxConnection,
                SgipConfig.SMG.MaxWaitAnswerCommand, SgipConfig.SMG.AnswerTimeOut, bindBody, _submitQueue);
        }

        protected override bool IsValidateLogin(Socket client, Bind bindBody)
        {
            if (bindBody.LoginType != 1)
            {
                return false;
            }
            SubSP subSP = SubSPManager.Manager.FindByLoginName(bindBody.LoginName);
            //注：是不是还应该判断对方的ip呢?
            return (subSP != null) && (subSP.Password == bindBody.LoginPassword);
        }

        private volatile int nTestSumbitCount = 0;

        protected override void ProcessOtherCommand(ListenServer.StateObject obj, Command cmd)
        {
            switch (cmd.Head.CommandID)
            {
                case Command.SGIP_SUBMIT:
                    Console.WriteLine("total sumbit count:{0}", ++nTestSumbitCount);
                    ProcessSubmitCommand(obj, cmd);
                    break;
                case Command.SGIP_TRACE:
                    ProcessTraceCommand(obj, cmd);
                    break;
                default:
                    break;
            }
        }

        protected override bool IsSocketQueueFull(ListenServer.StateObject obj)
        {
            return _submitQueue.IsSocketQueueFull(obj.WorkSocket);
        }

        /// <summary>
        ///  判断一个命令是否已存在于队列中，如果不存在则回应错误代码
        /// </summary>
        /// <param name="obj"></param>
        /// <param name="cmd"></param>
        /// <returns></returns>
        private bool IsCommandInQueue(StateObject obj, Command cmd)
        {
            QueueItem Item = _submitQueue.FindItem(cmd.Head.SequenceNumber);
            if (Item != null) //序列号已存在
            {
                //不判断命令内容是否相同, 直接返回错误代码.××××××有待改时进
                Submit_Resp resp = new Submit_Resp();
                resp.Result = Command.ERR_InvalidSerialNo;
                try
                {
                    obj.WorkSocket.Send(Command.CreateCommand(resp).GetBytes());
                }catch (Exception e){
                    System.Diagnostics.Debug.WriteLine(e);
                }
                return true;
            }
            return false;
        }

        /// <summary>
        /// 将命令加入到发送队列并发送
        /// </summary>
        /// <param name="obj"></param>
        /// <param name="cmd"></param>
        private void AddToQueueAndSendData(StateObject obj, Command cmd)
        {
            _submitQueue.Enqueue(obj.WorkSocket, cmd);
            _transmit.SendData();
        }

        /// <summary>
        /// 处理客户端发来的Submit命令, 把该命令加入到发送队列当中
        /// </summary>
        /// <param name="obj"></param>
        /// <param name="cmd"></param>
        private void ProcessSubmitCommand(StateObject obj, Command cmd)
        {
            if (IsCommandInQueue(obj, cmd))
            {
                return;
            }
            //如果需要状态报告, 则保存：序列号--用户名(2不需要, 其它值都可能需要)
            if (((Submit)cmd.Body).ReportFlag != 2)
            {
                s_reportDict.Add(cmd.Head.SequenceNumber.ToString(), obj.LoginName);
            }
            AddToQueueAndSendData(obj, cmd);
        }

        private void ProcessTraceCommand(StateObject obj, Command cmd)
        {
            if (IsCommandInQueue(obj, cmd))
            {
                return;
            }
            //Trace状态报告, 保存：序列号--用户名
            s_traceDict.Add(cmd.Head.SequenceNumber.ToString(), obj.LoginName);
            AddToQueueAndSendData(obj, cmd);
        }
    }

    public class DeliverListenServer : ListenServer
    {
        class DictData
        {
            private SocketCommandQueue _queue;