public void Decode(IoSession session, IoBuffer input, IProtocolDecoderOutput output)
 {
     lock (_decoder)
     {
         _decoder.Decode(session, input, output);
     }
 }

 public bool IsResponse(IoSession session, object message)
 {
     var isResponse = false;
     var heartbeat = message as DuplexMessage;
     if (heartbeat != null) isResponse = heartbeat.Header.MessageType == MessageType.Heartbeat;
     return isResponse;
 }

 public void FinishDecode(IoSession session, IProtocolDecoderOutput output)
 {
     lock (_decoder)
     {
         _decoder.FinishDecode(session, output);
     }
 }

 public bool IsRequest(IoSession session, object message)
 {
     var isRequest = false;
     var heartbeat = message as DuplexMessage;
     if (heartbeat != null) isRequest = heartbeat.Header.CommandCode == CommandCode.Heartbeat;
     return isRequest;
 }

 public IoFilterEvent(INextFilter nextFilter, IoEventType eventType, IoSession session, Object parameter)
     : base(eventType, session, parameter)
 {
     if (nextFilter == null)
         throw new ArgumentNullException("nextFilter");
     _nextFilter = nextFilter;
 }

 /// <inheritdoc/>
 public virtual void ExceptionCaught(IoSession session, Exception cause)
 {
     if (log.IsWarnEnabled)
     {
         log.WarnFormat("EXCEPTION, please implement {0}.ExceptionCaught() for proper handling: {1}", GetType().Name, cause);
     }
 }

        public void Decode(IoSession session, IoBuffer input, IProtocolDecoderOutput output)
        {
            if (_session == null)
                _session = session;
            else if (_session != session)
                throw new InvalidOperationException(GetType().Name + " is a stateful decoder.  "
                        + "You have to create one per session.");

            _undecodedBuffers.Enqueue(input);
            while (true)
            {
                IoBuffer b;
                if (!_undecodedBuffers.TryPeek(out b))
                    break;

                Int32 oldRemaining = b.Remaining;
                _state.Decode(b, output);
                Int32 newRemaining = b.Remaining;
                if (newRemaining != 0)
                {
                    if (oldRemaining == newRemaining)
                        throw new InvalidOperationException(_state.GetType().Name
                            + " must consume at least one byte per decode().");
                }
                else
                {
                    _undecodedBuffers.TryDequeue(out b);
                }
            }
        }

      public SampleCommand(
          IoSession session,
          TimeoutNotifyProducerConsumer<AbstractAsyncCommand> producer)
          :base(session, CommandCode.Test, producer)
      {
 
      }

        protected override AbstractMessage DecodeBody(IoSession session, IoBuffer input)
        {
            if (!_readCode)
            {
                if (input.Remaining < Constants.RESULT_CODE_LEN)
                {
                    return null; // Need more data.
                }

                _code = input.GetInt16();
                _readCode = true;
            }

            if (_code == Constants.RESULT_OK)
            {
                if (input.Remaining < Constants.RESULT_VALUE_LEN)
                {
                    return null;
                }

                ResultMessage m = new ResultMessage();
                m.OK = true;
                m.Value = input.GetInt32();
                _readCode = false;
                return m;
            }
            else
            {
                ResultMessage m = new ResultMessage();
                m.OK = false;
                _readCode = false;
                return m;
            }
        }

        /// <summary>
        /// Method responsible for deciding if a connection is OK to continue.
        /// </summary>
        /// <param name="session">the new session that will be verified</param>
        /// <returns>true if the session meets the criteria, otherwise false</returns>
        public Boolean IsConnectionOk(IoSession session)
        {
            IPEndPoint ep = session.RemoteEndPoint as IPEndPoint;
            if (ep != null)
            {
                String addr = ep.Address.ToString();
                DateTime now = DateTime.Now;
                DateTime? lastConnTime = null;

                _clients.AddOrUpdate(addr, now, (k, v) =>
                {
                    if (log.IsDebugEnabled)
                        log.Debug("This is not a new client");
                    lastConnTime = v;
                    return now;
                });

                if (lastConnTime.HasValue)
                {
                    // if the interval between now and the last connection is
                    // less than the allowed interval, return false
                    if ((now - lastConnTime.Value).TotalMilliseconds < _allowedInterval)
                    {
                        if (log.IsWarnEnabled)
                            log.Warn("Session connection interval too short");
                        return false;
                    }
                }

                return true;
            }

            return false;
        }

        public MessageDecoderResult Decodable(IoSession session, IoBuffer input)
        {
            if ((MessageType)input.Get() != MessageType.Update_ZipFiles)
            {
                return MessageDecoderResult.NotOK;
            }

            var zipFileInfo = JsonConvert.DeserializeObject<TransferingZipFileInfo>(input.GetString(Encoding.UTF8));
            var fileSize = zipFileInfo.FileSize;
            var hashBytes = zipFileInfo.HashBytes;
            if (input.Remaining < fileSize)
            {
                return MessageDecoderResult.NeedData;
            }

            var filesBytes = new byte[fileSize];
            input.Get(filesBytes, 0, (int)fileSize);

            if (FileHashHelper.CompareHashValue(FileHashHelper.ComputeFileHash(filesBytes), hashBytes))
            {
                _zipFileInfoMessage = new TransferingZipFile(filesBytes);
                return MessageDecoderResult.OK;
            }
            return MessageDecoderResult.NotOK;
        }

 public void Dispose(IoSession session)
 {
     lock (_encoder)
     {
         _encoder.Dispose(session);
     }
 }

        public MessageDecoderResult Decode(IoSession session, IoBuffer input, IProtocolDecoderOutput output)
        {
            // Try to skip header if not read.
            if (!_readHeader)
            {
                input.GetInt16(); // Skip 'type'.
                _sequence = input.GetInt32(); // Get 'sequence'.
                _readHeader = true;
            }

            // Try to decode body
            AbstractMessage m = DecodeBody(session, input);
            // Return NEED_DATA if the body is not fully read.
            if (m == null)
            {
                return MessageDecoderResult.NeedData;
            }
            else
            {
                _readHeader = false; // reset readHeader for the next decode
            }
            m.Sequence = _sequence;
            output.Write(m);

            return MessageDecoderResult.OK;
        }

        private void Write(IoSession session, IoBuffer data, IoBuffer buf)
        {
            try
            {
                Int32 len = data.Remaining;
                if (len >= buf.Capacity)
                {
                    /*
                     * If the request length exceeds the size of the output buffer,
                     * flush the output buffer and then write the data directly.
                     */
                    INextFilter nextFilter = session.FilterChain.GetNextFilter(this);
                    InternalFlush(nextFilter, session, buf);
                    nextFilter.FilterWrite(session, new DefaultWriteRequest(data));
                    return;
                }
                if (len > (buf.Limit - buf.Position))
                {
                    InternalFlush(session.FilterChain.GetNextFilter(this), session, buf);
                }

                lock (buf)
                {
                    buf.Put(data);
                }
            }
            catch (Exception e)
            {
                session.FilterChain.FireExceptionCaught(e);
            }
        }

 public void Put(IoSession session)
 {
     _sessionMap.StartExpiring();
     EndPoint key = session.RemoteEndPoint;
     if (!_sessionMap.ContainsKey(key))
         _sessionMap.Add(key, session);
 }

 public void Encode(IoSession session, Object message, IProtocolEncoderOutput output)
 {
     lock (_encoder)
     {
         _encoder.Encode(session, message, output);
     }
 }

        /// <inheritdoc/>
        public override void MessageReceived(INextFilter nextFilter, IoSession session, Object message)
        {
            //if (log.IsDebugEnabled)
            //    log.DebugFormat("Processing a MESSAGE_RECEIVED for session {0}", session.Id);

            IoBuffer input = message as IoBuffer;
            if (input == null)
            {
                nextFilter.MessageReceived(session, message);
                return;
            }

            IProtocolDecoder decoder = _factory.GetDecoder(session);
            IProtocolDecoderOutput decoderOutput = GetDecoderOut(session, nextFilter);

            // Loop until we don't have anymore byte in the buffer,
            // or until the decoder throws an unrecoverable exception or
            // can't decoder a message, because there are not enough
            // data in the buffer
            while (input.HasRemaining)
            {
                Int32 oldPos = input.Position;
                try
                {
                    // TODO may not need lock on UDP
                    lock (session)
                    {
                        // Call the decoder with the read bytes
                        decoder.Decode(session, input, decoderOutput);
                    }

                    // Finish decoding if no exception was thrown.
                    decoderOutput.Flush(nextFilter, session);
                }
                catch (Exception ex)
                {
                    ProtocolDecoderException pde = ex as ProtocolDecoderException;
                    if (pde == null)
                        pde = new ProtocolDecoderException(null, ex);
                    if (pde.Hexdump == null)
                    {
                        // Generate a message hex dump
                        Int32 curPos = input.Position;
                        input.Position = oldPos;
                        pde.Hexdump = input.GetHexDump();
                        input.Position = curPos;
                    }

                    decoderOutput.Flush(nextFilter, session);
                    nextFilter.ExceptionCaught(session, pde);

                    // Retry only if the type of the caught exception is
                    // recoverable and the buffer position has changed.
                    // We check buffer position additionally to prevent an
                    // infinite loop.
                    if (!(ex is RecoverableProtocolDecoderException) || input.Position == oldPos)
                        break;
                }
            }
        }

 /// <inheritdoc/>
 public override void MessageSent(INextFilter nextFilter, IoSession session, IWriteRequest writeRequest)
 {
     if (IsBlocked(session))
         BlockSession(session);
     else
         // forward if not blocked
         base.MessageSent(nextFilter, session, writeRequest);
 }

 /// <inheritdoc/>
 public override void MessageReceived(INextFilter nextFilter, IoSession session, Object message)
 {
     if (IsBlocked(session))
         BlockSession(session);
     else
         // forward if not blocked
         base.MessageReceived(nextFilter, session, message);
 }

 /// <inheritdoc/>
 public override void SessionIdle(INextFilter nextFilter, IoSession session, IdleStatus status)
 {
     if (IsBlocked(session))
         BlockSession(session);
     else
         // forward if not blocked
         base.SessionIdle(nextFilter, session, status);
 }