TCP服务器

初始化channel

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast(ProxyProtocolChannelHandler.NAME, new ProxyProtocolChannelHandler()); //这个放在最前面
        ch.pipeline().addLast(
                new StreamMessageDecoder(),
                new StreamMessageEncoder(),
                handler.newIdleStateHandler(),
                new StreamServerHandler(handler));
    }

ProxyProtocolChannelHandler的实现

public class ProxyProtocolChannelHandler extends ChannelInboundHandlerAdapter {
    public static final String NAME = ProxyProtocolChannelHandler.class.getSimpleName();

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof ByteBuf) {
            try {
                ProtocolDetectionResult<HAProxyProtocolVersion> result = HAProxyMessageDecoder.detectProtocol((ByteBuf) msg);

                // TODO - is it possible that this message could be split over multiple ByteBufS, and therefore this would fail?
                if (result.state() == ProtocolDetectionState.DETECTED) {
                    // Add the actual HAProxy decoder.
                    // Note that the HAProxyMessageDecoder will remove itself once it has finished decoding the initial ProxyProtocol message(s).
                    ctx.pipeline().addAfter(NAME, null, new HAProxyMessageDecoder());

                    // Remove this handler, as now no longer needed.
                    ctx.pipeline().remove(this);
                }
            } catch (Exception e) {
                ((ByteBuf) msg).release();
            }
        }
        super.channelRead(ctx, msg);
    }
}

正常的消息handler

  @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(msg instanceof HAProxyMessage){
            HAProxyMessage message = (HAProxyMessage) msg;
            ctx.channel().attr(ChannelUtil.CLIENT_IP).set(InetAddress.getByName(message.sourceAddress()));
            message.release();
        }else {

        }
    }

Websocket 服务器

初始化 channel

   @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("httpServerCodec", new HttpServerCodec());
        pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpHeadersHandler()); //放在 WebSocketServerProtocolHandler之前
        pipeline.addLast(new WebSocketServerProtocolHandler("/"));
        pipeline.addLast("httpHandler", new WebsocketServerHandler(server));
    }

HttpHeadersHandler 实现

public class HttpHeadersHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            HttpHeaders headers = ((FullHttpRequest) msg).headers();
            String realip = headers.get("X-Real-IP");
            if (realip != null && !realip.isEmpty()) {
                ctx.channel().attr(ChannelUtil.CLIENT_IP).set(InetAddress.getByName(realip));
            } else {
                String forwarded_for = headers.get("X-Forwarded-For");
                if (forwarded_for != null && !forwarded_for.isEmpty()) {
                    ctx.channel().attr(ChannelUtil.CLIENT_IP).set(InetAddress.getByName(forwarded_for.split(",")[0].trim()));
                }
            }
            ctx.channel().pipeline().remove(this);
        }
        ctx.fireChannelRead(msg);
    }
}
这里是保存到channel属性里