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

项目：hekate 文件：MulticastSeedNodeProvider.java

private SimpleChannelInboundHandler<DatagramPacket> createListenerHandler(SeedNode thisNode, ByteBuf seedNodeInfo) {
    return new SimpleChannelInboundHandler<DatagramPacket>() {
        @Override
        public void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
            ByteBuf buf = msg.content();

            if (buf.readableBytes() > 4 && buf.readInt() == Utils.MAGIC_BYTES) {
                MessageTYpe msgType = MessageTYpe.values()[buf.readByte()];

                if (msgType == MessageTYpe.DISCOVERY) {
                    String cluster = decodeUtf(buf);
                    InetSocketAddress address = decodeAddress(buf);

                    if (thisNode.cluster().equals(cluster) && !address.equals(thisNode.address())) {
                        onDiscoveryMessage(address);

                        DatagramPacket response = new DatagramPacket(seedNodeInfo.copy(), msg.sender());

                        ctx.writeAndFlush(response);
                    }
                }
            }
        }
    };
}

项目：riposte 文件：ComponentTestUtils.java

public static Bootstrap createNettyHttpClientBootstrap() {
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(new NioEventLoopGroup())
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new HttpClientCodec());
                     p.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
                     p.addLast("clientResponseHandler", new SimpleChannelInboundHandler<HttpObject>() {
                         @Override
                         protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
                             throw new RuntimeException("Client response handler was not setup before the call");
                         }
                     });
                 }
             });

    return bootstrap;
}

项目：riposte 文件：ComponentTestUtils.java

public static CompletableFuture<NettyHttpClientResponse> setupNettyHttpClientResponseHandler(
    Channel ch, Consumer<ChannelPipeline> pipelineAdjuster
) {
    CompletableFuture<NettyHttpClientResponse> responseFromServerFuture = new CompletableFuture<>();
    ch.pipeline().replace("clientResponseHandler", "clientResponseHandler", new SimpleChannelInboundHandler<HttpObject>() {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg)
            throws Exception {
            if (msg instanceof FullHttpResponse) {
                // Store the proxyServer response for asserting on later.
                responseFromServerFuture.complete(new NettyHttpClientResponse((FullHttpResponse) msg));
            } else {
                // Should never happen.
                throw new RuntimeException("Received unexpected message type: " + msg.getClass());
            }
        }
    });

    if (pipelineAdjuster != null)
        pipelineAdjuster.accept(ch.pipeline());

    return responseFromServerFuture;
}

项目：async-gamequery-lib 文件：ErrorHandler.java

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    if (log.isDebugEnabled()) {
        log.error("Unhandled exception caught within the pipeline {} for Channel {}, Id: {}", cause, ctx.channel(), ctx.channel().id());
        if (ctx.channel().hasAttr(ChannelAttributes.LAST_REQUEST_SENT)) {
            AbstractRequest request = ctx.channel().attr(ChannelAttributes.LAST_REQUEST_SENT).get();
            if (request != null && SocketChannel.class.isAssignableFrom(ctx.channel().getClass())) {
                Throwable ex = new ResponseException(request, cause);
                SimpleChannelInboundHandler responseRouter = ctx.pipeline().get(SimpleChannelInboundHandler.class);
                responseRouter.channelRead(ctx, ex);
                return;
            }
        }
        throw new TransportException(cause);
    }
}

项目：tealess 文件：HTTPSInitializer.java

@Override
protected void initChannel(SocketChannel ch) throws Exception {
  ChannelPipeline pipeline = ch.pipeline();

  SSLEngine engine = context.createSSLEngine();
  engine.setUseClientMode(true);
  SslHandler sslHandler = new SslHandler(engine);
  //pipeline.addLast(sslHandler);
  pipeline.addLast(new SimpleChannelInboundHandler<Object>() {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
      System.out.println(msg);
    }
  });
  //pipeline.addLast(new HttpRequestDecoder());
  //pipeline.addLast(new HttpResponseEncoder());
  //pipeline.addLast(new HttpContentCompressor());
  //pipeline.addLast(new HTTPClientHandler());
}

项目：netty-http-server 文件：ServerConfig.java

@Bean(name = "channelInitializer")
public ChannelInitializer<SocketChannel> initializerFactory(final ApplicationContext contxt) {
    return new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            SimpleChannelInboundHandler<?> channelInboundHandler = contxt.getBean(NettyHttpChannelHandler.class);
            ChannelPipeline pipeline = ch.pipeline();
            // HTTP
            pipeline.addLast(new HttpRequestDecoder());
            pipeline.addLast(new HttpResponseEncoder());
            pipeline.addLast(new HttpContentCompressor());
            pipeline.addLast(new ChunkedWriteHandler());
            // 设置处理的Handler
            pipeline.addLast(channelInboundHandler);
        }
    };
}

项目：netty-jssc 文件：SimpleLineBasedSerialChannel.java

public SimpleLineBasedSerialChannel(String port, final SimpleStringChannelHandler stringHandler) {
    group = new OioEventLoopGroup();
       Bootstrap b = new Bootstrap();
       b.group(group)
        .channel(JsscChannel.class)
        .handler(new ChannelInitializer<JsscChannel>() {
            @Override
            public void initChannel(JsscChannel ch) throws Exception {
                ch.pipeline().addLast(
                    new LineBasedFrameDecoder(Integer.MAX_VALUE),
                    new StringDecoder(),
                    new SimpleChannelInboundHandler<String>() {
                     @Override
                     protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, String msg) throws Exception {
                        stringHandler.channelRead(ctx, msg); 
                     }
                 }
                );
            }
        });

        f = b.connect(new JsscDeviceAddress(port)).syncUninterruptibly();
}

项目：sipstack 文件：SimpleSipStack.java

private ServerBootstrap createTCPListeningPoint(final SimpleChannelInboundHandler<SipMessageEvent> handler) {
    final ServerBootstrap b = new ServerBootstrap();

    b.group(this.bossGroup, this.workerGroup)
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(final SocketChannel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("decoder", new SipMessageStreamDecoder());
            pipeline.addLast("encoder", new SipMessageEncoder());
            pipeline.addLast("handler", handler);
        }
    })
    .option(ChannelOption.SO_BACKLOG, 128)
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
    .childOption(ChannelOption.SO_KEEPALIVE, true)
    .childOption(ChannelOption.TCP_NODELAY, true);
    return b;
}

项目：piezo 文件：ChannelInitializers.java

/**
 * Returns a new channel initializer suited to encode and decode a protocol
 * buffer message.
 * <p/>
 * <p>Message sizes over 10 MB are not supported.</p>
 * <p/>
 * <p>The handler will be executed on the I/O thread. Blocking operations
 * should be executed in their own thread.</p>
 *
 * @param defaultInstance an instance of the message to handle
 * @param handler the handler implementing the application logic
 * @param <M> the type of the support protocol buffer message
 */
public static final <M extends Message> ChannelInitializer<Channel> protoBuf(
    final M defaultInstance, final SimpleChannelInboundHandler<M> handler) {
  return new ChannelInitializer<Channel>() {

    @Override
    protected void initChannel(Channel channel) throws Exception {
      channel.pipeline().addLast("frameDecoder",
          new LengthFieldBasedFrameDecoder(10 * 1024 * 1024, 0, 4, 0, 4));
      channel.pipeline().addLast("protobufDecoder",
          new ProtobufDecoder(defaultInstance));
      channel.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
      channel.pipeline().addLast("protobufEncoder", new ProtobufEncoder());
      channel.pipeline().addLast("applicationHandler", handler);
    }
  };
}

项目：piezo 文件：ChannelInitializers.java

/**
 * Returns a server-side channel initializer capable of securely receiving
 * and sending HTTP requests and responses
 * <p/>
 * <p>Communications will be encrypted as per the configured SSL context</p>
 *
 * @param handler the handler implementing the business logic.
 * @param sslContext the SSL context which drives the security of the
 * link to the client.
 */
public static final ChannelInitializer<Channel> secureHttpServer(
    final SimpleChannelInboundHandler<HttpRequest> handler,
    final SSLContext sslContext) {
  return new ChannelInitializer<Channel>() {
    @Override
    protected void initChannel(Channel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();
      SSLEngine sslEngine = sslContext.createSSLEngine();
      sslEngine.setUseClientMode(false);
      pipeline.addLast("ssl", new SslHandler(sslEngine));
      pipeline.addLast("httpCodec", new HttpServerCodec());
      pipeline.addLast("aggregator", new HttpObjectAggregator(10 * 1024 * 1024));
      pipeline.addLast("httpServerHandler", handler);
    }
  };
}

项目：piezo 文件：ChannelInitializers.java

/**
 * Returns a client-side channel initializer capable of securely sending
 * and receiving HTTP requests and responses.
 * <p/>
 * <p>Communications will be encrypted as per the configured SSL context</p>
 *
 * @param handler the handler in charge of implementing the business logic
 * @param sslContext the SSL context which drives the security of the
 * link to the server.
 */
public static final ChannelInitializer<Channel> secureHttpClient(
    final SimpleChannelInboundHandler<HttpResponse> handler,
    final SSLContext sslContext) {
  return new ChannelInitializer<Channel>() {

    @Override
    protected void initChannel(Channel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();
      SSLEngine sslEngine = sslContext.createSSLEngine();
      sslEngine.setUseClientMode(true);
      pipeline.addLast("ssl", new SslHandler(sslEngine));
      pipeline.addLast("httpCodec", new HttpClientCodec());
      pipeline.addLast("aggregator", new HttpObjectAggregator(10 * 1024 * 1024));
      pipeline.addLast("httpClientHandler", handler);
    }
  };
}

项目：nitmproxy 文件：NitmProxyInitializer.java

@Override
protected void initChannel(Channel channel) throws Exception {
    InetSocketAddress address = (InetSocketAddress) channel.remoteAddress();
    Address clientAddress = new Address(address.getHostName(), address.getPort());
    channel.pipeline().addLast(
            master.proxyHandler(clientAddress),
            new SimpleChannelInboundHandler<Object>() {
                @Override
                protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o)
                        throws Exception {
                    LOGGER.info("[Client ({})] => Unhandled inbound: {}", clientAddress, o);
                }
            });
}

项目：netty-connection-pool 文件：EpollConnDroppingServer.java

public EpollConnDroppingServer(final int port, final int dropEveryRequest)
throws InterruptedException {
    dispatchGroup = new EpollEventLoopGroup();
    workerGroup = new EpollEventLoopGroup();
    final ServerBootstrap bootstrap = new ServerBootstrap()
        .group(dispatchGroup, workerGroup)
        .channel(EpollServerSocketChannel.class)
        .childHandler(
            new ChannelInitializer<SocketChannel>() {
                @Override
                public final void initChannel(final SocketChannel ch) {
                    if(dropEveryRequest > 0) {
                        ch.pipeline().addLast(
                            new SimpleChannelInboundHandler<Object>() {
                                @Override
                                protected final void channelRead0(
                                    final ChannelHandlerContext ctx, final Object msg
                                ) throws Exception {
                                    if(0 == reqCounter.incrementAndGet() % dropEveryRequest) {
                                        final Channel conn = ctx.channel();
                                        System.out.println("Dropping the connection " + conn);
                                        conn.close();
                                    }
                                }
                            }
                        );
                    }
                }
            }
        );

    bindFuture = bootstrap.bind(port).sync();
}

项目：netty-connection-pool 文件：NioConnDroppingServer.java

public NioConnDroppingServer(final int port, final int dropEveryRequest)
throws InterruptedException {
    dispatchGroup = new NioEventLoopGroup();
    workerGroup = new NioEventLoopGroup();
    final ServerBootstrap bootstrap = new ServerBootstrap()
        .group(dispatchGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(
            new ChannelInitializer<SocketChannel>() {
                @Override
                public final void initChannel(final SocketChannel ch) {
                    ch.pipeline().addLast(
                        new SimpleChannelInboundHandler<Object>() {
                            @Override
                            protected final void channelRead0(
                                final ChannelHandlerContext ctx, final Object msg
                            ) throws Exception {
                                if(0 == reqCounter.incrementAndGet() % dropEveryRequest) {
                                    final Channel conn = ctx.channel();
                                    System.out.println("Dropping the connection " + conn);
                                    conn.close();
                                }
                            }
                        }
                    );
                }
            }
        );

    bindFuture = bootstrap.bind(port).sync();
}

项目：mongoose-base 文件：HttpStorageDriverBase.java

protected FullHttpResponse executeHttpRequest(final FullHttpRequest request)
throws InterruptedException, ConnectException {

    ThreadContext.put(KEY_TEST_STEP_ID, stepId);
    ThreadContext.put(KEY_CLASS_NAME, CLS_NAME);

    final Channel channel = getUnpooledConnection();
    try {
        final ChannelPipeline pipeline = channel.pipeline();
        Loggers.MSG.debug(
            "{}: execute the HTTP request using the channel {} w/ pipeline: {}", stepId,
            channel.hashCode(), pipeline
        );
        pipeline.removeLast(); // remove the API specific handler
        final SynchronousQueue<FullHttpResponse> fullRespSync = new SynchronousQueue<>();
        pipeline.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
        pipeline.addLast(
            new SimpleChannelInboundHandler<HttpObject>() {
                @Override
                protected final void channelRead0(
                    final ChannelHandlerContext ctx, final HttpObject msg
                ) throws Exception {
                    if(msg instanceof FullHttpResponse) {
                        fullRespSync.put(((FullHttpResponse) msg).retain());
                    }
                }
            }
        );
        channel.writeAndFlush(request).sync();
        return fullRespSync.take();
    } finally {
        channel.close();
    }
}