Java 类io.netty.channel.ChannelHandlerContext 实例源码

@Override
protected void encode(ChannelHandlerContext ctx, ConnectionResponse msg, ByteBuf out) throws Exception {
    ChannelPipeline pipeline = ctx.pipeline();

    switch (msg.getType()) {
    case HANDSHAKE_CONNECTION:
        pipeline.addAfter("decoder", "handshake.encoder", new HandshakeEncoder());
        pipeline.replace("decoder", "handshake.decoder", new HandshakeDecoder());
        break;
    case LOGIN_CONNECTION:
        out.writeByte(ClientMessage.SUCCESSFUL_CONNECTION.getId());
        pipeline.addAfter("decoder", "login.encoder", new LoginEncoder());
        pipeline.replace("decoder", "login.decoder", new LoginDecoder());
        break;
    }

    pipeline.remove(this);
}

/**
 * Synchronized by {@code synchronized (this)} in {@link #channelRead0}
 */
private void handlePacket(ChannelHandlerContext ctx, byte[] data) {
    try {
        temp = Utils.arrayAppend(temp, data);
        while (temp.length != 0) {
            int position = Utils.arraySearch(temp, Protocol.SIGNATURE);
            if (position < 0) {
                return;//收到的是子包, 数据未结尾
            }

            byte[] d = Utils.arrayGetCenter(temp, 0, position);
            temp = Utils.arrayDelete(temp, position + Protocol.SIGNATURE.length);

         JPREMain.getInstance().getScheduler().addTask(() -> processPacket(ctx, d));
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
}

private Runnable createFlushOperation(ChannelHandlerContext ctx, long cmdHandle) {
/* todo  we must drain all NBD_CMD_WRITE and NBD_WRITE_TRIM from the queue
 * before processing NBD_CMD_FLUSH
 */
    return () -> {
        int err = 0;
        try {
            unflushedBytes.set(0);
            exportProvider.flush();
        } catch (Exception e) {
            LOGGER.error("error during flush", e);
            err = Protocol.EIO_ERROR;
        } finally {
            sendTransmissionSimpleReply(ctx, err, cmdHandle, null);
        }
    };
}

@Override
protected void decode(ChannelHandlerContext ctx, WebSocketFrame frame, List<Object> out) throws Exception {
    ByteBuf buf = frame.content().order(ByteOrder.LITTLE_ENDIAN);
    if (buf.capacity() < 1) {
        // Discard empty messages
        return;
    }

    buf.resetReaderIndex();
    int packetId = buf.readUnsignedByte();
    Packet packet = reg.SERVERBOUND.constructPacket(packetId);

    if (packet == null) {
        throw new UnknownPacketException("Unknown packet ID: " + packetId);
    }

    Server.log.finest("Received packet ID " + packetId + " (" + packet.getClass().getSimpleName() + ") from " + ctx.channel().remoteAddress());

    packet.readData(buf);
    out.add(packet);
}

@Override
public void handle(ChannelHandlerContext ctx, AmqpConnectionHandler connectionHandler) {
    // TODO handle exclusive param
    AmqpChannel channel = connectionHandler.getChannel(getChannel());

    ctx.fireChannelRead((BlockingTask) () -> {
        try {
            channel.declareQueue(queue, passive, durable, autoDelete);
            ctx.writeAndFlush(new QueueDeclareOk(getChannel(), queue, 0, 0));
        } catch (BrokerException e) {
            LOGGER.warn("Error declaring queue.", e);
            ctx.writeAndFlush(new ChannelClose(getChannel(),
                                               ChannelException.NOT_ALLOWED,
                                               ShortString.parseString(e.getMessage()),
                                               CLASS_ID,
                                               METHOD_ID));
        }
    });
}

public static RemotingServer createRemotingServer() throws InterruptedException {
    NettyServerConfig config = new NettyServerConfig();
    RemotingServer remotingServer = new NettyRemotingServer(config);
    remotingServer.registerProcessor(0, new NettyRequestProcessor() {
        @Override
        public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) {
            request.setRemark("Hi " + ctx.channel().remoteAddress());
            return request;
        }

        @Override
        public boolean rejectRequest() {
            return false;
        }
    }, Executors.newCachedThreadPool());

    remotingServer.start();

    return remotingServer;
}

private Message decodeMessage(ChannelHandlerContext ctx, List<Frame> frames) throws IOException {
    long frameType = frames.get(0).getType();

    byte[] payload = new byte[frames.size() == 1 ? frames.get(0).getSize() : frames.get(0).totalFrameSize];
    int pos = 0;
    for (Frame frame : frames) {
        pos += ByteStreams.read(frame.getStream(), payload, pos, frame.getSize());
    }

    if (loggerWire.isDebugEnabled()) {
        loggerWire.debug("Recv: Encoded: {} [{}]", frameType, Hex.toHexString(payload));
    }

    Message msg = createMessage((byte) frameType, payload);

    if (loggerNet.isInfoEnabled()) {
        loggerNet.info("From: \t{} \tRecv: \t{}", channel, msg.toString());
    }

    ethereumListener.onRecvMessage(channel, msg);

    channel.getNodeStatistics().rlpxInMessages.add();
    return msg;
}

@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    SendMessageContext mqtraceContext;
    switch (request.getCode()) {
        case RequestCode.CONSUMER_SEND_MSG_BACK:
            return this.consumerSendMsgBack(ctx, request);
        default:
            SendMessageRequestHeader requestHeader = parseRequestHeader(request);
            if (requestHeader == null) {
                return null;
            }

            //默认情况下，mqtraceContext = null;
            mqtraceContext = buildMsgContext(ctx, requestHeader);
            this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
            final RemotingCommand response = this.sendMessage(ctx, request, mqtraceContext, requestHeader);

            this.executeSendMessageHookAfter(response, mqtraceContext);
            return response;
    }
}

@Test
@Tag("fast")
public void decodeConnectionRequest() throws Exception {
    ChannelHandlerContext ctx = new MockChannelHandlerContext();
    ByteBuf buf = Unpooled.buffer();
    ArrayList<Object> out = new ArrayList<>();

    buf.writeByte(0x6); // header length
    buf.writeByte(TpduCode.CONNECTION_REQUEST.getCode());
    buf.writeShort(0x01); // destination reference
    buf.writeShort(0x02); // source reference
    buf.writeByte(ProtocolClass.CLASS_0.getCode());
    IsoOnTcpMessage in = new IsoOnTcpMessage(buf);

    isoTPProtocol.decode(ctx, in, out);

    assertTrue(out.size() == 1, "Message not decoded");

    ConnectionRequestTpdu requestTpdu = (ConnectionRequestTpdu) ((IsoTPMessage)out.get(0)).getTpdu();

    assertTrue(requestTpdu.getTpduCode() == TpduCode.CONNECTION_REQUEST, "Message code not correct");
    assertTrue(requestTpdu.getDestinationReference() == (short) 0x1, "Message destination reference not correct");
    assertTrue(requestTpdu.getSourceReference() == (short) 0x2, "Message source reference not correct");
    assertTrue(requestTpdu.getProtocolClass() == ProtocolClass.CLASS_0, "Message protocol class reference not correct");
    assertTrue(requestTpdu.getParameters().isEmpty(), "Message contains paramaters");
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

    // read idle event.
    if (evt == IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT
            || evt == IdleStateEvent.READER_IDLE_STATE_EVENT) {

        if (null != operation) {
            throw new FastdfsReadTimeoutException(
                    String.format(
                            "execute %s read timeout.",
                            operation
                    )
            );
        }

        return;
    }

    // all idle event.
    if (evt == IdleStateEvent.FIRST_ALL_IDLE_STATE_EVENT
            || evt == IdleStateEvent.ALL_IDLE_STATE_EVENT) {
        throw new FastdfsTimeoutException("fastdfs channel was idle timeout.");
    }
}

public void write(ChannelHandlerContext context, Object packet, ChannelPromise promise) throws Exception
{
    BaseComponent[] components = interceptor.getComponents(packet);
    if(components != null)
    {
        boolean allowed = isAllowed(components);
        boolean paused = isPaused();
        if(!paused || !allowed)
        {
            while(messageQueue.size() > 20)
                messageQueue.remove();
            messageQueue.add(components);
        }

        if(paused && !allowed)
            return;
    }

    super.write(context, packet, promise);
}

@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
    switch (request.getCode()) {
    case RequestCode.HEART_BEAT:
        return this.heartBeat(ctx, request);
    case RequestCode.UNREGISTER_CLIENT:
        return this.unregisterClient(ctx, request);
    case RequestCode.GET_CONSUMER_LIST_BY_GROUP:
        return this.getConsumerListByGroup(ctx, request);
    case RequestCode.UPDATE_CONSUMER_OFFSET:
        return this.updateConsumerOffset(ctx, request);
    case RequestCode.QUERY_CONSUMER_OFFSET:
        return this.queryConsumerOffset(ctx, request);
    default:
        break;  
    }
    return null;
}

protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception {
    if(clazz.isInstance(o)){
        Kryo kryo = null;
        try{
            kryo = pool.borrow();
            ByteArrayOutputStream baos = new ByteArrayOutputStream();  
            Output output = new Output(baos);  
            kryo.writeObject(output, o);  
            output.flush();  
            output.close();  

            byte[] data = baos.toByteArray(); 
            byteBuf.writeInt(data.length);
            byteBuf.writeBytes(data);
            baos.close();
        }catch(Exception e){
            LOG.warn("MessageEncoder happen exception.", e);
        }finally{
            if(kryo != null){
                 pool.release(kryo);
            }
        }

    }

}

protected void onPubComp(ChannelHandlerContext ctx, MqttMessage msg) {
    if (!this.connected) {
        logger.debug("Protocol violation: Client {} must first sent a CONNECT message, now received PUBCOMP message, disconnect the client", this.clientId);
        ctx.close();
        return;
    }

    logger.debug("Message received: Received PUBCOMP message from client {} user {}", this.clientId, this.userName);

    MqttPacketIdVariableHeader variable = (MqttPacketIdVariableHeader) msg.variableHeader();
    int packetId = variable.packetId();

    // In the QoS 2 delivery protocol, the Sender
    // MUST treat the PUBREL packet as “unacknowledged” until it has received the corresponding
    // PUBCOMP packet from the receiver.
    logger.trace("Remove in-flight: Remove in-flight PUBREL message {} for client {}", packetId, this.clientId);
    this.redis.removeInFlightMessage(this.clientId, packetId);
}

@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    LOGGER.info("{} : handlerAdded", connectionInfo);

    Http2Connection connection = new DefaultHttp2Connection(false);
    ChannelHandler http2ConnHandler = new HttpToHttp2ConnectionHandlerBuilder()
            .frameListener(new DelegatingDecompressorFrameListener(
                    connection,
                    new InboundHttp2ToHttpAdapterBuilder(connection)
                            .maxContentLength(master.config().getMaxContentLength())
                            .propagateSettings(true)
                            .build()))
            .frameLogger(new Http2FrameLogger(LogLevel.DEBUG))
            .connection(connection)
            .build();
    ctx.pipeline()
       .addBefore(ctx.name(), null, http2ConnHandler)
       .addBefore(ctx.name(), null, new Http2Handler());
}

@Override
public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
    Channel channel = ctx.channel();
    logger.info("Disconnected from {}", NetUtils.channelToString(channel.remoteAddress(), channel.localAddress()));
    BaseServerHandler.removeChannel(channel);
    if (connectListeners != null) {
        serverHandler.getBizThreadPool().execute(new Runnable() {
            @Override
            public void run() {
                for (ConnectListener connectListener : connectListeners) {
                    try {
                        connectListener.disconnected(ctx);
                    } catch (Exception e) {
                        logger.warn("Failed to call connect listener when channel inactive", e);
                    }
                }
            }
        });
    }
    CallbackUtil.removeTransport(channel);
}

public RemotingCommand resetOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final ResetOffsetRequestHeader requestHeader =
        (ResetOffsetRequestHeader) request.decodeCommandCustomHeader(ResetOffsetRequestHeader.class);
    log.info("[reset-offset] reset offset started by {}. topic={}, group={}, timestamp={}, isForce={}",
        RemotingHelper.parseChannelRemoteAddr(ctx.channel()), requestHeader.getTopic(), requestHeader.getGroup(),
        requestHeader.getTimestamp(), requestHeader.isForce());
    boolean isC = false;
    LanguageCode language = request.getLanguage();
    switch (language) {
        case CPP:
            isC = true;
            break;
    }
    return this.brokerController.getBroker2Client().resetOffset(requestHeader.getTopic(), requestHeader.getGroup(),
        requestHeader.getTimestamp(), requestHeader.isForce(), isC);
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    // If the connection has just been established, start setting up the connection
    // by sending a connection request to the plc.
    if(evt instanceof S7ConnectionEvent && ((S7ConnectionEvent) evt).getState() == S7ConnectionState.INITIAL) {
        logger.debug("ISO Transport Protocol Sending Connection Request");
        // Open the session on ISO Transport Protocol first.
        ConnectionRequestTpdu connectionRequest = new ConnectionRequestTpdu(
            (short) 0x0000, (short) 0x000F, ProtocolClass.CLASS_0,
            Arrays.asList(
                new CalledTsapParameter(DeviceGroup.PG_OR_PC, (byte) 0, (byte) 0),
                new CallingTsapParameter(DeviceGroup.OTHERS, rackNo, slotNo),
                new TpduSizeParameter(tpduSize)),
            Unpooled.buffer());
        ctx.channel().writeAndFlush(connectionRequest);
    } else {
        super.userEventTriggered(ctx, evt);
    }
}

@Override
public void channelRead ( final ChannelHandlerContext ctx, final Object msg ) throws Exception
{
    if ( msg instanceof SyslogMessage )
    {
        handleMessage ( (SyslogMessage)msg );
    }
}

private RemotingCommand unlockBatchMQ(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    UnlockBatchRequestBody requestBody = UnlockBatchRequestBody.decode(request.getBody(), UnlockBatchRequestBody.class);

    this.brokerController.getRebalanceLockManager().unlockBatch(//
        requestBody.getConsumerGroup(),//
        requestBody.getMqSet(),//
        requestBody.getClientId());

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}

protected void channelRead0(ChannelHandlerContext p_channelRead0_1_, Packet p_channelRead0_2_) throws Exception
{
    if (this.channel.isOpen())
    {
        try
        {
            p_channelRead0_2_.processPacket(this.packetListener);
        }
        catch (ThreadQuickExitException var4)
        {
            ;
        }
    }
}

protected void channelRead0(ChannelHandlerContext p_channelRead0_1_, Packet<?> p_channelRead0_2_) throws Exception
{
    if (this.channel.isOpen())
    {
        try
        {
            ((Packet<INetHandler>)p_channelRead0_2_).processPacket(this.packetListener);
        }
        catch (ThreadQuickExitException var4)
        {
            ;
        }
    }
}

private RemotingCommand deleteTopicInNamesrv(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final DeleteTopicInNamesrvRequestHeader requestHeader =
        (DeleteTopicInNamesrvRequestHeader) request.decodeCommandCustomHeader(DeleteTopicInNamesrvRequestHeader.class);

    this.namesrvController.getRouteInfoManager().deleteTopic(requestHeader.getTopic());

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}

public RemotingCommand registerBrokerWithFilterServer(ChannelHandlerContext ctx, RemotingCommand request)
    throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
    final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
    final RegisterBrokerRequestHeader requestHeader =
        (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);

    RegisterBrokerBody registerBrokerBody = new RegisterBrokerBody();

    if (request.getBody() != null) {
        registerBrokerBody = RegisterBrokerBody.decode(request.getBody(), RegisterBrokerBody.class);
    } else {
        registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setCounter(new AtomicLong(0));
        registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setTimestamp(0);
    }

    RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
        requestHeader.getClusterName(),
        requestHeader.getBrokerAddr(),
        requestHeader.getBrokerName(),
        requestHeader.getBrokerId(),
        requestHeader.getHaServerAddr(),
        registerBrokerBody.getTopicConfigSerializeWrapper(),
        registerBrokerBody.getFilterServerList(),
        ctx.channel());

    responseHeader.setHaServerAddr(result.getHaServerAddr());
    responseHeader.setMasterAddr(result.getMasterAddr());

    byte[] jsonValue = this.namesrvController.getKvConfigManager().getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
    response.setBody(jsonValue);

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}

@Override
    public void onRequest(ChannelHandlerContext ctx, long userId, SocketASK ask) {
        switch (ask.getForward()) {
            case 0:
                break;

        }

//        CompletableFuture.supplyAsync(() -> {
//
//        });

    }

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
    log.info("NETTY SERVER PIPELINE: channelInactive, the channel[{}]", remoteAddress);
    super.channelInactive(ctx);

    if (NettyRemotingServer.this.channelEventListener != null) {
        NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress, ctx.channel()));
    }
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    super.channelInactive(ctx);
    ProcessorTable.getAll().forEach((key,instance)->{
        instance.doDestory(application);
    });
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    System.out.println("[Network] RemoteClient: " + ctx.channel().remoteAddress() + " disconnected.");

    for (MPQClient client : clients) {
        if (client.is((InetSocketAddress) ctx.channel().remoteAddress())) {
            FrameDisconnectionEvent event = new FrameDisconnectionEvent(client.getFrame());
            client.getFrame().getPluginManager().callEvent(event);
            break;
        }
    }

    super.channelInactive(ctx);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  if (expectingNoopResponse && msg instanceof SmtpResponse) {
    LOG.debug("[{}] NOOP response received {}", connectionId, SmtpResponses.toString((SmtpResponse) msg));
    swallowNoopResponse((SmtpResponse) msg);
    sendPendingWrites(ctx);
    return;
  }

  super.channelRead(ctx, msg);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
  // queue any pending writes until the response is received
  if (expectingNoopResponse) {
    pendingWrites.add(new PendingWrite(msg, promise));
    return;
  }

  super.write(ctx, msg, promise);
}

public void broadcastToMobile(IMProtoMessage<MessageLite> message, ChannelHandlerContext fromCtx) {
    for (ChannelHandlerContext conn: connMap.values()) {
        if (conn != fromCtx && CommonUtils.isMobile(conn.attr(CLIENT_TYPE).get())) {
            logger.debug("发送消息> {}", conn.channel().remoteAddress());
            conn.writeAndFlush(message);
        }
    }
}

private void initHandshake(ChannelHandlerContext ctx) throws IOException {
    /* initiate handshake */
    ByteBuf bb = ctx.alloc().buffer(20);
    bb.writeLong(Protocol.NBDMAGIC);
    bb.writeLong(Protocol.IHAVEOPT);

    // "handshake flags"
    handshakeFlags = Protocol.NBD_FLAG_FIXED_NEWSTYLE & Protocol.NBD_FLAG_NO_ZEROES;
    bb.writeShort(handshakeFlags);
    ctx.channel().writeAndFlush(bb);
}

private RemotingCommand updateBrokerConfig(ChannelHandlerContext ctx, RemotingCommand request) {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);

    log.info("updateBrokerConfig called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

    byte[] body = request.getBody();
    if (body != null) {
        try {
            String bodyStr = new String(body, MixAll.DEFAULT_CHARSET);
            Properties properties = MixAll.string2Properties(bodyStr);
            if (properties != null) {
                log.info("updateBrokerConfig, new config: [{}] client: {} ", properties, ctx.channel().remoteAddress());
                this.brokerController.getConfiguration().update(properties);
                if (properties.containsKey("brokerPermission")) {
                    this.brokerController.registerBrokerAll(false, false);
                    this.brokerController.getTopicConfigManager().getDataVersion().nextVersion();
                }
            } else {
                log.error("string2Properties error");
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("string2Properties error");
                return response;
            }
        } catch (UnsupportedEncodingException e) {
            log.error("", e);
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("UnsupportedEncodingException " + e);
            return response;
        }
    }

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}

@Override
public void channelRead0(ChannelHandlerContext ctx, AbstractPacket packet) throws Exception {
    if(handle == null) return;
    try {
        this.handle.getNetworkBus().processIn(ctx.channel(), packet);
    }
    finally {
        //packet.trySingleRelease();
    }
}

public UserConn getUserConn() {
    int count = 0;
    for (ChannelHandlerContext conn: connMap.values()) {
        if (conn.channel().isOpen()) {
            count++;
        }
    }
    return new UserConn(this.userId, count);
}

@Override
protected void encode( ChannelHandlerContext channelHandlerContext, ByteBuf a, ByteBuf byteBuf ) throws Exception {
    int i = a.readableBytes();
    int j = Message.getVarIntSize( i );
    if ( j > 3 ) {
        throw new IllegalArgumentException( "unable to fit " + i + " into " + 3 );
    }

    byteBuf.ensureWritable( j + i );
    Message.writeVarInt( i, byteBuf );
    byteBuf.writeBytes( a, a.readerIndex(), i );
}

/** Triggered based on events from an {@link io.netty.handler.timeout.IdleStateHandler}. */
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  if (evt instanceof IdleStateEvent) {
    IdleStateEvent e = (IdleStateEvent) evt;
    // See class comment for timeout semantics. In addition to ensuring we only timeout while
    // there are outstanding requests, we also do a secondary consistency check to ensure
    // there's no race between the idle timeout and incrementing the numOutstandingRequests
    // (see SPARK-7003).
    //
    // To avoid a race between TransportClientFactory.createClient() and this code which could
    // result in an inactive client being returned, this needs to run in a synchronized block.
    synchronized (this) {
      boolean isActuallyOverdue =
        System.nanoTime() - responseHandler.getTimeOfLastRequestNs() > requestTimeoutNs;
      if (e.state() == IdleState.ALL_IDLE && isActuallyOverdue) {
        if (responseHandler.numOutstandingRequests() > 0) {
          String address = getRemoteAddress(ctx.channel());
          logger.error("Connection to {} has been quiet for {} ms while there are outstanding " +
            "requests. Assuming connection is dead; please adjust spark.network.timeout if " +
            "this is wrong.", address, requestTimeoutNs / 1000 / 1000);
          client.timeOut();
          ctx.close();
        } else if (closeIdleConnections) {
          // While CloseIdleConnections is enable, we also close idle connection
          client.timeOut();
          ctx.close();
        }
      }
    }
  }
  ctx.fireUserEventTriggered(evt);
}

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    if (!in.isReadable() || in.readableBytes() < 4) {
        return;
    }

    Optional<UpdateType> request = UpdateType.getType(in.readUnsignedByte());
    if (request.isPresent()) {
        UpdateType updateType = request.get();
        switch (updateType) {
        case LOW_PRIORITY_UPDATE:
        case HIGH_PRIORITY_UPDATE:
            int uid = in.readUnsignedMedium();
            int type = (uid >> 16);
            int id = (uid & 0xffff);

            out.add(new UpdateRequest(this, type, id, updateType == UpdateType.HIGH_PRIORITY_UPDATE));
            break;
        case XOR_ENCRYPTION_UPDATE:
            int key = in.readUnsignedByte();
            in.readUnsignedShort();
            out.add(new XOREncryptionRequest(this, key));
            break;
        }
    } else {
        in.readUnsignedMedium();
    }
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    if (this.userid != null) {
        if (this.topics != null) {
            this.dispatcher.deregister(this.topics, this);
        }
        state.exit(userid);
        System.out.println(this.userid + " disconnected");
    }
}

AbstractSession(SessionManager sessionManager, InetSocketAddress address, ChannelHandlerContext ctx) {
    this.sessionManager = sessionManager;
    this.address = address;
    this.ctx = ctx;
    tickTask = ctx.executor().scheduleAtFixedRate(this::update0, 10, 10, TimeUnit.MILLISECONDS);

    for (int i = 0; i < 32; i++) {
        orderChannels.put(i, new OrderChannel(this));
    }
}