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

@Test
public void initChannel_adds_HttpRequestDecoder_as_the_first_inbound_handler_after_sslCtx() {
    // given
    HttpChannelInitializer hci = basicHttpChannelInitializerNoUtilityHandlers();

    // when
    hci.initChannel(socketChannelMock);

    // then
    ArgumentCaptor<ChannelHandler> channelHandlerArgumentCaptor = ArgumentCaptor.forClass(ChannelHandler.class);
    verify(channelPipelineMock, atLeastOnce()).addLast(anyString(), channelHandlerArgumentCaptor.capture());
    List<ChannelHandler> handlers = channelHandlerArgumentCaptor.getAllValues();
    Pair<Integer, ChannelInboundHandler> firstInboundHandler = findChannelHandler(handlers, ChannelInboundHandler.class);
    Pair<Integer, HttpRequestDecoder> foundHandler = findChannelHandler(handlers, HttpRequestDecoder.class);

    assertThat(firstInboundHandler, notNullValue());
    assertThat(foundHandler, notNullValue());

    // No SSL Context was passed, so HttpRequestDecoder should be the first inbound handler.
    assertThat(foundHandler.getLeft(), is(firstInboundHandler.getLeft()));
    assertThat(foundHandler.getRight(), is(firstInboundHandler.getRight()));
}

/**
 * When this channel is registered, we will add all the ChannelHandlers passed into our
 * constructor to the pipeline.
 */
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  /**
   * This check is necessary as a channel may be registered with different event loops during it
   * lifetime and we only want to configure it once.
   */
  if (handlers != null) {
    for (ChannelHandler handler : handlers) {
      ctx.pipeline().addBefore(ctx.name(), null, handler);
    }
    ChannelHandler handler0 = handlers[0];
    ChannelHandlerContext handler0Ctx = ctx.pipeline().context(handlers[0]);
    handlers = null;
    if (handler0Ctx != null) { // The handler may have removed itself immediately
      if (handler0 instanceof ChannelInboundHandler) {
        ((ChannelInboundHandler) handler0).channelRegistered(handler0Ctx);
      } else {
        handler0Ctx.fireChannelRegistered();
      }
    }
  } else {
    super.channelRegistered(ctx);
  }
}

/**
 * Write to channel.
 *
 * @param channel the channel
 * @param magicBytes the magic bytes
 * @param pathOrCommand the path or command
 * @param attachment the attachment
 */
private void writeToChannel(Channel channel, String[] magicBytes, Object pathOrCommand, Object attachment) {
    long firstAttempt = System.currentTimeMillis();
    long timeOut = RemotingConstants.TEN * RemotingConstants.THOUSAND;
    while (!channel.isOpen() || !channel.isActive()) {
        if (System.currentTimeMillis() - firstAttempt >= timeOut) {
            try {
                throw new TimeoutException();
            } catch (TimeoutException e) {
                logger.error("Waited for 10 sec for connection reattempt to JumbuneAgent, but failed to connect", e);
            }
            break;
        }
    }
    if (channel.isActive()) {
        logger.debug("channel #" + channel.hashCode() + " connected");
    } else {
        logger.warn("channel #" + channel.hashCode() + " still disconnected, about to write on disconnected Channel");
    }
    if (attachment != null && attachment instanceof CyclicBarrier) {
        channel.attr(RemotingConstants.barrierKey).set((CyclicBarrier)attachment);
    }else if (attachment != null) {
        channel.attr(RemotingConstants.handlerKey).set((ChannelInboundHandler)attachment);
    }
    channel.write(Unpooled.wrappedBuffer(magicBytes[0].getBytes(), magicBytes[1].getBytes(), magicBytes[2].getBytes()));
    channel.write(pathOrCommand);
    channel.flush();
}

@Override
public ChannelInboundHandler newInstance() {
    NetHandler handler = new NetHandler();
    handler.setChannelHandlerFactory(this);
    handler.setCache(cache);
    return handler;
}

public OlapPipelineFactory(ChannelInboundHandler submitHandler, ChannelInboundHandler cancelHandler, ChannelInboundHandler statusHandler){
    this.submitHandler=submitHandler;
    this.cancelHandler=cancelHandler;
    this.statusHandler=statusHandler;

    this.decoder = new ProtobufDecoder(OlapMessage.Command.getDefaultInstance(),buildExtensionRegistry());
}

public void startServer(SConfiguration config) throws IOException {

        ScheduledExecutorService executor = Executors.newScheduledThreadPool(15, new ThreadFactoryBuilder().setNameFormat("OlapServer-%d").setDaemon(true).build());

        SpliceLogUtils.warn(LOG, "Olap Server starting (binding to port %s)...", port);

        ServerBootstrap bootstrap = new ServerBootstrap();

        // Instantiate handler once and share it
        OlapJobRegistry registry = new MappedJobRegistry(config.getOlapClientTickTime(),
                config.getOlapServerTickLimit(),
                TimeUnit.MILLISECONDS);
        ChannelInboundHandler submitHandler = new OlapRequestHandler(config,
                registry,clock,config.getOlapClientTickTime());
        ChannelInboundHandler statusHandler = new OlapStatusHandler(registry);
        ChannelInboundHandler cancelHandler = new OlapCancelHandler(registry);

        bossGroup = new NioEventLoopGroup(2, new ThreadFactoryBuilder().setNameFormat("OlapServer-boss-%d").setDaemon(true).build());
        workerGroup = new NioEventLoopGroup(15, new ThreadFactoryBuilder().setNameFormat("OlapServer-%d").setDaemon(true).build());
        bootstrap.group(bossGroup, workerGroup);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new OlapPipelineFactory(submitHandler,cancelHandler,statusHandler));
        bootstrap.option(ChannelOption.TCP_NODELAY, false);
        bootstrap.childOption(ChannelOption.TCP_NODELAY, false);
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.childOption(ChannelOption.SO_REUSEADDR, true);

        try {
            this.channel = bootstrap.bind(new InetSocketAddress(getPortNumber())).sync().channel();
        } catch (InterruptedException e) {
            throw new IOException(e);
        }
        port = ((InetSocketAddress)channel.localAddress()).getPort();

        SpliceLogUtils.warn(LOG, "Olap Server started at port " + port);

    }

public ChannelInboundHandler decoder() {
    return decoder;
}

@Test
public void itCompletesCloseFutureExceptionallyWhenTheConnectionIsClosed() throws Exception {
  ChannelInboundHandler errorHandler = getErrorHandler();

  Exception testException = new Exception();
  ChannelHandlerContext context = mock(ChannelHandlerContext.class);

  errorHandler.exceptionCaught(context, testException);

  verify(context).close();

  errorHandler.channelInactive(context);

  assertThat(session.getCloseFuture().isCompletedExceptionally()).isTrue();
  assertThatThrownBy(() -> session.getCloseFuture().get()).hasCause(testException);
}

private ChannelInboundHandler getErrorHandler() {
  ArgumentCaptor<ChannelHandler> captor = ArgumentCaptor.forClass(ChannelHandler.class);
  verify(pipeline).addLast(captor.capture());
  return (ChannelInboundHandler) captor.getValue();
}

public SocketThread(String strHost, int nPort, ChannelInboundHandler handler) {
    this.strHost = strHost;
    this.nPort = nPort;
    init(handler);
}

@Override
protected ChannelInboundHandler createHttpRequestHandlerForHttp() {
    return new SpdyServerHandler();
}

@Override
protected ChannelInboundHandler createHttpRequestHandlerForHttp() {
    return new SpdyServerHandler();
}

@Test(timeout = 10000)
public void testBindDeadLock() throws Exception {
    EventLoopGroup groupA = new LocalEventLoopGroup(1);
    EventLoopGroup groupB = new LocalEventLoopGroup(1);

    try {
        ChannelInboundHandler dummyHandler = new DummyHandler();

        final Bootstrap bootstrapA = new Bootstrap();
        bootstrapA.group(groupA);
        bootstrapA.channel(LocalChannel.class);
        bootstrapA.handler(dummyHandler);

        final Bootstrap bootstrapB = new Bootstrap();
        bootstrapB.group(groupB);
        bootstrapB.channel(LocalChannel.class);
        bootstrapB.handler(dummyHandler);

        List<Future<?>> bindFutures = new ArrayList<Future<?>>();

        // Try to bind from each other.
        for (int i = 0; i < 1024; i ++) {
            bindFutures.add(groupA.next().submit(new Runnable() {
                @Override
                public void run() {
                    bootstrapB.bind(LocalAddress.ANY);
                }
            }));

            bindFutures.add(groupB.next().submit(new Runnable() {
                @Override
                public void run() {
                    bootstrapA.bind(LocalAddress.ANY);
                }
            }));
        }

        for (Future<?> f: bindFutures) {
            f.sync();
        }
    } finally {
        groupA.shutdownGracefully();
        groupB.shutdownGracefully();
        groupA.terminationFuture().sync();
        groupB.terminationFuture().sync();
    }
}

@Test(timeout = 10000)
public void testConnectDeadLock() throws Exception {
    EventLoopGroup groupA = new LocalEventLoopGroup(1);
    EventLoopGroup groupB = new LocalEventLoopGroup(1);

    try {
        ChannelInboundHandler dummyHandler = new DummyHandler();

        final Bootstrap bootstrapA = new Bootstrap();
        bootstrapA.group(groupA);
        bootstrapA.channel(LocalChannel.class);
        bootstrapA.handler(dummyHandler);

        final Bootstrap bootstrapB = new Bootstrap();
        bootstrapB.group(groupB);
        bootstrapB.channel(LocalChannel.class);
        bootstrapB.handler(dummyHandler);

        List<Future<?>> bindFutures = new ArrayList<Future<?>>();

        // Try to connect from each other.
        for (int i = 0; i < 1024; i ++) {
            bindFutures.add(groupA.next().submit(new Runnable() {
                @Override
                public void run() {
                    bootstrapB.connect(LocalAddress.ANY);
                }
            }));

            bindFutures.add(groupB.next().submit(new Runnable() {
                @Override
                public void run() {
                    bootstrapA.connect(LocalAddress.ANY);
                }
            }));
        }

        for (Future<?> f: bindFutures) {
            f.sync();
        }
    } finally {
        groupA.shutdownGracefully();
        groupB.shutdownGracefully();
        groupA.terminationFuture().sync();
        groupB.terminationFuture().sync();
    }
}

public static CompletionStage<RawMemcacheClient> connect(
        final HostAndPort address,
        final int outstandingRequestLimit,
        final boolean binary,
        final Executor executor,
        final long timeoutMillis,
        final Charset charset,
        final Metrics metrics,
        final int maxSetLength) {

  final ChannelInboundHandler decoder;
  if (binary) {
    decoder = new BinaryMemcacheDecoder();
  } else {
    decoder = new AsciiMemcacheDecoder(charset);
  }

  final ChannelHandler initializer = new ChannelInitializer<Channel>() {
    @Override
    protected void initChannel(final Channel ch) throws Exception {
      ch.pipeline().addLast(
          new TcpTuningHandler(),
          decoder,

          // Downstream
          new MemcacheEncoder()
      );
    }
  };

  final CompletableFuture<RawMemcacheClient> clientFuture = new CompletableFuture<>();

  final Bootstrap bootstrap = new Bootstrap()
      .group(EVENT_LOOP_GROUP)
      .handler(initializer)
      .channel(NioSocketChannel.class)
      .option(ChannelOption.MESSAGE_SIZE_ESTIMATOR, SimpleSizeEstimator.INSTANCE);

  final ChannelFuture connectFuture = bootstrap.connect(
      new InetSocketAddress(address.getHostText(), address.getPort()));

  connectFuture.addListener((ChannelFutureListener) future -> {
    if (future.isSuccess()) {
      // Create client
      final RawMemcacheClient client = new DefaultRawMemcacheClient(
          address,
          future.channel(),
          outstandingRequestLimit,
          executor,
          timeoutMillis,
          metrics,
          maxSetLength);
      clientFuture.complete(client);
    } else {
      future.channel().close();
      clientFuture.completeExceptionally(future.cause());
    }
  });

  return onExecutor(clientFuture, executor);
}

public void setDelegate(ChannelInboundHandler delegate) {
  this.delegate = delegate;
}

public TransactionTimeoutHandler(String id, ChannelHandlerContext context, ChannelInboundHandler handler) {
    this.id = id;
    this.context = context;
    this.handler = handler;
}

@Test(timeout = 10000)
public void testBindDeadLock() throws Exception {
    EventLoopGroup groupA = new LocalEventLoopGroup(1);
    EventLoopGroup groupB = new LocalEventLoopGroup(1);

    try {
        ChannelInboundHandler dummyHandler = new DummyHandler();

        final Bootstrap bootstrapA = new Bootstrap();
        bootstrapA.group(groupA);
        bootstrapA.channel(LocalChannel.class);
        bootstrapA.handler(dummyHandler);

        final Bootstrap bootstrapB = new Bootstrap();
        bootstrapB.group(groupB);
        bootstrapB.channel(LocalChannel.class);
        bootstrapB.handler(dummyHandler);

        List<Future<?>> bindFutures = new ArrayList<Future<?>>();

        // Try to bind from each other.
        for (int i = 0; i < 1024; i ++) {
            bindFutures.add(groupA.next().submit(new Runnable() {
                @Override
                public void run() {
                    bootstrapB.bind(LocalAddress.ANY);
                }
            }));

            bindFutures.add(groupB.next().submit(new Runnable() {
                @Override
                public void run() {
                    bootstrapA.bind(LocalAddress.ANY);
                }
            }));
        }

        for (Future<?> f: bindFutures) {
            f.sync();
        }
    } finally {
        groupA.shutdownGracefully();
        groupB.shutdownGracefully();
        groupA.terminationFuture().sync();
        groupB.terminationFuture().sync();
    }
}

@Test(timeout = 10000)
public void testConnectDeadLock() throws Exception {
    EventLoopGroup groupA = new LocalEventLoopGroup(1);
    EventLoopGroup groupB = new LocalEventLoopGroup(1);

    try {
        ChannelInboundHandler dummyHandler = new DummyHandler();

        final Bootstrap bootstrapA = new Bootstrap();
        bootstrapA.group(groupA);
        bootstrapA.channel(LocalChannel.class);
        bootstrapA.handler(dummyHandler);

        final Bootstrap bootstrapB = new Bootstrap();
        bootstrapB.group(groupB);
        bootstrapB.channel(LocalChannel.class);
        bootstrapB.handler(dummyHandler);

        List<Future<?>> bindFutures = new ArrayList<Future<?>>();

        // Try to connect from each other.
        for (int i = 0; i < 1024; i ++) {
            bindFutures.add(groupA.next().submit(new Runnable() {
                @Override
                public void run() {
                    bootstrapB.connect(LocalAddress.ANY);
                }
            }));

            bindFutures.add(groupB.next().submit(new Runnable() {
                @Override
                public void run() {
                    bootstrapA.connect(LocalAddress.ANY);
                }
            }));
        }

        for (Future<?> f: bindFutures) {
            f.sync();
        }
    } finally {
        groupA.shutdownGracefully();
        groupB.shutdownGracefully();
        groupA.terminationFuture().sync();
        groupB.terminationFuture().sync();
    }
}

@Override
public ChannelInboundHandler newInstance() {
    return new IOClientHandler(this);
}

@Override
public ChannelInboundHandler newInstance() {
    return new IOServerHandler(this);
}

@Override
public abstract ChannelInboundHandler newInstance();

@Override
public ChannelInboundHandler newDecoder() {
    return new DefaultIByteArrayDecoderHandler(maxDataLength, headLenType);
}

@Override
public ChannelInboundHandler newDecoder() {
    return new ChannelDecoder();
}

@Test
public void testCreation() throws Exception {
  ChannelHandler handler = provider.createContainer(ChannelHandler.class, applicationHandler);

  assertNotNull(handler);

  ChannelInboundHandler inboundHandler = provider.createContainer(ChannelInboundHandler.class, applicationHandler);

  assertNotNull(inboundHandler);

  NettyRestHandlerContainer container = provider.createContainer(NettyRestHandlerContainer.class, applicationHandler);

  assertNotNull(container);
}

/**
 * Create a new {@link ChannelInboundHandler} that will invoke
 * {@link BiConsumer#accept} on
 * {@link ChannelInboundHandler#channelRead(ChannelHandlerContext, Object)}.
 *
 * @param handler the channel-read callback
 *
 * @return a marking event used when a netty connector handler terminates
 */
static ChannelInboundHandler inboundHandler(BiConsumer<? super ChannelHandlerContext, Object> handler) {
    return new ReactorNetty.ExtractorHandler(handler);
}

/**
 * Returns a ChannelInboundHandler that sends the first message of the correct type it receives
 * into the given observer.
 *
 * @param observer observer to send the message to
 * @param <T> type of message to read. All other messages are forwarded upstream.
 */
public static <T> ChannelInboundHandler channelReader(
    SingleObserver<T> observer, Class<T> klass) {
  return new ChannelSingleObserver<>(observer, klass);
}

/**
 * Create the {@link ChannelInboundHandler} that is responsible for handling the http requests
 * when the {@link SelectedProtocol} was {@link SelectedProtocol#HTTP_1_0} or
 * {@link SelectedProtocol#HTTP_1_1}
 */
protected abstract ChannelInboundHandler createHttpRequestHandlerForHttp();

/**
 * Create the {@link ChannelInboundHandler} that is responsible for handling the http responses
 * when the {@link SelectedProtocol} was {@link SelectedProtocol#SPDY_3_1}.
 *
 * By default this getMethod will just delecate to {@link #createHttpRequestHandlerForHttp()}, but sub-classes may
 * override this to change the behaviour.
 */
protected ChannelInboundHandler createHttpRequestHandlerForSpdy() {
    return createHttpRequestHandlerForHttp();
}

/**
 * Create the {@link ChannelInboundHandler} that is responsible for handling the http requests
 * when the {@link SelectedProtocol} was {@link SelectedProtocol#HTTP_1_0} or
 * {@link SelectedProtocol#HTTP_1_1}
 */
protected abstract ChannelInboundHandler createHttpRequestHandlerForHttp();