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

public void setServerHandler(SimpleChannelInboundHandler<T> handler) {      
    FMLEmbeddedChannel channel = channels.get(Side.SERVER);
    String codecName = channel.findChannelHandlerNameForType(codec.getClass());
    if(handlers.get(Side.SERVER) != null) {
        channel.pipeline().remove("ServerHandler");
    }
       channel.pipeline().addAfter(codecName, "ServerHandler", handler);
       handlers.put(Side.SERVER, handler);
}

public ListenableFuture<CommandReply> sendRequest(CommandRpcRequest<?> request,
                                                  InetSocketAddress remoteAddress) {
  SettableFuture<CommandReply> replyMessageFuture = SettableFuture.create();
  ChannelFuture connectFuture = client.connect(remoteAddress);
  connectFuture.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
      if (future.isSuccess()) {
        future.channel().pipeline().addLast(new SimpleChannelInboundHandler<CommandReply>() {
          @Override
          protected void channelRead0(ChannelHandlerContext ctx, CommandReply msg) throws Exception {
            replyMessageFuture.set(msg);
            ctx.channel().close();
          }
        });

        // connected is fine, flush message:
        future.channel().writeAndFlush(request);
      } else {
        replyMessageFuture.setException(future.cause());
        future.channel().close();
      }
    }
  });

  return replyMessageFuture;
}

@Override
protected void init(SpecificTransportContext<AvroTcpConfig> context)
    throws TransportLifecycleException {
  AvroTcpConfig configuration = context.getConfiguration();
  configuration.setBindInterface(replaceProperty(configuration.getBindInterface(),
      BIND_INTERFACE_PROP_NAME,
      context.getCommonProperties().getProperty(BIND_INTERFACE_PROP_NAME, LOCALHOST)));

  configuration.setPublicInterface(replaceProperty(configuration.getPublicInterface(),
      PUBLIC_INTERFACE_PROP_NAME,
      context.getCommonProperties().getProperty(PUBLIC_INTERFACE_PROP_NAME, LOCALHOST)));

  final KaaTcpCommandFactory factory = new KaaTcpCommandFactory();
  this.netty = new
      AbstractNettyServer(configuration.getBindInterface(), configuration.getBindPort()) {

    @Override
    protected ChannelInitializer<SocketChannel> configureInitializer() throws Exception {
      return new AbstractKaaTcpServerInitializer() {
        @Override
        protected SimpleChannelInboundHandler<AbstractKaaTcpCommandProcessor> getMainHandler(
            UUID uuid) {
          return new TcpHandler(uuid, TcpTransport.this.handler);
        }

        @Override
        protected KaaTcpDecoder getDecoder() {
          return new KaaTcpDecoder(factory);
        }
      };
    }
  };
}

@Test
public void listenerTest() throws Exception {
    final CountDownLatch awaitLatch = new CountDownLatch(1);

    Bootstrap bootstrap = client();
    Channel channel = bootstrap.connect(webServer.getHostname(), webServer.getListeningPort()).sync().channel();

    channel.pipeline().addLast(new SimpleChannelInboundHandler<FullHttpResponse>() {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
            awaitLatch.countDown();
        }
    });

    HttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
    channel.writeAndFlush(request);

    boolean await = awaitLatch.await(3000, TimeUnit.MILLISECONDS);
    Assert.assertTrue(await);

    PluginTestVerifier verifier = PluginTestVerifierHolder.getInstance();
    verifier.printCache();

    verifier.verifyTrace(event("NETTY", Bootstrap.class.getMethod("connect", SocketAddress.class), annotation("netty.address", webServer.getHostAndPort())));
    verifier.verifyTrace(event("NETTY", "io.netty.channel.DefaultChannelPipeline.writeAndFlush(java.lang.Object)"));
    verifier.verifyTrace(event("ASYNC", "Asynchronous Invocation"));
    verifier.verifyTrace(event("NETTY_HTTP", "io.netty.handler.codec.http.HttpObjectEncoder.encode(io.netty.channel.ChannelHandlerContext, java.lang.Object, java.util.List)", annotation("http.url", "/")));
}

@Test
public void writeTest() throws Exception {
    final CountDownLatch awaitLatch = new CountDownLatch(1);

    Bootstrap bootstrap = client();
    bootstrap.connect(webServer.getHostname(), webServer.getListeningPort()).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                Channel channel = future.channel();
                channel.pipeline().addLast(new SimpleChannelInboundHandler() {

                    @Override
                    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
                        awaitLatch.countDown();
                    }

                });
                HttpRequest request = new DefaultFullHttpRequest(
                        HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
                future.channel().writeAndFlush(request);
            }
        }

    });

    boolean await = awaitLatch.await(3000, TimeUnit.MILLISECONDS);
    Assert.assertTrue(await);

    PluginTestVerifier verifier = PluginTestVerifierHolder.getInstance();
    verifier.printCache();

    verifier.verifyTrace(event("NETTY", Bootstrap.class.getMethod("connect", SocketAddress.class), annotation("netty.address", webServer.getHostAndPort())));
    verifier.verifyTrace(event("NETTY", "io.netty.channel.DefaultChannelPromise.addListener(io.netty.util.concurrent.GenericFutureListener)"));
    verifier.verifyTrace(event("ASYNC", "Asynchronous Invocation"));
    verifier.verifyTrace(event("NETTY_INTERNAL", "io.netty.util.concurrent.DefaultPromise.notifyListenersNow()"));
    verifier.verifyTrace(event("NETTY_INTERNAL", "io.netty.util.concurrent.DefaultPromise.notifyListener0(io.netty.util.concurrent.Future, io.netty.util.concurrent.GenericFutureListener)"));
    verifier.verifyTrace(event("NETTY", "io.netty.channel.DefaultChannelPipeline.writeAndFlush(java.lang.Object)"));
    verifier.verifyTrace(event("NETTY_HTTP", "io.netty.handler.codec.http.HttpObjectEncoder.encode(io.netty.channel.ChannelHandlerContext, java.lang.Object, java.util.List)", annotation("http.url", "/")));
}

private void start() throws Exception {
    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(parent, child)
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
                pipeline.addLast("codec", new HttpServerCodec());
                pipeline.addLast("httpHandler", new SimpleChannelInboundHandler<Object>() {
                    @Override
                    protected void channelRead0(ChannelHandlerContext context, Object message) throws Exception {
                        if (message instanceof HttpRequest) {
                            // Keep track of processed requests
                            HttpRequest request = (HttpRequest) message;
                            requests.add(request);
                        }

                        if (message instanceof HttpContent) {
                            // Write an empty JSON response back to the client
                            FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.copiedBuffer("{}", CharsetUtil.UTF_8));
                            response.headers().set(CONTENT_TYPE, "application/json; charset=UTF-8");
                            response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
                            context.writeAndFlush(response);
                        }
                    }
                });
            }
        });

    channel = bootstrap.bind(port).syncUninterruptibly().channel();
}

public SimpleSipStack(final SimpleChannelInboundHandler<SipMessageEvent> handler, final String ip, final int port) {
    this.ip = ip;
    this.port = port;

    this.bootstrap = createUDPListeningPoint(handler);
    this.serverBootstrap = createTCPListeningPoint(handler);
}

private Bootstrap createUDPListeningPoint(final SimpleChannelInboundHandler<SipMessageEvent> handler) {
    final Bootstrap b = new Bootstrap();
    b.group(this.udpGroup)
    .channel(NioDatagramChannel.class)
    .handler(new ChannelInitializer<DatagramChannel>() {
        @Override
        protected void initChannel(final DatagramChannel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("decoder", new SipMessageDatagramDecoder());
            pipeline.addLast("encoder", new SipMessageEncoder());
            pipeline.addLast("handler", handler);
        }
    });
    return b;
}

/**
 * Returns a new chanel initializer suited to decode and process HTTP
 * requests.
 *
 * @param handler the handler implementing the application logic
 */
public static final ChannelInitializer<Channel> httpServer(
    final SimpleChannelInboundHandler<HttpRequest> handler) {
  Preconditions.checkArgument(handler.isSharable());
  return new ChannelInitializer<Channel>() {

    @Override
    protected void initChannel(Channel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();
      pipeline.addLast("httpCodec", new HttpServerCodec());
      pipeline.addLast("aggregator", new HttpObjectAggregator(10 * 1024 * 1024));
      pipeline.addLast("httpServerHandler", handler);
    }
  };
}

/**
 * Returns a channel initializer suited to decode and process HTTP responses.
 *
 * @param handler the handler implementing the application logic
 */
public static final ChannelInitializer<Channel> httpClient(
    final SimpleChannelInboundHandler<HttpResponse> handler) {
  return new ChannelInitializer<Channel>() {

    @Override
    protected void initChannel(Channel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();
      pipeline.addLast("httpCodec", new HttpClientCodec());
      pipeline.addLast("aggregator", new HttpObjectAggregator(10 * 1024 * 1024));
      pipeline.addLast("httpClientHandler", handler);
    }
  };
}