import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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);
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
/**
 * 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();
    }
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
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);
        }
    };
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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);
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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));
        }
    });
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
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;
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
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;
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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;
    }
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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");
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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.");
    }
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
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);
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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;
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
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);
        	}
        }
        
    }

}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
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);
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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());
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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);
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
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);
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@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);
    }
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
@Override
public void channelRead ( final ChannelHandlerContext ctx, final Object msg ) throws Exception
{
    if ( msg instanceof SyslogMessage )
    {
        handleMessage ( (SyslogMessage)msg );
    }
}
 

import io.netty.channel.ChannelHandlerContext; //导入依赖的package包/类
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;
}