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

 public override void OnPostRemove(IoFilterChain parent, String name, INextFilter nextFilter)
 {
     _filter.OnPostRemove(parent, name, nextFilter);
     _count--;
     if (_count == 0)
         _filter.Destroy();
 }

        /// <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;
                }
            }
        }

 public override void OnPreAdd(IoFilterChain parent, String name, INextFilter nextFilter)
 {
     if (_count == 0)
         _filter.Init();
     _count++;
     _filter.OnPreAdd(parent, name, nextFilter);
 }

 /// <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);
 }

 /// <inheritdoc/>
 public override void FilterWrite(INextFilter nextFilter, IoSession session, IWriteRequest writeRequest)
 {
     IoBuffer buf = writeRequest.Message as IoBuffer;
     if (buf == null)
         throw new ArgumentException("This filter should only buffer IoBuffer objects");
     else
         Write(session, buf);
 }

 /// <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 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 SessionClosed(INextFilter nextFilter, IoSession session)
 {
     if (IsBlocked(session))
         BlockSession(session);
     else
         // forward if not blocked
         base.SessionClosed(nextFilter, session);
 }

        public override void SessionCreated(INextFilter nextFilter, IoSession session)
        {
            foreach (KeyValuePair<String, Object> pair in _attributes)
            {
                session.SetAttribute(pair.Key, pair.Value);
            }

            base.SessionCreated(nextFilter, session);
        }

 public override void SessionCreated(INextFilter nextFilter, IoSession session)
 {
     if (!IsConnectionOk(session))
     {
         Debug.WriteLine("Connections coming in too fast; closing.");
         session.Close(true);
     }
     base.SessionCreated(nextFilter, session);
 }

 /// <inheritdoc/>
 public override void SessionCreated(INextFilter nextFilter, IoSession session)
 {
     if (!IsConnectionOk(session))
     {
         if (log.IsWarnEnabled)
             log.Warn("Connections coming in too fast; closing.");
         session.Close(true);
     }
     base.SessionCreated(nextFilter, session);
 }

        public override void OnPreAdd(IoFilterChain parent, String name, INextFilter nextFilter)
        {
            if (parent.Contains<SslFilter>())
                throw new InvalidOperationException("Only one SSL filter is permitted in a chain.");

            IoSession session = parent.Session;
            session.SetAttribute(NEXT_FILTER, nextFilter);
            // Create a SSL handler and start handshake.
            SslHandler handler = new SslHandler(this, session);
            session.SetAttribute(SSL_HANDLER, handler);
        }

 public override void ExceptionCaught(INextFilter nextFilter, IoSession session, Exception cause)
 {
     Log.Error(ErrorCode.UnhandledExceptionCaught, cause);
     var counter = session.GetAttribute<int>(KeyName.SESSION_ERROR_COUNTER);
     if (counter >= MAX_UNCAUGHT_SESSION_ERROR_COUNT) session.Close(true);
     else
     {
         session.SetAttribute(KeyName.SESSION_ERROR_COUNTER, ++counter);
         base.ExceptionCaught(nextFilter, session, cause);
     }
 }

 public override void SessionIdle(INextFilter nextFilter, IoSession session, IdleStatus status)
 {
     if ((_eventTypes & IoEventType.SessionIdle) == IoEventType.SessionIdle)
     {
         IoFilterEvent ioe = new IoFilterEvent(nextFilter, IoEventType.SessionIdle, session, status);
         FireEvent(ioe);
     }
     else
     {
         base.SessionIdle(nextFilter, session, status);
     }
 }

 /// <inheritdoc/>
 public override void SessionOpened(INextFilter nextFilter, IoSession session)
 {
     if ((_eventTypes & IoEventType.SessionOpened) == IoEventType.SessionOpened)
     {
         IoFilterEvent ioe = new IoFilterEvent(nextFilter, IoEventType.SessionOpened, session, null);
         FireEvent(ioe);
     }
     else
     {
         base.SessionOpened(nextFilter, session);
     }
 }

 public override void SessionClosed(INextFilter nextFilter, IoSession session)
 {
     SslHandler handler = GetSslSessionHandler(session);
     try
     {
         // release resources
         handler.Destroy();
     }
     finally
     {
         // notify closed session
         base.SessionClosed(nextFilter, session);
     }
 }

        /// <inheritdoc/>
        public override void FilterWrite(INextFilter nextFilter, IoSession session, IWriteRequest writeRequest)
        {
            IoEvent ioe = new IoEvent(IoEventType.Write, session, writeRequest);
            if (_queueHandler.Accept(this, ioe))
            {
                nextFilter.FilterWrite(session, writeRequest);
                IWriteFuture writeFuture = writeRequest.Future;
                if (writeFuture == null)
                    return;

                // We can track the write request only when it has a future.
                _queueHandler.Offered(this, ioe);
                writeFuture.Complete += (s, e) => _queueHandler.Polled(this, ioe);
            }
        }

            public override void FilterWrite(INextFilter nextFilter, IoSession session, IWriteRequest writeRequest)
            {
                _counter++;

                IoBuffer buf = writeRequest.Message as IoBuffer;
                if (buf == null)
                    throw new AssertFailedException("Wrong message type");
                if (_counter == 3)
                {
                    Assert.AreEqual(1, buf.Limit);
                    Assert.AreEqual(0, buf.Get());
                }
                else
                {
                    Assert.AreEqual(10, buf.Limit);
                }
            }

 public override void SessionCreated(INextFilter nextFilter, IoSession session)
 {
     Metric.Gauge("01.message.read.count", () => session.Service.Statistics.ReadMessages, Unit.Requests);
     Metric.Gauge("02.message.read.bytes", () => session.Service.Statistics.ReadBytes / 1024 / 1024, Unit.MegaBytes);
     Metric.Gauge("03.message.read.throughput", () => session.Service.Statistics.ReadMessagesThroughput, Unit.Requests);
     Metric.Gauge("04.message.read.throughput.max", () => session.Service.Statistics.LargestReadMessagesThroughput, Unit.Requests);
     Metric.Gauge("05.message.read.bytes.throughput", () => session.Service.Statistics.ReadBytesThroughput / 1024, Unit.KiloBytes);
     Metric.Gauge("06.message.read.bytes.throughput.max", () => session.Service.Statistics.LargestReadBytesThroughput / 1024, Unit.KiloBytes);
     Metric.Gauge("07.message.write.count", () => session.Service.Statistics.WrittenMessages, Unit.Results);
     Metric.Gauge("08.message.write.bytes", () => session.Service.Statistics.WrittenBytes / 1024 / 1024, Unit.MegaBytes);
     Metric.Gauge("09.message.write.throughput", () => session.Service.Statistics.WrittenMessagesThroughput, Unit.Results);
     Metric.Gauge("10.message.write.throughput.max", () => session.Service.Statistics.LargestWrittenMessagesThroughput, Unit.Requests);
     Metric.Gauge("11.message.write.bytes.throughput", () => session.Service.Statistics.WrittenBytesThroughput / 1024, Unit.KiloBytes);
     Metric.Gauge("12.message.write.bytes.throughput.max", () => session.Service.Statistics.LargestWrittenBytesThroughput / 1024, Unit.KiloBytes);
     Metric.Gauge("13.message.write.queue.count", () => session.Service.Statistics.ScheduledWriteMessages, Unit.Results);
     Metric.Gauge("14.message.write.queue.bytes", () => session.Service.Statistics.ScheduledWriteBytes / 1024, Unit.KiloBytes);
     base.SessionCreated(nextFilter, session);
 }