Java 类io.netty.handler.codec.http.HttpClientCodec 实例源码

/**
 * Adds pipelines to channel.
 * 
 *  @param ch channel to be operated on
 */
protected void initChannel(SocketChannel ch) throws Exception {
  ChannelPipeline pipe = ch.pipeline();

  if (ssl) {
    // HTTPs connection
    SSLEngine sslEng = getSsl(null);
    sslEng.setUseClientMode(true);
    pipe.addLast("SSL", new SslHandler(sslEng, false));
  }

  pipe.addFirst("Timer", new ReadTimeoutHandler(30));
  pipe.addLast("Codec", new HttpClientCodec());
  pipe.addLast("Inflater", new HttpContentDecompressor());
  pipe.addLast("Handler", new HTTPMessageHandler(builder));
}

/**
     * 适配
     */
    @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 pipelines to channel.
 * 
 *  @param ch channel to be operated on
 */
protected void initChannel(SocketChannel ch) throws Exception {
  ChannelPipeline pipe = ch.pipeline();

  if (ssl) {
    // HTTPs connection
    SSLEngine sslEng = getSsl(null);
    sslEng.setUseClientMode(true);
    pipe.addLast("SSL", new SslHandler(sslEng, false));
  }

  pipe.addFirst("Timer", new ReadTimeoutHandler(30));
  pipe.addLast("Codec", new HttpClientCodec());
  pipe.addLast("Inflater", new HttpContentDecompressor());
  pipe.addLast("Handler", new HTTPMessageHandler(builder));
}

@Override
public void channelCreated(Channel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();

    if (sslContext != null) {
        SslHandler handler = sslContext.newHandler(ch.alloc());
        p.addLast(handler);
        handler.handshakeFuture().addListener(future -> {
            if (!future.isSuccess()) {
                log.error(() -> "SSL handshake failed.", future.cause());
            }
        });
    }

    p.addLast(new HttpClientCodec());
    p.addLast(handlers);
    // Disabling auto-read is needed for backpressure to work
    ch.config().setOption(ChannelOption.AUTO_READ, false);
}

@Override
    public void initChannel(SocketChannel ch) throws Exception {
        // Create a default pipeline implementation.
        ChannelPipeline p = ch.pipeline();

        p.addLast("log", new LoggingHandler(LogLevel.INFO));
        // Enable HTTPS if necessary.
/*        if (ssl) {
            SSLEngine engine =
                SecureChatSslContextFactory.getClientContext().createSSLEngine();
            engine.setUseClientMode(true);

            p.addLast("ssl", new SslHandler(engine));
        }*/

        p.addLast("codec", new HttpClientCodec());

        // Remove the following line if you don't want automatic content decompression.
        p.addLast("inflater", new HttpContentDecompressor());

        // Uncomment the following line if you don't want to handle HttpChunks.
        //p.addLast("aggregator", new HttpObjectAggregator(1048576));

        p.addLast("handler", new HttpSnoopClientHandler());
    }

@Override
protected void initChannel(SocketChannel channel) throws SSLException {
    URI uri = config.getConnectionWebsocketUri();

    DefaultHttpHeaders headers = new DefaultHttpHeaders();
    headers.add(USER_ID_HEADER, config.getConnectionUserId().toString());
    headers.add(USER_PASSWORD_HEADER, config.getConnectionUserPassword());
    headers.add(SUPPLIER_ID_HEADER, config.getConnectionServerId());

    WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri, WS_VERSION, null, false, headers);

    ChannelPipeline pipeline = channel.pipeline();
    if (config.isConnectionSecure()) {
        try {
            SslContext sslContext = SslContext.newClientContext(InsecureTrustManagerFactory.INSTANCE);
            pipeline.addLast(sslContext.newHandler(channel.alloc()));
        } catch (SSLException e) {
            logger.log(Level.SEVERE, "Shutting down client due to unexpected failure to create SSL context", e);
            throw e;
        }
    }
    pipeline.addLast(new HttpClientCodec());
    pipeline.addLast(new HttpObjectAggregator(8192));
    pipeline.addLast(new AudioConnectClientHandler(handshaker));
}

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;
}

@Override
public void configure(final ChannelPipeline pipeline) {
    final String scheme = connection.getUri().getScheme();
    if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("Unsupported scheme (only ws: or wss: supported): " + scheme);

    if (!supportsSsl() && "wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("To use wss scheme ensure that enableSsl is set to true in configuration");

    final int maxContentLength = cluster.connectionPoolSettings().maxContentLength;
    handler = new WebSocketClientHandler(
            WebSocketClientHandshakerFactory.newHandshaker(
                    connection.getUri(), WebSocketVersion.V13, null, false, HttpHeaders.EMPTY_HEADERS, maxContentLength));

    pipeline.addLast("http-codec", new HttpClientCodec());
    pipeline.addLast("aggregator", new HttpObjectAggregator(maxContentLength));
    pipeline.addLast("ws-handler", handler);
    pipeline.addLast("gremlin-encoder", webSocketGremlinRequestEncoder);
    pipeline.addLast("gremlin-decoder", webSocketGremlinResponseDecoder);
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();

    if (sslCtx != null) {
        pipeline.addLast("ssl", sslCtx.newHandler(ch.alloc()));
    }

    pipeline.addLast("codec", new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    pipeline.addLast("inflater", new HttpContentDecompressor());

    // to be used since huge file transfer
    pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());

    pipeline.addLast("handler", new HttpUploadClientHandler());
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();

    // Enable HTTPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }

    p.addLast(new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    p.addLast(new HttpContentDecompressor());

    // Uncomment the following line if you don't want to handle HttpContents.
    //p.addLast(new HttpObjectAggregator(1048576));

    p.addLast(new HttpSnoopClientHandler());
}

@Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();

        // Enable HTTPS if necessary.
//      if (sslCtx != null) {
//          pipeline.addLast(sslCtx.newHandler(ch.alloc()));
//      }

        pipeline.addLast(new HttpClientCodec());
        // Remove the following line if you don't want automatic content decompression.
        pipeline.addLast(new HttpContentDecompressor());

        // Uncomment the following line if you don't want to handle HttpContents.
        //pipeline.addLast(new HttpObjectAggregator(65536));
        pipeline.addLast(new HttpObjectAggregator(65536 * 3));


//      pipeline.addLast(new HttpClientHandler(null, mHttpClientListener));
    }

@Provides
@HttpsRelayProtocol
static ImmutableList<Provider<? extends ChannelHandler>> provideHandlerProviders(
    Provider<SslClientInitializer<NioSocketChannel>> sslClientInitializerProvider,
    Provider<HttpClientCodec> httpClientCodecProvider,
    Provider<HttpObjectAggregator> httpObjectAggregatorProvider,
    Provider<BackendMetricsHandler> backendMetricsHandlerProvider,
    Provider<LoggingHandler> loggingHandlerProvider,
    Provider<FullHttpResponseRelayHandler> relayHandlerProvider) {
  return ImmutableList.of(
      sslClientInitializerProvider,
      httpClientCodecProvider,
      httpObjectAggregatorProvider,
      backendMetricsHandlerProvider,
      loggingHandlerProvider,
      relayHandlerProvider);
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();

    if (sslCtx != null) {
        pipeline.addLast("ssl", sslCtx.newHandler(ch.alloc()));
    }

    pipeline.addLast("codec", new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    pipeline.addLast("inflater", new HttpContentDecompressor());

    // to be used since huge file transfer
    pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());

    pipeline.addLast("handler", new HttpUploadClientHandler());
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();

    // Enable HTTPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }

    p.addLast(new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    p.addLast(new HttpContentDecompressor());

    // Uncomment the following line if you don't want to handle HttpContents.
    //p.addLast(new HttpObjectAggregator(1048576));

    p.addLast(new HttpSnoopClientHandler());
}

@Override
protected ChannelInitializer<SocketChannel> getChannelInitializer() {
    return new ChannelInitializer<SocketChannel> () {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            if (sslCtx != null) {
                p.addLast(sslCtx.newHandler(ch.alloc(), host, port));
            }
            p.addLast(
                    new HttpClientCodec(),
                    new HttpObjectAggregator(8192),
                    handler,
                    new MessageDecoder(new GlobalStats())
            );
        }
    };
}

@Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();

        // Enable HTTPS if necessary.
//      if (sslCtx != null) {
//          pipeline.addLast(sslCtx.newHandler(ch.alloc()));
//      }

        pipeline.addLast(new HttpClientCodec());
        // Remove the following line if you don't want automatic content decompression.
        pipeline.addLast(new HttpContentDecompressor());

        // Uncomment the following line if you don't want to handle HttpContents.
        //pipeline.addLast(new HttpObjectAggregator(65536));
        pipeline.addLast(new HttpObjectAggregator(65536 * 3));


//      pipeline.addLast(new HttpClientHandler(null, mHttpClientListener));
    }

@Override
public void configure(final ChannelPipeline pipeline) {
    final String scheme = connection.getUri().getScheme();
    if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("Unsupported scheme (only ws: or wss: supported): " + scheme);

    if (!supportsSsl() && "wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("To use wss scheme ensure that enableSsl is set to true in configuration");

    final int maxContentLength = cluster.connectionPoolSettings().maxContentLength;
    handler = new WebSocketClientHandler(
            WebSocketClientHandshakerFactory.newHandshaker(
                    connection.getUri(), WebSocketVersion.V13, null, false, HttpHeaders.EMPTY_HEADERS, maxContentLength));

    pipeline.addLast("http-codec", new HttpClientCodec());
    pipeline.addLast("aggregator", new HttpObjectAggregator(maxContentLength));
    pipeline.addLast("ws-handler", handler);
    pipeline.addLast("gremlin-encoder", webSocketGremlinRequestEncoder);
    pipeline.addLast("gremlin-decoder", webSocketGremlinResponseDecoder);
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();

    // Enable HTTPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }

    p.addLast(new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    p.addLast(new HttpContentDecompressor());

    // Uncomment the following line if you don't want to handle HttpContents.
    //p.addLast(new HttpObjectAggregator(1048576));
    p.addLast(handler);
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    // Create a default pipeline implementation.
    ChannelPipeline pipeline = ch.pipeline();

    if (ssl) {
        SSLEngine engine = SecureChatSslContextFactory.getClientContext().createSSLEngine();
        engine.setUseClientMode(true);
        pipeline.addLast("ssl", new SslHandler(engine));
    }

    pipeline.addLast("codec", new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    pipeline.addLast("inflater", new HttpContentDecompressor());

    // to be used since huge file transfer
    pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());

    pipeline.addLast("handler", new HttpUploadClientHandler());
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    // Create a default pipeline implementation.
    ChannelPipeline p = ch.pipeline();

    p.addLast("log", new LoggingHandler(LogLevel.INFO));
    // Enable HTTPS if necessary.
    if (ssl) {
        SSLEngine engine =
            SecureChatSslContextFactory.getClientContext().createSSLEngine();
        engine.setUseClientMode(true);

        p.addLast("ssl", new SslHandler(engine));
    }

    p.addLast("codec", new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    p.addLast("inflater", new HttpContentDecompressor());

    // Uncomment the following line if you don't want to handle HttpChunks.
    //p.addLast("aggregator", new HttpObjectAggregator(1048576));

    p.addLast("handler", new HttpSnoopClientHandler());
}

private ChannelInitializer<Channel> buildInitializer() {
  // TODO(CK): This logic should be move outside of XioClientBootstrap to something HTTP related
  if (proto != null && (proto == Protocol.HTTP || proto == Protocol.HTTPS)) {
    applicationProtocol = () -> new HttpClientCodec();
  } else if (applicationProtocol == null) {
    throw new RuntimeException(
        "Cannot build initializer, specify either protocol or applicationProtocol");
  }

  ClientState state =
      new ClientState(
          config,
          address,
          handler,
          (ssl ? sslContext : null),
          applicationProtocol,
          tracingHandler);
  if (initializerFactory != null) {
    return initializerFactory.apply(state);
  }

  return new DefaultChannelInitializer(state);
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();

    // Enable HTTPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }

    p.addLast(new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    p.addLast(new HttpContentDecompressor());

    // Uncomment the following line if you don't want to handle HttpContents.
    p.addLast(new HttpObjectAggregator(1048576));

    p.addLast(nettyHttpClientHandler);
}

private void createClient() {
    client.group(ioWorkerGroup)
        .channel(NioSocketChannel.class)
        .option(ChannelOption.SO_REUSEADDR, true)
        .option(ChannelOption.TCP_NODELAY, true)
        .handler(new ChannelInitializer<SocketChannel>() {
          @Override
          protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
//            pipeline.addLast("logger", new LoggingHandler(LogLevel.WARN));
            pipeline.addLast("http-client", new HttpClientCodec());
            pipeline.addLast("aggregator", new HttpObjectAggregator(C5ServerConstants.MAX_CALL_SIZE));

            pipeline.addLast("encode", new ClientHttpProtostuffEncoder());
            pipeline.addLast("decode", new ClientHttpProtostuffDecoder());

            pipeline.addLast("translate", new ClientEncodeCommandRequest());
          }
        });
  }

@Override
public void initChannel(SocketChannel ch) throws Exception {
    // Create a default pipeline implementation.
    ChannelPipeline pipeline = ch.pipeline();

    if (ssl) {
        SSLEngine engine = SecureChatSslContextFactory.getClientContext().createSSLEngine();
        engine.setUseClientMode(true);
        pipeline.addLast("ssl", new SslHandler(engine));
    }

    pipeline.addLast("codec", new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    pipeline.addLast("inflater", new HttpContentDecompressor());

    // to be used since huge file transfer
    pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());

    pipeline.addLast("handler", new HttpUploadClientHandler());
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    // Create a default pipeline implementation.
    ChannelPipeline p = ch.pipeline();

    p.addLast("log", new LoggingHandler(LogLevel.INFO));
    // Enable HTTPS if necessary.
    if (ssl) {
        SSLEngine engine =
            SecureChatSslContextFactory.getClientContext().createSSLEngine();
        engine.setUseClientMode(true);

        p.addLast("ssl", new SslHandler(engine));
    }

    p.addLast("codec", new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    p.addLast("inflater", new HttpContentDecompressor());

    // Uncomment the following line if you don't want to handle HttpChunks.
    //p.addLast("aggregator", new HttpObjectAggregator(1048576));

    p.addLast("handler", new HttpSnoopClientHandler());
}

public EventLoopGroup epollGroup() {
    EventLoopGroup epollEventLoopGroup = new EpollEventLoopGroup(0, new DefaultThreadFactory(threadPrefix));

    ChannelFactory<EpollDomainSocketChannel> factory = new ChannelFactory<EpollDomainSocketChannel>() {
        @Override
        public EpollDomainSocketChannel newChannel() {
            return configure(new EpollDomainSocketChannel());
        }
    };

    bootstrap.group(epollEventLoopGroup).channelFactory(factory).handler(new ChannelInitializer<UnixChannel>() {
        @Override
        protected void initChannel(final UnixChannel channel) throws Exception {
            channel.pipeline().addLast(new HttpClientCodec());
        }
    });
    return epollEventLoopGroup;
}

public EventLoopGroup epollGroup() {
    EventLoopGroup epollEventLoopGroup = new EpollEventLoopGroup(0, new DefaultThreadFactory(threadPrefix));

    ChannelFactory<EpollDomainSocketChannel> factory = new ChannelFactory<EpollDomainSocketChannel>() {
        @Override
        public EpollDomainSocketChannel newChannel() {
            return configure(new EpollDomainSocketChannel());
        }
    };

    bootstrap.group(epollEventLoopGroup).channelFactory(factory).handler(new ChannelInitializer<UnixChannel>() {
        @Override
        protected void initChannel(final UnixChannel channel) throws Exception {
            channel.pipeline().addLast(new HttpClientCodec());
        }
    });
    return epollEventLoopGroup;
}

/**
 * Starts the NettyPerfClient.
 * @throws InterruptedException
 */
protected void start() throws InterruptedException {
  logger.info("Starting NettyPerfClient");
  reporter.start();
  group = new NioEventLoopGroup(concurrency);
  b.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel ch) throws Exception {
      if (sslFactory != null) {
        ch.pipeline()
            .addLast("sslHandler", new SslHandler(sslFactory.createSSLEngine(host, port, SSLFactory.Mode.CLIENT)));
      }
      ch.pipeline().addLast(new HttpClientCodec()).addLast(new ChunkedWriteHandler()).addLast(new ResponseHandler());
    }
  });
  logger.info("Connecting to {}:{}", host, port);
  b.remoteAddress(host, port);
  perfClientStartTime = System.currentTimeMillis();
  for (int i = 0; i < concurrency; i++) {
    b.connect().addListener(channelConnectListener);
  }
  isRunning = true;
  logger.info("Created {} channel(s)", concurrency);
  logger.info("NettyPerfClient started");
}

/**
 * Create a NettyClient.
 * @param hostname the host to connect to.
 * @param port the port to connect to.
 * @param sslFactory the {@link SSLFactory} to use if SSL is enabled.
 */
public NettyClient(final String hostname, final int port, final SSLFactory sslFactory) throws InterruptedException {
  this.hostname = hostname;
  this.port = port;
  b.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel ch) throws Exception {
      ChannelPipeline pipeline = ch.pipeline();
      if (sslFactory != null) {
        pipeline.addLast("sslHandler",
            new SslHandler(sslFactory.createSSLEngine(hostname, port, SSLFactory.Mode.CLIENT)));
      }
      pipeline.addLast(new HttpClientCodec()).addLast(new ChunkedWriteHandler()).addLast(communicationHandler);
    }
  });
  createChannel();
}

public WebSocketClient(InetSocketAddress serverAddress, String contextPath) {
    try {
        final WebSocketClientHandler webSocketClientHandler = new WebSocketClientHandler(serverAddress, contextPath, this);

        channel = new Bootstrap().group(group)
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ch.pipeline()
                        .addLast(
                            new HttpClientCodec(),
                            new HttpObjectAggregator(Integer.MAX_VALUE),
                            webSocketClientHandler
                        );
                }
            }).connect(serverAddress).sync().channel();

    } catch (Exception e) {
        throw new WebSocketException("Exception while starting web socket client", e);
    }
}

/**
 * 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);
    }
  };
}

void createBootstrap() throws InterruptedException {
    bootstrap = new Bootstrap();
    bootstrap.group(group)
       .channel(NioSocketChannel.class)
       .handler(new ChannelInitializer<SocketChannel>() {
           @Override
           public void initChannel(SocketChannel ch) throws Exception {
               ChannelPipeline pipeline = ch.pipeline();
               pipeline.addLast("http-codec", new HttpClientCodec());
               pipeline.addLast("aggregator", new HttpObjectAggregator(8192));
               pipeline.addLast("ws-handler", handler);
           }
       });

    channel = bootstrap.connect(uri.getHost(), uri.getPort()).sync().channel();
}

/**
 * Adds pipelines to channel.
 * 
 *  @param ch channel to be operated on
 */
protected void initChannel(SocketChannel ch) throws Exception {
  ChannelPipeline pipe = ch.pipeline();

  if (ssl) {
    // HTTPs connection
    SSLEngine sslEng = getSsl(null);
    sslEng.setUseClientMode(true);
    pipe.addLast("SSL", new SslHandler(sslEng, false));
  }

  pipe.addFirst("Timer", new ReadTimeoutHandler(30));
  pipe.addLast("Codec", new HttpClientCodec());
  pipe.addLast("Inflater", new HttpContentDecompressor());
  pipe.addLast("Handler", new HTTPMessageHandler(builder));
}

@Override
public void connect() {
  checkState(channel == null, "channel already initialized");
  try {
    TrustManagerFactory trustFactory = TrustManagerFactory.getInstance(
        TrustManagerFactory.getDefaultAlgorithm());
    trustFactory.init((KeyStore) null);
    final SslContext sslContext = SslContextBuilder.forClient()
        .trustManager(trustFactory).build();
    Bootstrap bootstrap = new Bootstrap();
    final int port = uri.getPort() != -1 ? uri.getPort() : DEFAULT_WSS_PORT;
    bootstrap.group(group)
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<SocketChannel>() {
          @Override
          protected void initChannel(SocketChannel ch) {
            ChannelPipeline p = ch.pipeline();
            p.addLast(sslContext.newHandler(ch.alloc(), uri.getHost(), port));
            p.addLast(
                new HttpClientCodec(),
                // Set the max size for the HTTP responses. This only applies to the WebSocket
                // handshake response from the server.
                new HttpObjectAggregator(32 * 1024),
                channelHandler);
          }
        });

    ChannelFuture channelFuture = bootstrap.connect(uri.getHost(), port);
    this.channel = channelFuture.channel();
    channelFuture.addListener(
        new ChannelFutureListener() {
          @Override
          public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
              eventHandler.onError(future.cause());
            }
          }
        }
    );
  } catch (Exception e) {
    eventHandler.onError(e);
  }
}

@Override public void prepare(final Benchmark benchmark) {
  this.concurrencyLevel = benchmark.concurrencyLevel;
  this.targetBacklog = benchmark.targetBacklog;

  ChannelInitializer<SocketChannel> channelInitializer = new ChannelInitializer<SocketChannel>() {
    @Override public void initChannel(SocketChannel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();

      if (benchmark.tls) {
        SslClient sslClient = SslClient.localhost();
        SSLEngine engine = sslClient.sslContext.createSSLEngine();
        engine.setUseClientMode(true);
        pipeline.addLast("ssl", new SslHandler(engine));
      }

      pipeline.addLast("codec", new HttpClientCodec());
      pipeline.addLast("inflater", new HttpContentDecompressor());
      pipeline.addLast("handler", new HttpChannel(channel));
    }
  };

  bootstrap = new Bootstrap();
  bootstrap.group(new NioEventLoopGroup(concurrencyLevel))
      .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
      .channel(NioSocketChannel.class)
      .handler(channelInitializer);
}

@Override public void prepare(final Benchmark benchmark) {
  this.concurrencyLevel = benchmark.concurrencyLevel;
  this.targetBacklog = benchmark.targetBacklog;

  ChannelInitializer<SocketChannel> channelInitializer = new ChannelInitializer<SocketChannel>() {
    @Override public void initChannel(SocketChannel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();

      if (benchmark.tls) {
        SslClient sslClient = SslClient.localhost();
        SSLEngine engine = sslClient.sslContext.createSSLEngine();
        engine.setUseClientMode(true);
        pipeline.addLast("ssl", new SslHandler(engine));
      }

      pipeline.addLast("codec", new HttpClientCodec());
      pipeline.addLast("inflater", new HttpContentDecompressor());
      pipeline.addLast("handler", new HttpChannel(channel));
    }
  };

  bootstrap = new Bootstrap();
  bootstrap.group(new NioEventLoopGroup(concurrencyLevel))
      .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
      .channel(NioSocketChannel.class)
      .handler(channelInitializer);
}

@Override
protected void initChannel(Channel ch) throws Exception {
  if (proxyHandler != null) {
    ch.pipeline().addLast(proxyHandler);
  }
  if (requestProto.getSsl()) {
    ch.pipeline().addLast(
        ((HttpProxyServerHandle) clientChannel.pipeline().get("serverHandle")).getServerConfig()
            .getClientSslCtx()
            .newHandler(ch.alloc(), requestProto.getHost(), requestProto.getPort()));
  }
  ch.pipeline().addLast("httpCodec", new HttpClientCodec());
  ch.pipeline().addLast("proxyClientHandle", new HttpProxyClientHandle(clientChannel));
}

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

    ctx.pipeline()
       .addBefore(ctx.name(), null, new HttpClientCodec())
       .addBefore(ctx.name(), null, delayOutboundHandler);
}

/**
 * 通道注册的时候配置websocket解码handler
 */
@Override
protected final void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline=ch.pipeline();
    pipeline.addLast(new HttpClientCodec());
    pipeline.addLast(new ChunkedWriteHandler());
    pipeline.addLast(new HttpObjectAggregator(64*1024));
    pipeline.addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(new URI(url), WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
       pipeline.addLast(new WebSocketConnectedClientHandler());//连接成功监听handler
}