public override void ExceptionCaught(IChannelHandlerContext context, Exception exception)
        {
            if (!(exception is ObjectDisposedException))
                log.Warn($"Exception in NettyServerHandler {context.Channel.Id.AsShortText()}: {exception.Message}");

            context.CloseAsync();
        }

        //---------------------------------------------------------------------
        public override void ChannelActive(IChannelHandlerContext context)
        {
            var session = (GatewaySession)this.factory.createRpcSession(null);
            mapSession[context] = session;

            session.ChannelActive(context);
        }

        /// <summary>
        /// Process data that was received from the channel.
        /// </summary>
        /// <param name="ctx">Context which is used for the currently processed channel</param>
        /// <param name="e">Event being processed.</param>
        public void HandleUpstream(IChannelHandlerContext ctx, IChannelEvent e)
        {
            if (e is ConnectedEvent)
            {
                ctx.State = new Context();
            }
            if (e is ClosedEvent)
            {
                Reset();
            }
            else if (e is MessageEvent)
            {
                _context = ctx.State.As<Context>();
                _context._buffer = e.As<MessageEvent>().Message.As<BufferSlice>();
                _context._reader.Assign(_context._buffer);
                while (_context._parserMethod()) ;

                if (_context._isComplete)
                {
                    e.As<MessageEvent>().Message = _context._message;
                    OnComplete(ctx, e);
                }
                return;
            }

            ctx.SendUpstream(e);
        }

        public override void ChannelRead(IChannelHandlerContext context, object message)
        {
            var buffer = message as IByteBuffer;
            if (buffer != null)
            {
                int stringLength = buffer.ReadByte();
                var b = new byte[stringLength];
                buffer.ReadBytes(b, 0, b.Length);
                string instanceId = Encoding.UTF8.GetString(b);

                stringLength = buffer.ReadByte();
                b = new byte[stringLength];
                buffer.ReadBytes(b, 0, b.Length);
                string messageType = Encoding.UTF8.GetString(b);

                this.parent.SetInstanceIdChannel(instanceId, context.Channel);

                if (!this.clientConnectedInvoked)
                {
                    this.clientConnectedAction?.Invoke(instanceId, context.Channel.Id.AsShortText());
                    this.clientConnectedInvoked = true;
                }

                Task.Run(() =>
                {
                    this.dataReceivedAction(instanceId, context.Channel.Id.AsShortText(), messageType, buffer.ToArray());
                });
            }
        }

        //---------------------------------------------------------------------
        public async void ChannelActive(IChannelHandlerContext context)
        {
            this.context = context;
            listener.GatewaySession = this;

            await this.listener.OnSessionCreate();
        }

 public override void ChannelRead(IChannelHandlerContext context, object message)
 {
     var buffer = (IByteBuffer)message;
     var data = buffer.ToArray();
     var transportMessage = _serializer.Deserialize<byte[], TransportMessage>(data);
     context.FireChannelRead(transportMessage);
 }

        protected override void HandleMessage(IChannelHandlerContext ctx, MessageEvent e)
        {
            var msg = (Message) e.Message;
            string contentType = msg.Headers["Content-Type"];

            var reader = new StreamReader(msg.Body);
            string body = reader.ReadToEnd();
            msg.Body.Position = 0;

            switch (contentType)
            {
                case "auth/request":
                    SendCommand("auth", _password);
                    break;
                case "command/reply":
                    {
                        Command cmd;
                        if (!_commands.TryPop(out cmd))
                            throw new InvalidOperationException(
                                "Failed to find a command for the recieved command/reply");
                        OnCommand(cmd, msg);
                    }

                    break;
                case "text/event-plain":
                    ParseEvent(msg.Body);
                    break;
                default:
                    break;
            }
        }

 public override void ChannelInactive(IChannelHandlerContext context)
 {
     if (!WrappedTransport.ConnectionGroup.TryRemove(context.Channel))
     {
         Log.Warning("Unable to ADD channel [{0}->{1}](Id={2}) to connection group. May not shut down cleanly.",
             context.Channel.LocalAddress, context.Channel.RemoteAddress, context.Channel.Id);
     }
 }

        public override void ChannelActive(IChannelHandlerContext context)
        {
            log.Info($"Channel {context.Channel.Id.AsShortText()} connected");

            this.clientConnectedInvoked = false;

            base.ChannelActive(context);
        }

 public override void ChannelRead(IChannelHandlerContext context, object message)
 {
     Received.Add((IByteBuf) message);
     if (++_actualReadCount == _expectedReadCount)
     {
         _readFinished.Set();
     }
 }

 /// <summary>
 /// Handle the data that is going to be sent to the remote end point
 /// </summary>
 /// <param name="ctx">Context information</param>
 /// <param name="e">Chennel event.</param>
 public void HandleDownstream(IChannelHandlerContext ctx, IChannelEvent e)
 {
     if (e is CloseEvent && _timer != null)
     {
         _timer.Dispose();
         _timer = null;
     }
 }

 public override void ChannelRead(IChannelHandlerContext context, object message)
 {
     if (message is int)
     {
         State = State.ReceiveMessages((int) message);
         Logger.Debug("[Client-Read] Received: {0}", message);
     }
 }

 public override void ChannelRead(IChannelHandlerContext context, object message)
 {
     ReferenceCountUtil.Release(message);
     if (++this.actualReads == this.expectedReads)
     {
         this.signal.Signal();
     }
 }

 protected internal sealed override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output)
 {
     object decode = this.Decode(context, input);
     if (decode != null)
     {
         output.Add(decode);
     }
 }

        public PendingWriteQueue(IChannelHandlerContext ctx)
        {
            Contract.Requires(ctx != null);

            this.ctx = ctx;
            this.buffer = ctx.Channel.Unsafe.OutboundBuffer;
            this.estimatorHandle = ctx.Channel.Configuration.MessageSizeEstimator.NewHandle();
        }

 public override void ChannelRead(IChannelHandlerContext context, object message)
 {
     if (++_actualReads == _expectedReads)
     {
         _signal.Signal();
     }
     context.FireChannelRead(message);
 }

 public override Task WriteAsync(IChannelHandlerContext context, object message)
 {
     if (message is int)
     {
         State = State.WriteMessages((int) message);
     }
     return base.WriteAsync(context, message);
 }

 public override Task WriteAsync(IChannelHandlerContext context, object message)
 {
     if (message is int)
     {
         var buf = Unpooled.Buffer(4).WriteInt((int) message);
         return context.WriteAsync(buf);
     }
     return context.WriteAsync(message);
 }

        //---------------------------------------------------------------------
        public override void ChannelActive(IChannelHandlerContext context)
        {
            //context.WriteAndFlushAsync(this.initialMessage);

            var session = (ClientSession)this.factory.createRpcSession(null);
            mapSession[context] = session;

            session.ChannelActive(context);
        }

 public override Task WriteAsync(IChannelHandlerContext context, object message)
 {
     if (message is int)
     {
         State = State.WriteMessages((int) message);
         Logger.Debug("[Client-Write] Writing: {0}", message);
     }
     return context.WriteAsync(message);
 }