internal void InitConnect(IPEndPoint serverEndPoint,
                                  Action<IPEndPoint, Socket> onConnectionEstablished,
                                  Action<IPEndPoint, SocketError> onConnectionFailed,
                                  ITcpConnection connection,
                                  TimeSpan connectionTimeout)
        {
            if (serverEndPoint == null)
                throw new ArgumentNullException("serverEndPoint");
            if (onConnectionEstablished == null)
                throw new ArgumentNullException("onConnectionEstablished");
            if (onConnectionFailed == null)
                throw new ArgumentNullException("onConnectionFailed");

            var socketArgs = _connectSocketArgsPool.Get();
            var connectingSocket = new Socket(serverEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            socketArgs.RemoteEndPoint = serverEndPoint;
            socketArgs.AcceptSocket = connectingSocket;
            var callbacks = (CallbacksStateToken) socketArgs.UserToken;
            callbacks.OnConnectionEstablished = onConnectionEstablished;
            callbacks.OnConnectionFailed = onConnectionFailed;
            callbacks.PendingConnection = new PendingConnection(connection, DateTime.UtcNow.Add(connectionTimeout));

            AddToConnecting(callbacks.PendingConnection);

            try
            {
                var firedAsync = connectingSocket.ConnectAsync(socketArgs);
                if (!firedAsync)
                    ProcessConnect(socketArgs);
            }
            catch (ObjectDisposedException)
            {
                HandleBadConnect(socketArgs);
            }
        }

 private void OnConnectionClosed(ITcpConnection connection, SocketError socketError)
 {
     if (Interlocked.CompareExchange(ref _isClosed, 1, 0) == 0)
     {
         Console.WriteLine("Connection '{0}' [{1:B}] closed: {2}.", connection.ConnectionId, connection.RemoteEndPoint, socketError);
     }
 }

 public void RegisterConsumer(string groupName, string consumerId, IEnumerable<string> subscriptionTopics, IEnumerable<string> consumingQueues, ITcpConnection connection)
 {
     var consumerGroup = _consumerGroupDict.GetOrAdd(groupName, key => new ConsumerGroup(key));
     consumerGroup.Register(consumerId, connection);
     consumerGroup.UpdateConsumerSubscriptionTopics(consumerId, subscriptionTopics);
     consumerGroup.UpdateConsumerConsumingQueues(consumerId, consumingQueues);
 }

        private void HandleRemotingRequest(ITcpConnection connection, byte[] message, Action<byte[]> sendReplyAction)
        {
            var remotingRequest = RemotingUtil.ParseRequest(message);
            var requestHandlerContext = new SocketRequestHandlerContext(connection, sendReplyAction);

            IRequestHandler requestHandler;
            if (!_requestHandlerDict.TryGetValue(remotingRequest.Code, out requestHandler))
            {
                var errorMessage = string.Format("No request handler found for remoting request, request code:{0}", remotingRequest.Code);
                _logger.Error(errorMessage);
                requestHandlerContext.SendRemotingResponse(new RemotingResponse(-1, remotingRequest.Sequence, Encoding.UTF8.GetBytes(errorMessage)));
                return;
            }

            try
            {
                _logger.DebugFormat("Handling remoting request, request code:{0}, request sequence:{1}", remotingRequest.Code, remotingRequest.Sequence);
                var remotingResponse = requestHandler.HandleRequest(requestHandlerContext, remotingRequest);
                if (!remotingRequest.IsOneway && remotingResponse != null)
                {
                    requestHandlerContext.SendRemotingResponse(remotingResponse);
                }
            }
            catch (Exception ex)
            {
                var errorMessage = string.Format("Exception raised when handling remoting request, request code:{0}.", remotingRequest.Code);
                _logger.Error(errorMessage, ex);
                if (!remotingRequest.IsOneway)
                {
                    requestHandlerContext.SendRemotingResponse(new RemotingResponse(-1, remotingRequest.Sequence, Encoding.UTF8.GetBytes(ex.Message)));
                }
            }
        }

 private void OnConnectionFailed(ITcpConnection connection, SocketError socketError)
 {
     if (Interlocked.CompareExchange(ref _isClosed, 1, 0) == 0)
     {
         Console.WriteLine("Connection '{0}' ({1:B}) to [{2}] failed: {3}.", _connectionName, _connectionId, connection.RemoteEndPoint, socketError);
     }
 }

        private void OnConnectionClosed(ITcpConnection connection, SocketError socketError)
        {
            connection.ConnectionClosed -= OnConnectionClosed;

            var handler = ConnectionClosed;
            if (handler != null)
                handler(this, socketError);
        }

 public SocketRequestHandlerContext(ITcpConnection connection, Action<byte[]> sendReplyAction)
 {
     Channel = new SocketChannel(connection);
     SendRemotingResponse = remotingResponse =>
     {
         sendReplyAction(RemotingUtil.BuildResponseMessage(remotingResponse));
     };
 }

 public void Init(ITcpConnection connection, ICommandFactory commandFactory, int id, Car car)
 {
     Connection = connection;
     Id = id;
     Car = car;
     CommandFactory = commandFactory;
     _initilized = true;
 }

 public void Register(string consumerId, ITcpConnection connection)
 {
     var consumerHeartbeatInfo = _consumerDict.GetOrAdd(consumerId, key =>
     {
         _logger.InfoFormat("Consumer registered to group: {0}, consumerId: {1}", _groupName, consumerId);
         return new ConsumerHeartbeatInfo(key, connection);
     });
     consumerHeartbeatInfo.LastHeartbeatTime = DateTime.Now;
 }

        private void HandleRemotingRequest(ITcpConnection connection, byte[] message, Action<byte[]> sendReplyAction)
        {
            if (_isShuttingdown) return;

            var remotingRequest = RemotingUtil.ParseRequest(message);
            var requestHandlerContext = new SocketRequestHandlerContext(connection, sendReplyAction);

            IRequestHandler requestHandler;
            if (!_requestHandlerDict.TryGetValue(remotingRequest.Code, out requestHandler))
            {
                var errorMessage = string.Format("No request handler found for remoting request:{0}", remotingRequest);
                _logger.Error(errorMessage);
                if (remotingRequest.Type != RemotingRequestType.Oneway)
                {
                    requestHandlerContext.SendRemotingResponse(new RemotingResponse(
                        remotingRequest.Type,
                        remotingRequest.Code,
                        remotingRequest.Sequence,
                        remotingRequest.CreatedTime,
                        -1,
                        Encoding.UTF8.GetBytes(errorMessage),
                        DateTime.Now,
                        remotingRequest.Header,
                        null));
                }
                return;
            }

            try
            {
                var remotingResponse = requestHandler.HandleRequest(requestHandlerContext, remotingRequest);
                if (remotingRequest.Type != RemotingRequestType.Oneway && remotingResponse != null)
                {
                    requestHandlerContext.SendRemotingResponse(remotingResponse);
                }
            }
            catch (Exception ex)
            {
                var errorMessage = string.Format("Unknown exception raised when handling remoting request:{0}.", remotingRequest);
                _logger.Error(errorMessage, ex);
                if (remotingRequest.Type != RemotingRequestType.Oneway)
                {
                    requestHandlerContext.SendRemotingResponse(new RemotingResponse(
                        remotingRequest.Type,
                        remotingRequest.Code,
                        remotingRequest.Sequence,
                        remotingRequest.CreatedTime,
                        -1,
                        Encoding.UTF8.GetBytes(ex.Message),
                        DateTime.Now,
                        remotingRequest.Header,
                        null));
                }
            }
        }

 public void RegisterProducer(string producerId, ITcpConnection connection)
 {
     _producerDict.AddOrUpdate(producerId, key =>
     {
         _logger.InfoFormat("Producer registered, producerId: {0}", key);
         return new ProducerHeartbeatInfo(connection) { LastHeartbeatTime = DateTime.Now };
     }, (key, existing) =>
     {
         existing.LastHeartbeatTime = DateTime.Now;
         return existing;
     });
 }

        public void RegisterBroker(ITcpConnection connection, BrokerRegistrationRequest request)
        {
            lock (_lockObj)
            {
                var brokerInfo = request.BrokerInfo;
                var cluster = _clusterDict.GetOrAdd(brokerInfo.ClusterName, x => new Cluster { ClusterName = x });
                var brokerGroup = cluster.BrokerGroups.GetOrAdd(brokerInfo.GroupName, x => new BrokerGroup { GroupName = x });
                Broker broker;
                if (!brokerGroup.Brokers.TryGetValue(brokerInfo.BrokerName, out broker))
                {
                    var connectionId = connection.RemotingEndPoint.ToAddress();
                    broker = new Broker
                    {
                        BrokerInfo = request.BrokerInfo,
                        TotalSendThroughput = request.TotalSendThroughput,
                        TotalConsumeThroughput = request.TotalConsumeThroughput,
                        TotalUnConsumedMessageCount = request.TotalUnConsumedMessageCount,
                        TopicQueueInfoList = request.TopicQueueInfoList,
                        TopicConsumeInfoList = request.TopicConsumeInfoList,
                        ProducerList = request.ProducerList,
                        ConsumerList = request.ConsumerList,
                        Connection = connection,
                        ConnectionId = connectionId,
                        LastActiveTime = DateTime.Now,
                        FirstRegisteredTime = DateTime.Now,
                        Group = brokerGroup
                    };
                    if (brokerGroup.Brokers.TryAdd(brokerInfo.BrokerName, broker))
                    {
                        _logger.InfoFormat("Registered new broker, brokerInfo: {0}", _jsonSerializer.Serialize(brokerInfo));
                    }
                }
                else
                {
                    broker.LastActiveTime = DateTime.Now;
                    broker.TotalSendThroughput = request.TotalSendThroughput;
                    broker.TotalConsumeThroughput = request.TotalConsumeThroughput;
                    broker.TotalUnConsumedMessageCount = request.TotalUnConsumedMessageCount;

                    if (!broker.BrokerInfo.IsEqualsWith(request.BrokerInfo))
                    {
                        var logInfo = string.Format("Broker basicInfo changed, old: {0}, new: {1}", broker.BrokerInfo, request.BrokerInfo);
                        broker.BrokerInfo = request.BrokerInfo;
                        _logger.Info(logInfo);
                    }

                    broker.TopicQueueInfoList = request.TopicQueueInfoList;
                    broker.TopicConsumeInfoList = request.TopicConsumeInfoList;
                    broker.ProducerList = request.ProducerList;
                    broker.ConsumerList = request.ConsumerList;
                }
            }
        }

 private void OnRawDataReceived(ITcpConnection connection, IEnumerable<ArraySegment<byte>> data)
 {
     try
     {
         _framer.UnFrameData(data);
     }
     catch (PackageFramingException exc)
     {
         Console.WriteLine(exc.Message);
         return;
     }
     connection.ReceiveAsync(OnRawDataReceived);
 }

        public AcceptedConnection(ITcpConnection connection)
        {
            _framer = new LengthPrefixMessageFramer();
            _framer.RegisterMessageArrivedCallback(OnMessageArrived);
            _connection = connection;

            _connection.ConnectionClosed += OnConnectionClosed;
            if (_connection.IsClosed)
            {
                OnConnectionClosed(_connection, SocketError.Success);
                return;
            }
        }

 private void OnConnectionClosed(ITcpConnection connection, SocketError socketError)
 {
     foreach (var listener in _connectionEventListeners)
     {
         try
         {
             listener.OnConnectionClosed(connection, socketError);
         }
         catch (Exception ex)
         {
             _logger.Error(string.Format("Notify connection closed failed, listener type:{0}", listener.GetType().Name), ex);
         }
     }
 }

 public void RegisterProducer(ITcpConnection connection, string producerId)
 {
     var connectionId = connection.RemotingEndPoint.ToAddress();
     _producerInfoDict.AddOrUpdate(connectionId, key =>
     {
         var producerInfo = new ProducerInfo
         {
             ProducerId = producerId,
             HeartbeatInfo = new ClientHeartbeatInfo(connection) { LastHeartbeatTime = DateTime.Now }
         };
         _logger.InfoFormat("Producer registered, producerId: {0}, connectionId: {1}", producerId, key);
         return producerInfo;
     }, (key, existingProducerInfo) =>
     {
         existingProducerInfo.HeartbeatInfo.LastHeartbeatTime = DateTime.Now;
         return existingProducerInfo;
     });
 }

        public void RegisterConsumer(ITcpConnection connection, string consumerId, IList<string> subscriptionTopics, IList<MessageQueueEx> consumingMessageQueues)
        {
            var connectionId = connection.RemotingEndPoint.ToAddress();

            _consumerInfoDict.AddOrUpdate(connectionId, key =>
            {
                var newConsumerInfo = new ConsumerInfo
                {
                    ConsumerId = consumerId,
                    HeartbeatInfo = new ClientHeartbeatInfo(connection) { LastHeartbeatTime = DateTime.Now },
                    SubscriptionTopics = subscriptionTopics,
                    ConsumingQueues = consumingMessageQueues
                };
                _logger.InfoFormat("Consumer registered to group, groupName: {0}, consumerId: {1}, connectionId: {2}, subscriptionTopics: {3}, consumingQueues: {4}", _groupName, consumerId, key, string.Join("|", subscriptionTopics), string.Join("|", consumingMessageQueues));
                return newConsumerInfo;
            },
            (key, existingConsumerInfo) =>
            {
                existingConsumerInfo.HeartbeatInfo.LastHeartbeatTime = DateTime.Now;

                var oldSubscriptionList = existingConsumerInfo.SubscriptionTopics.ToList();
                var newSubscriptionList = subscriptionTopics.ToList();
                if (IsStringCollectionChanged(oldSubscriptionList, newSubscriptionList))
                {
                    existingConsumerInfo.SubscriptionTopics = newSubscriptionList;
                    _logger.InfoFormat("Consumer subscriptionTopics changed. groupName: {0}, consumerId: {1}, connectionId: {2}, old: {3}, new: {4}", _groupName, consumerId, key, string.Join("|", oldSubscriptionList), string.Join("|", newSubscriptionList));
                }

                var oldConsumingQueues = existingConsumerInfo.ConsumingQueues;
                var newConsumingQueues = consumingMessageQueues;
                if (IsMessageQueueChanged(oldConsumingQueues, newConsumingQueues))
                {
                    existingConsumerInfo.ConsumingQueues = newConsumingQueues;
                    _logger.InfoFormat("Consumer consumingQueues changed. groupName: {0}, consumerId: {1}, connectionId: {2}, old: {3}, new: {4}", _groupName, consumerId, key, string.Join("|", oldConsumingQueues), string.Join("|", newConsumingQueues));
                }

                return existingConsumerInfo;
            });
        }

        private void OnDataReceived(ITcpConnection conn, IEnumerable<ArraySegment<byte>> data)
        {
            IMessageFramer framer;
            Tuple<ITcpConnection, IMessageFramer> pair;
            if (!_clientFramers.TryGetValue(conn.ConnectionId, out pair))
            {
                framer = new CrappyTemporaryFramer();
                framer.RegisterMessageArrivedCallback(CompleteMessageArrived);
                _clientFramers.TryAdd(conn.ConnectionId, new Tuple<ITcpConnection, IMessageFramer>(conn, framer));

                //Note: we stick the connection ID in the first part of the message just so we
                // can find it later. This isn't especially nice and is fixed in real code
                var connectionId = conn.ConnectionId.ToByteArray();
                framer.UnFrameData(new ArraySegment<byte>(connectionId, 0, connectionId.Length));
            }
            else
            {
                framer = pair.Item2;
            }

            framer.UnFrameData(data);
            conn.ReceiveAsync(OnDataReceived);
        }

 public PendingConnection(ITcpConnection connection, DateTime whenToKill)
 {
     Connection = connection;
     WhenToKill = whenToKill;
 }

 public void OnConnectionEstablished(ITcpConnection connection)
 {
     _consumer.StartBackgroundJobs();
 }