java类io.netty.channel.ChannelHandler的实例源码

/**
 * Add a custom channel handler to the given channel.
 * 
 * Method modified by @Edasakii to fix bug on reload caused by CastException
 * 
 * @param player - the channel to inject.
 * @return The packet interceptor.
 */
private PacketInterceptor injectChannelInternal(Channel channel) {
    try {
        ChannelHandler ch = channel.pipeline().get(handlerName);
        PacketInterceptor interceptor = null;
        if (ch != null && ch instanceof PacketInterceptor) {
            interceptor = (PacketInterceptor) ch;
        } else {
            interceptor = new PacketInterceptor();
            channel.pipeline().addBefore("packet_handler", handlerName, interceptor);
            uninjectedChannels.remove(channel);
        }
        return interceptor;
    } catch (IllegalArgumentException e) {
        e.printStackTrace();
        return null;
    }
}

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

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

@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
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline pipeline = socketChannel.pipeline();

    ChannelHandler lengthFieldBasedFrameDecoder = NettyUtils.createLengthFieldBasedFrameDecoder(32768, 0, 4);
    pipeline.addLast(lengthFieldBasedFrameDecoder);

    ChannelHandler packetDecoder = new SkyllaPacketDecoder(this.protocol);
    pipeline.addLast(packetDecoder);

    ChannelHandler lengthFieldPrepender = NettyUtils.createLengthFieldPrepender(4);
    pipeline.addLast(lengthFieldPrepender);

    ChannelHandler packetEncoder = new SkyllaPacketEncoder(this.protocol);
    pipeline.addLast(packetEncoder);

    ChannelHandler packetHandler = new SkyllaConnection(socketChannel, this.protocol);
    pipeline.addLast(packetHandler);
}

@Override
public void connect() {
    this.workerGroup = NettyUtils.createEventLoopGroup(4);

    Class<? extends Channel> channelClazz = NettyUtils.getChannel();
    ChannelHandler channelInitializer = new SkyllaChannelInitializer(this.config.getProtocol());

    Bootstrap bootstrap = new Bootstrap();

    try {
        channel = bootstrap
                .channel(channelClazz)
                .group(this.workerGroup)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_BACKLOG, 50)
                .handler(channelInitializer)
                .connect(this.config.getServerHost(), this.config.getServerPort())
                .sync().channel();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

/**
 * Adds a channel that listens locally
 */
@SideOnly(Side.CLIENT)
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)SERVER_NIO_EVENTLOOP.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

/**
     * 适配
     */
    @Override
    protected ChannelHandler fixHandlerBeforeConnect(final ChannelHandler handler) {
        ChannelHandler result=new ShareableChannelInboundHandler() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Channel ch=ctx.channel();
                ch.pipeline().addLast(new HttpClientCodec());
                ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
                ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
                ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
                ctx.pipeline().remove(this);//移除当前handler
                ctx.pipeline().fireChannelRegistered();//重新从第一个handler抛出事件
            }
        };
//      ChannelInitializer<SocketChannel> result=new ChannelInitializer<SocketChannel>() {
//            @Override
//            protected void initChannel(SocketChannel ch) {
//              ch.pipeline().addLast(new HttpClientCodec());
//              ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
//              ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
//              ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
//            }
//        };
        return result;
    }

/**
     * 适配
     */
    @Override
    protected ChannelHandler fixHandlerBeforeConnect(final ChannelHandler handler) {
        ChannelHandler result=new ShareableChannelInboundHandler() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Channel ch=ctx.channel();
                ch.pipeline().addLast(new HttpClientCodec());
                ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
                ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
                ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
                ctx.pipeline().remove(this);//移除当前handler
                ctx.fireChannelRegistered();//重新从第一个handler抛出事件
            }
        };
//      ChannelInitializer<SocketChannel> result=new ChannelInitializer<SocketChannel>() {
//            @Override
//            protected void initChannel(SocketChannel ch) {
//              ch.pipeline().addLast(new HttpClientCodec());
//              ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
//              ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
//              ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
//            }
//        };
        return result;
    }

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

protected ChannelHandler setupWSChannel(SslContext sslCtx, Configuration conf, DataStore datastore) {
    return new ChannelInitializer<SocketChannel>() {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast("ssl", sslCtx.newHandler(ch.alloc()));
            ch.pipeline().addLast("httpServer", new HttpServerCodec());
            ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
            ch.pipeline().addLast("sessionExtractor", new WebSocketHttpCookieHandler(config));
            ch.pipeline().addLast("idle-handler", new IdleStateHandler(conf.getWebsocket().getTimeout(), 0, 0));
            ch.pipeline().addLast("ws-protocol", new WebSocketServerProtocolHandler(WS_PATH, null, true));
            ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(datastore, config));
            ch.pipeline().addLast("error", new WSExceptionHandler());
        }
    };

}

@Override
protected ChannelHandler newNonSslHandler(ChannelHandlerContext context) {
    return new ChannelInboundHandlerAdapter() {

        private HttpResponseEncoder encoder = new HttpResponseEncoder();

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            LOG.trace("Received non-SSL request, returning redirect");
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                    HttpResponseStatus.MOVED_PERMANENTLY, Unpooled.EMPTY_BUFFER);
            response.headers().set(Names.LOCATION, redirectAddress);
            LOG.trace(Constants.LOG_RETURNING_RESPONSE, response);
            encoder.write(ctx, response, ctx.voidPromise());
            ctx.flush();
        }
    };
}

private SslHandler getSslHandler() throws Exception {
  // get SslHandler if it was added to the pipeline
  ArgumentCaptor<ChannelHandler> captor = ArgumentCaptor.forClass(ChannelHandler.class);
  verify(pipeline).addFirst(captor.capture());
  SslHandler sslHandler = (SslHandler) captor.getValue();

  // mock and store the context so we can get the handshake future
  ChannelHandlerContext context = mock(ChannelHandlerContext.class);
  when(context.executor()).thenReturn(ImmediateEventExecutor.INSTANCE);
  when(context.channel()).thenReturn(mock(Channel.class, Answers.RETURNS_MOCKS.get()));

  // add the handler but prevent the handshake from running automatically
  when(channel.isActive()).thenReturn(false);
  sslHandler.handlerAdded(context);

  return sslHandler;
}

/**
 * INTERNAL Create a new channel pair with the specified name and channel handlers.
 * This is used internally in forge and FML
 *
 * @param container The container to associate the channel with
 * @param name The name for the channel
 * @param handlers Some {@link ChannelHandler} for the channel
 * @return an {@link EnumMap} of the pair of channels. keys are {@link Side}. There will always be two entries.
 */
public EnumMap<Side,FMLEmbeddedChannel> newChannel(ModContainer container, String name, ChannelHandler... handlers)
{
    if (channels.get(Side.CLIENT).containsKey(name) || channels.get(Side.SERVER).containsKey(name) || name.startsWith("MC|") || name.startsWith("\u0001") || (name.startsWith("FML") && !("FML".equals(container.getModId()))))
    {
        throw new RuntimeException("That channel is already registered");
    }
    EnumMap<Side,FMLEmbeddedChannel> result = Maps.newEnumMap(Side.class);

    for (Side side : Side.values())
    {
        FMLEmbeddedChannel channel = new FMLEmbeddedChannel(container, name, side, handlers);
        channels.get(side).put(name,channel);
        result.put(side, channel);
    }
    return result;
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)SERVER_NIO_EVENTLOOP.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

public static ChannelHandler[] getPreInterceptors(){
    preLock.lock();
    try {
        if(preInterceptors==null){
            preInterceptors = getInterceptors(PreHandleInterceptor.class);
        }
    }finally {
        preLock.unlock();
    }
    return ArrayUtil.clone(preInterceptors);
}

public static ChannelHandler[] getAfterInterceptors(){
    afterLock.lock();
    try {
        if(afterInterceptors==null){
            afterInterceptors = getInterceptors(AfterHandleInterceptor.class);
        }
    }finally {
        afterLock.unlock();
    }
    return ArrayUtil.clone(afterInterceptors);
}

public ChannelFuture connect(ChannelHandlerContext fromCtx, NitmProxyMaster master, ConnectionInfo connectionInfo,
                             ChannelHandler handler) {
    return new Bootstrap()
            .group(fromCtx.channel().eventLoop())
            .channel(fromCtx.channel().getClass())
            .handler(handler)
            .connect(connectionInfo.getServerAddr().getHost(),
                     connectionInfo.getServerAddr().getPort());
}

public String findChannelHandlerNameForType(Class<? extends ChannelHandler> type)
{
    String targetName = null;
    for (Entry<String, ChannelHandler> entry : pipeline())
    {
        if (type.isInstance(entry.getValue()))
        {
            targetName = entry.getKey();
            break;
        }
    }
    return targetName;
}

public ChannelHandler proxyHandler(Address clientAddress) {
    switch (config.getProxyMode()) {
    case HTTP:
        return new HttpProxyHandler(this, new ConnectionInfo(clientAddress));
    case SOCKS:
        return new SocksProxyHandler(this, new ConnectionInfo(clientAddress));
    default:
        throw new IllegalStateException("No proxy mode available: " + config.getProxyMode());
    }
}

private Http1FrontendHandler httpProxyHandler(boolean outboundAvailable) {
    if (outboundAvailable) {
        when(master.connect(any(), any(), any())).then(
                invocationOnMock ->  {
                    outboundChannel = new EmbeddedChannel((ChannelHandler) invocationOnMock.getArguments()[2]);
                    return outboundChannel.newSucceededFuture();
                });
    } else {
        when(master.connect(any(), any(), any())).then(
                invocationOnMock ->  inboundChannel.newPromise().setFailure(new Exception()));
    }
    return new Http1FrontendHandler(master, connectionInfo());
}

public FMLEmbeddedChannel(ModContainer container, String channelName, Side source, ChannelHandler... handlers)
{
    super(handlers);
    this.attr(NetworkRegistry.FML_CHANNEL).set(channelName);
    this.attr(NetworkRegistry.CHANNEL_SOURCE).set(source);
    this.attr(NetworkRegistry.MOD_CONTAINER).setIfAbsent(container);
    this.pipeline().addFirst("fml:outbound",new FMLOutboundHandler());
}

/**
 * Prepares a clientside NetworkManager: establishes a connection to the socket supplied and configures the channel
 * pipeline. Returns the newly created instance.
 */
public static NetworkManager provideLocalClient(SocketAddress address)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)CLIENT_LOCAL_EVENTLOOP.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(LocalChannel.class)).connect(address).syncUninterruptibly();
    return networkmanager;
}

public NettyClient(NettyClientConfig config,InetSocketAddress target,ChannelHandler handler) {
    super(target);
    if(config.getIoWorkers().isShutdown())
    {
        throw new UnsupportedOperationException("config is unActive");
    }
    this.config=config;
    this.handler=handler;
}

/**
 * 包装一个初始化父类channel的handler
 * @param handler 业务handler
 * @return
 */
private ChannelHandler channelInitFix(final ChannelHandler handler)
{
    ChannelHandler fixedHandler=new ShareableChannelInboundHandler() {
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            Channel ch=ctx.channel();
            setChannel(ch);
            ctx.pipeline().addLast(handler);
            ctx.pipeline().remove(this);//移除当前handler
            ctx.fireChannelRegistered();//从当前handler往后抛出事件
        }
    };
    return fixedHandler;
}

private String generateName(ChannelPipeline pipeline, ChannelHandler handler)
{
    try
    {
        return (String)generateName.invoke(defaultChannelPipeline.cast(pipeline), handler);
    }
    catch (Exception e)
    {
        FMLLog.log(Level.FATAL, e, "It appears we somehow have a not-standard pipeline. Huh");
        throw Throwables.propagate(e);
    }
}

/**
 * 因为每次连接执行都会init都会被remove,所以每次调用booter都会用新的handler来进行连接配置
 * @param address
 * @param init
 * @return
 */
protected ChannelFuture doBooterConnect(InetSocketAddress address,final ChannelHandler init)
{
    ChannelFuture cf;
    synchronized (booter) {
        if(booter.config().group()==null)
        {
            booterInit();
        }
        final CountDownLatch latch=new CountDownLatch(1);
        ChannelHandler handler=initHandlerAdapter(init);
        booter.handler(handler);
        cf=booter.connect(address);
        cf.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                log.trace("connect operationComplete:isDone="+future.isDone()+",isSuccess="+future.isSuccess());
                if(future.isDone() && future.isSuccess())
                {
                    latch.countDown();
                }
            }
        });
        try {
            latch.await(getConnectTimeOutMills(),TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
    }
    return cf;
}

/**
     * 初始化handler适配包装
     * @param init
     * @return
     */
    protected ChannelHandler initLogHandlerAdapter(ChannelHandler init)
    {
        ChannelHandler handler=new  ShareableChannelInboundHandler() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Channel ch=ctx.channel();
                manageChannel(ch);
                LogLevel level=config.getChannelLevel();
                if(level!=null)
                {//单个链路的日志记录器
                    ch.pipeline().addLast(new LoggerHandler(level));
                }
                ch.pipeline().addLast(init);
                ctx.pipeline().remove(this);//移除当前handler
                ctx.fireChannelRegistered();//从当前handler往后抛出事件
            }
        };
//      ChannelHandler handler=new ChannelInitializer<Channel>() {
//          @Override
//          protected void initChannel(Channel ch) throws Exception {
//              channelGroup.add(ch);
//              LogLevel level=config.getLevel();
//              if(level!=null)
//              {
//                  ch.pipeline().addLast(new LoggerHandler(config.getLevel()));
//              }
//              ch.pipeline().addLast(init);
//          }
//      };
        return handler;
    }

@Override
protected final ChannelFuture doBind(InetSocketAddress local) {
    booter.localAddress(local);
    initServerOptions(optionConfig());
    ChannelHandler fixedHandler=fixHandlerBeforeDoBooterBind(handler);//修正handler
    return doBooterBind(local,fixedHandler);//启动端口绑定
}