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

/**
 * 接受http信息
 * @param ctx
 * @param req
 */
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
    // Handle a bad request.
    if (!req.getDecoderResult().isSuccess()) {
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
        return;
    }

    // Handshake
    WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
            getWebSocketLocation(req), null, true);
    handshaker = wsFactory.newHandshaker(req);
    if (handshaker == null) {
        WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    } else {
        handshaker.handshake(ctx.channel(), req);
    }
}

/**
 * 返回http信息
 * @param ctx
 * @param req
 * @param res
 */
private static void sendHttpResponse(
        ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
    // Generate an error page if response getStatus code is not OK (200).
    if (res.getStatus().code() != 200) {
        ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();
        HttpHeaders.setContentLength(res, res.content().readableBytes());
    }

    // Send the response and close the connection if necessary.
    ChannelFuture f = ctx.channel().writeAndFlush(res);
    if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
        f.addListener(ChannelFutureListener.CLOSE);
    }
}

public void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
   if (res.getStatus().code() != 200) {
      ByteBuf f = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
      res.content().clear();
      res.content().writeBytes(f);
      f.release();
   }

   HttpHeaders.setContentLength(res, res.content().readableBytes());
   ChannelFuture f1;
   f1 = ctx.channel().writeAndFlush(res);

   if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
      f1.addListener(ChannelFutureListener.CLOSE);
   }
}

/**
 * HTTP握手反馈
 */
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request){
    //判断是否是WebSocket协议
    if (!request.decoderResult().isSuccess() || !"websocket".equals(request.headers().get("Upgrade"))) {
        logger.warn("protobuf don't support WebSocket");
        ctx.channel().close();
        return;
    }
    WebSocketServerHandshakerFactory handshakerFactory = new WebSocketServerHandshakerFactory(
            WEBSOCKET_URL, null, true);
    handshaker = handshakerFactory.newHandshaker(request);
    if (handshaker == null){
        WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    }else {
        // 动态加入websocket的编解码处理
        handshaker.handshake(ctx.channel(), request);
        // 存储已经连接的Channel
        manager.addChannel(ctx.channel());
    }
}

public FullHttpRequest getHttpRequest() {
  if (h1Request != null) {
    return h1Request;
  }

  if (h2Headers != null) {
    try {
      // Fake out a full HTTP request.
      FullHttpRequest synthesizedRequest =
          HttpConversionUtil.toFullHttpRequest(0, h2Headers, alloc, true);
      if (data != null) {
        synthesizedRequest.replace(data);
      }

      return synthesizedRequest;
    } catch (Http2Exception e) {
      // TODO(JR): Do something more meaningful with this exception
      e.printStackTrace();
    }
  }

  throw new IllegalStateException("Cannot get the http request for an empty XrpcRequest");
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    LOGGER.info("[Client ({})] => [Server ({})] : {}",
                connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                msg);

    if (msg instanceof FullHttpRequest) {
        String streamId = ((HttpRequest) msg).headers().get(
                HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
        if (streamId == null) {
            throw new IllegalStateException("No streamId");
        }
        streams.offer(streamId);
    } else if (msg instanceof HttpObject) {
        throw new IllegalStateException("Cannot handle message: " + msg.getClass());
    }

    outboundChannel.writeAndFlush(msg);
}

public void testReleaseOnSendToClosedChannel() {
    final Settings settings = Settings.builder().build();
    final NamedXContentRegistry registry = xContentRegistry();
    try (Netty4HttpServerTransport httpServerTransport =
                 new Netty4HttpServerTransport(settings, networkService, bigArrays, threadPool, registry, new NullDispatcher())) {
        final FullHttpRequest httpRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
        final EmbeddedChannel embeddedChannel = new EmbeddedChannel();
        final Netty4HttpRequest request = new Netty4HttpRequest(registry, httpRequest, embeddedChannel);
        final HttpPipelinedRequest pipelinedRequest = randomBoolean() ? new HttpPipelinedRequest(request.request(), 1) : null;
        final Netty4HttpChannel channel =
                new Netty4HttpChannel(httpServerTransport, request, pipelinedRequest, randomBoolean(), threadPool.getThreadContext());
        final TestResponse response = new TestResponse(bigArrays);
        assertThat(response.content(), instanceOf(Releasable.class));
        embeddedChannel.close();
        channel.sendResponse(response);
        // ESTestCase#after will invoke ensureAllArraysAreReleased which will fail if the response content was not released
    }
}

private FullHttpResponse executeRequest(final Settings settings, final String originValue, final String host) {
    // construct request and send it over the transport layer
    try (Netty4HttpServerTransport httpServerTransport =
                 new Netty4HttpServerTransport(settings, networkService, bigArrays, threadPool, xContentRegistry(),
                         new NullDispatcher())) {
        httpServerTransport.start();
        final FullHttpRequest httpRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
        if (originValue != null) {
            httpRequest.headers().add(HttpHeaderNames.ORIGIN, originValue);
        }
        httpRequest.headers().add(HttpHeaderNames.HOST, host);
        final WriteCapturingChannel writeCapturingChannel = new WriteCapturingChannel();
        final Netty4HttpRequest request = new Netty4HttpRequest(xContentRegistry(), httpRequest, writeCapturingChannel);

        Netty4HttpChannel channel =
                new Netty4HttpChannel(httpServerTransport, request, null, randomBoolean(), threadPool.getThreadContext());
        channel.sendResponse(new TestResponse());

        // get the response
        List<Object> writtenObjects = writeCapturingChannel.getWrittenObjects();
        assertThat(writtenObjects.size(), is(1));
        return (FullHttpResponse) writtenObjects.get(0);
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof FullHttpRequest) {
        FullHttpRequest request = (FullHttpRequest) msg;
        Object response = null;
        if (staticFilePattern.matcher(request.uri()).find()) {
            super.channelRead(ctx, msg);
            return;
        }
        try {
            response = dispatcher.doDispatcher(request);
        } catch (Exception ex) {
            // TODO: 异常处理
            ex.printStackTrace();
        }
        ObjectMapper om = new ObjectMapper();
        String jsonStr = om.writer().writeValueAsString(response);


        sendResponse(ctx, request, jsonStr);
    }

}

/**
 * 请求分发与处理
 *
 * @param request http协议请求
 * @return 处理结果
 * @throws InvocationTargetException 调用异常
 * @throws IllegalAccessException    参数异常
 */
public Object doDispatcher(FullHttpRequest request) throws InvocationTargetException, IllegalAccessException {
    Object[] args;
    String uri = request.uri();
    if (uri.endsWith("favicon.ico")) {
        return "";
    }

    AceServiceBean aceServiceBean = Context.getAceServiceBean(uri);
    AceHttpMethod aceHttpMethod = AceHttpMethod.getAceHttpMethod(request.method().toString());
    ByteBuf content = request.content();
    //如果要多次解析，请用 request.content().copy()
    QueryStringDecoder decoder = new QueryStringDecoder(uri);
    Map<String, List<String>> requestMap = decoder.parameters();
    Object result = aceServiceBean.exec(uri, aceHttpMethod, requestMap, content == null ? null : content.toString(CharsetUtil.UTF_8));
    String contentType = request.headers().get("Content-Type");
    if (result == null) {
        ApplicationInfo mock = new ApplicationInfo();
        mock.setName("ace");
        mock.setVersion("1.0");
        mock.setDesc(" mock  !!! ");
        result = mock;
    }
    return result;

}

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req)
        throws Exception {
    if (!req.getDecoderResult().isSuccess()) {
        logger.debug("invalid http request");
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                BAD_REQUEST));
        return;
    }

    if (req.getUri().equalsIgnoreCase(this.websocketUri)) {
        logger.debug("it is websocket request");
        ctx.fireChannelRead(req.retain());
        return;
    }

    HttpTransport transport = getTransport(req);
    if (transport == null) {
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                BAD_REQUEST));
    } else {
        transport.handleRequest(ctx, req);
    }
}

@Override
public void handleRequest(ChannelHandlerContext ctx, FullHttpRequest req)
        throws Exception {
    if (req.getUri().contains("/jsonp/connect")) {
        handleConnect(ctx, req);
    } else if (req.getUri().contains("/jsonp/subscribe")) {
        handleSubscrible(ctx, req);
    } else if (req.getUri().contains("/jsonp/waiting")) {
        handleWaitingMsg(ctx, req);
    } else if (req.getUri().contains("/jsonp/unsubscrible")) {
        handleUnsubscrible(ctx, req);
    } else if (req.getUri().contains("/jsonp/publish")) {
        handlePublish(ctx, req);
    } else if (req.getUri().contains("/jsonp/disconnect")) {
        handleDisconnect(ctx, req);
    } else { // invalid request
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                BAD_REQUEST));
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) {  
    StringUtils.Pair url = HttpUtils.parseUriIntoUrlBaseAndPath(msg.uri());
    HttpRequest request = new HttpRequest();
    if (url.left == null) {
        String requestScheme = provider.isSsl() ? "https" : "http";
        String host = msg.headers().get(HttpUtils.HEADER_HOST);
        request.setUrlBase(requestScheme + "://" + host);
    } else {
        request.setUrlBase(url.left);            
    }                                
    request.setUri(url.right);
    request.setMethod(msg.method().name());
    msg.headers().forEach(h -> request.addHeader(h.getKey(), h.getValue()));
    QueryStringDecoder decoder = new QueryStringDecoder(url.right);                
    decoder.parameters().forEach((k, v) -> request.putParam(k, v));
    HttpContent httpContent = (HttpContent) msg;
    ByteBuf content = httpContent.content();
    if (content.isReadable()) {
        byte[] bytes = new byte[content.readableBytes()];
        content.readBytes(bytes);
        request.setBody(bytes);
    }
    writeResponse(request, ctx);
    ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
}

private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
    if (!request.decoderResult().isSuccess() || !"websocket".equals(request.headers().get("Upgrade"))) {
        logger.warn("protobuf don't support websocket");
        ctx.channel().close();
        return;
    }
    WebSocketServerHandshakerFactory handshakerFactory = new WebSocketServerHandshakerFactory(
            Constants.WEBSOCKET_URL, null, true);
    handshaker = handshakerFactory.newHandshaker(request);
    if (handshaker == null) {
        WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    } else {
        // 动态加入websocket的编解码处理
        handshaker.handshake(ctx.channel(), request);
        UserInfo userInfo = new UserInfo();
        userInfo.setAddr(NettyUtil.parseChannelRemoteAddr(ctx.channel()));
        // 存储已经连接的Channel
        UserInfoManager.addChannel(ctx.channel());
    }
}

public static String getSessionId(FullHttpRequest msg, boolean anonymousAccessAllowed) {
    final StringBuilder buf = new StringBuilder();
    msg.headers().getAll(Names.COOKIE).forEach(h -> {
        ServerCookieDecoder.STRICT.decode(h).forEach(c -> {
            if (c.name().equals(Constants.COOKIE_NAME)) {
                if (buf.length() == 0) {
                    buf.append(c.value());
                }
            }
        });
    });
    String sessionId = buf.toString();
    if (sessionId.length() == 0 && anonymousAccessAllowed) {
        sessionId = NO_AUTHORIZATIONS;
    } else if (sessionId.length() == 0) {
        sessionId = null;
    }
    return sessionId;
}

private void sendTextResource(String prepend, String name, String mimeType, FullHttpRequest req, ChannelHandlerContext ctx) throws IOException {
    BufferedReader reader = new BufferedReader(new InputStreamReader((this.getResourceAsStream(name))));
    // TODO: read only once and buffer
    String line;
    StringBuffer buffer = new StringBuffer();
    if (prepend != null) buffer.append(prepend);
    while ((line = reader.readLine()) != null) {
        buffer.append(line);
        buffer.append('\n');
    }
    ByteBuf content = Unpooled.copiedBuffer(buffer, java.nio.charset.Charset.forName("UTF-8"));

    FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, OK, content);

    res.headers().set(HttpHeaderNames.CONTENT_TYPE, mimeType);
    HttpUtil.setContentLength(res, content.readableBytes());

    sendHttpResponse(ctx, req, res);
}

@Override
public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
    if (HttpUtil.is100ContinueExpected(req)) {
        ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE));
    }
    boolean keepAlive = HttpUtil.isKeepAlive(req);

    ByteBuf content = ctx.alloc().buffer();
    content.writeBytes(HelloWorldHttp2Handler.RESPONSE_BYTES.duplicate());
    ByteBufUtil.writeAscii(content, " - via " + req.protocolVersion() + " (" + establishApproach + ")");

    FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, content);
    response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
    response.headers().setInt(CONTENT_LENGTH, response.content().readableBytes());

    if (!keepAlive) {
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    } else {
        response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        ctx.writeAndFlush(response);
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
    QueryStringDecoder queryString = new QueryStringDecoder(request.uri());
    String streamId = streamId(request);
    int latency = toInt(firstValue(queryString, LATENCY_FIELD_NAME), 0);
    if (latency < MIN_LATENCY || latency > MAX_LATENCY) {
        sendBadRequest(ctx, streamId);
        return;
    }
    String x = firstValue(queryString, IMAGE_COORDINATE_X);
    String y = firstValue(queryString, IMAGE_COORDINATE_Y);
    if (x == null || y == null) {
        handlePage(ctx, streamId, latency, request);
    } else {
        handleImage(x, y, ctx, streamId, latency, request);
    }
}

protected HttpResponse handleNonProxyRequest(FullHttpRequest req) {
    String uri = req.getUri();
    if ("/version".equals(uri)) {
        if (HttpMethod.GET.equals(req.getMethod())) {
            JsonObject jsonObj = new JsonObject();
            jsonObj.addProperty("name", m_appConfig.getAppName());
            jsonObj.addProperty("version", m_appConfig.getAppVersion());
            byte[] content = jsonObj.toString().getBytes(CharsetUtil.UTF_8);
            DefaultFullHttpResponse resp = new DefaultFullHttpResponse(
                    HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
                    Unpooled.copiedBuffer(content));
            HttpHeaders.setKeepAlive(resp, false);
            HttpHeaders.setHeader(resp, HttpHeaders.Names.CONTENT_TYPE,
                    "application/json");
            HttpHeaders.setContentLength(resp, content.length);
            return resp;
        }
    }

    return RESPONSE_404;
}

public static OneM2mRequest decode(FullHttpRequest request, String remoteHost) throws Exception {

    Byte content[];
    String method;
    HashMap<String, String> headerMap = new HashMap<String, String>();
    HttpHeaders headers = request.headers();
    Iterator<Entry<String, String>> it = headers.iterator();
    while (it.hasNext()) {
        Entry<String, String> header = it.next();
        headerMap.put(header.getKey().toUpperCase(), header.getValue());
    }

    if(request.content().isReadable()) {
        return decode(request.getMethod().name(), request.getUri(), headerMap, remoteHost, request.content().copy().array());
    } else {
        return decode(request.getMethod().name(), request.getUri(), headerMap, remoteHost, null);               
    }
}

@Test
public void testSuccess_badResponse() {
  FullHttpRequest request = makeHttpPostRequest("some request", HOST, "/");
  FullHttpResponse response =
      makeHttpResponse("some bad response", HttpResponseStatus.BAD_REQUEST);
  // outbound message passed to the next handler.
  assertThat(channel.writeOutbound(request)).isTrue();
  assertHttpRequestEquivalent(request, channel.readOutbound());
  fakeClock.advanceOneMilli();
  // inbound message passed to the next handler.
  // Even though the response status is not OK, the metrics handler only logs it and pass it
  // along to the next handler, which handles it.
  assertThat(channel.writeInbound(response)).isTrue();
  assertHttpResponseEquivalent(response, channel.readInbound());

  verify(metrics).requestSent(RELAYED_PROTOCOL_NAME, CLIENT_CERT_HASH, request);
  verify(metrics).responseReceived(RELAYED_PROTOCOL_NAME, CLIENT_CERT_HASH, response, 1);
  verifyNoMoreInteractions(metrics);
}

@Test
@SuppressWarnings("unchecked")
public void testIssueDebugRequest_fromDebugFilter() {
    Mockito.when(m_appConfiguration.getBoolean("debugManager.debugEnabled")).thenReturn(true);
    FullHttpRequest request = Mockito.mock(FullHttpRequest.class);
    FullHttpResponse cacheResponse = Mockito.mock(FullHttpResponse.class);
    ChannelHandlerContext ctx = Mockito.mock(ChannelHandlerContext.class);
    Mockito.when(request.copy()).thenReturn(request);
    Mockito.when(m_cacheManager.get("test_req")).thenReturn(cacheResponse);
    Mockito.when(m_policyManager.generateCacheKey(request)).thenReturn("test_req");

    Attribute<CacheResultVerifier> debugging = Mockito.mock(Attribute.class);
    Mockito.when(ctx.attr(DebugManager.DEBUG_RESULT)).thenReturn(debugging);
    debugManager.issueDebugRequest(request, ctx, true);
    CacheResultVerifier verifier = new CacheResultVerifier("test_req", request, cacheResponse);
    Mockito.verify(debugging, Mockito.times(1)).set(Mockito.refEq(verifier));
}

@Test
@SuppressWarnings("unchecked")
public void testIssueDebugRequest_sslFromDebugFilter() {
    Mockito.when(m_appConfiguration.getBoolean("debugManager.debugEnabled")).thenReturn(true);
    FullHttpRequest request = Mockito.mock(FullHttpRequest.class);
    FullHttpResponse cacheResponse = Mockito.mock(FullHttpResponse.class);
    ChannelHandlerContext ctx = Mockito.mock(ChannelHandlerContext.class);
    Mockito.when(ctx.handler()).thenReturn(clientToProxyConnection);
    Mockito.when(request.copy()).thenReturn(request);
    String key = "https://serverHostAndPort=www.ebay.com:443";
    Mockito.when(m_cacheManager.get(key)).thenReturn(cacheResponse);
    Mockito.when(m_policyManager.generateCacheKey(request)).thenReturn(key);

    Attribute<CacheResultVerifier> debugging = Mockito.mock(Attribute.class);
    Mockito.when(ctx.attr(DebugManager.DEBUG_RESULT)).thenReturn(debugging);
    debugManager.issueDebugRequest(request, ctx, true);
    Assert.assertTrue((Boolean) readField(clientToProxyConnection, "mitming"));
    CacheResultVerifier verifier = new CacheResultVerifier(key, request, cacheResponse);
    Mockito.verify(debugging, Mockito.times(1)).set(Mockito.refEq(verifier));
}

@Before
public void setup() {
    m_config = Mockito.mock(Config.class);
    Mockito.when(m_config.asMap()).thenReturn(m_configMap);

    m_appConfiguration = Mockito.mock(AppConfiguration.class);
    Mockito.when(m_appConfiguration.getConfig()).thenReturn(m_config);

    m_keyGen = Mockito.mock(IKeyGenerator.class);
    policyManager = new PolicyManager<FullHttpRequest, FullHttpResponse, CacheResponse>(m_appConfiguration, m_keyGen, null) {
        @Override
        protected CacheManager<FullHttpRequest, FullHttpResponse, CacheResponse> initCacheManager() {
            return Mockito.mock(CacheManager.class);
        }
    };
}

/**
 * 获取请求参数 Map
 */
private Map<String, List<String>> getParamMap(){
    Map<String, List<String>> paramMap = new HashMap<String, List<String>>();

    Object msg = DataHolder.getRequest();
    HttpRequest request = (HttpRequest) msg;
    HttpMethod method = request.method();
    if(method.equals(HttpMethod.GET)){
        String uri = request.uri();
        QueryStringDecoder queryDecoder = new QueryStringDecoder(uri, Charset.forName(CharEncoding.UTF_8));
        paramMap = queryDecoder.parameters();

    }else if(method.equals(HttpMethod.POST)){
        FullHttpRequest fullRequest = (FullHttpRequest) msg;
        paramMap = getPostParamMap(fullRequest);
    }

    return paramMap;
}

@Test
public void testFilterResponse() {
    List<IHttpResponseProxyFilter<FullHttpResponse, CacheResponse>> respFilters = new ArrayList<>();
    FilterManager<FullHttpRequest, FullHttpResponse, CacheResponse> fm = new FilterManager<>(
            null, respFilters);
    assertNull(fm.filterResponse(null, null));

    FullHttpResponse resp = mock(FullHttpResponse.class);
    assertNotNull(fm.filterResponse(resp, null));

    IHttpResponseProxyFilter f1 = mock(IHttpResponseProxyFilter.class);
    respFilters.add(f1);
    FullHttpResponse changedResp = mock(FullHttpResponse.class);
    when(f1.filterResponse(resp, null)).thenReturn(changedResp);
    assertEquals(changedResp, fm.filterResponse(resp, null));

}

@Test
public void test() throws InterruptedException, ExecutionException {
  int streamId = 3;
  FullHttpRequest request =
      new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
  request.headers().add(HttpUtil.ExtensionHeaderNames.STREAM_ID.text(),
    streamId);
  Promise<FullHttpResponse> promise = CHANNEL.eventLoop().newPromise();
  synchronized (RESPONSE_HANDLER) {
    CHANNEL.writeAndFlush(request);
    RESPONSE_HANDLER.put(streamId, promise);
  }
  assertEquals(HttpResponseStatus.OK, promise.get().status());
  ByteBuf content = promise.get().content();
  assertEquals("HTTP/2 DTP", content.toString(StandardCharsets.UTF_8));
}

private void handleHttpRequest(ChannelHandlerContext ctx,
        FullHttpRequest req) {
    if (!req.getDecoderResult().isSuccess()
            || (!"websocket".equals(req.headers().get("Upgrade")))) {
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
        return;
    }
    WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
            "ws://localhost:7777/websocket", null, false);
    socketServerHandshaker = wsFactory.newHandshaker(req);
    if (socketServerHandshaker == null) {
        WebSocketServerHandshakerFactory
                .sendUnsupportedWebSocketVersionResponse(ctx.channel());
    } else {
        socketServerHandshaker.handshake(ctx.channel(), req);
    }
}

@Nullable
@Override
public String execute(@NotNull QueryStringDecoder urlDecoder, @NotNull FullHttpRequest request, @NotNull ChannelHandlerContext context) throws IOException {
  final JsonReader reader = createJsonReader(request);
  reader.beginObject();
  final String name = reader.nextName();
  final String url = reader.nextString();
  reader.endObject();

  if (URL_PARAM_NAME.equals(name) && url != null && url.startsWith(JetBrainsProtocolHandler.PROTOCOL)) {
    JetBrainsProtocolHandler.processJetBrainsLauncherParameters(url);
    ApplicationManager.getApplication().invokeLater(new Runnable() {
      @Override
      public void run() {
        JBProtocolCommand.handleCurrentCommand();
      }
    }, ModalityState.any());
  }

  sendOk(request, context);
  return null;
}

@Test
public void testSuccess_oneRequest() {
  String content = "some content";
  FullHttpRequest request = makeHttpPostRequest(content, host, "/");
  metrics.requestSent(protocol, certHash, request);

  assertThat(BackendMetrics.requestsCounter)
      .hasValueForLabels(1, protocol, certHash)
      .and()
      .hasNoOtherValues();
  assertThat(BackendMetrics.requestBytes)
      .hasDataSetForLabels(ImmutableSet.of(content.length()), protocol, certHash)
      .and()
      .hasNoOtherValues();
  assertThat(BackendMetrics.responsesCounter).hasNoOtherValues();
  assertThat(BackendMetrics.responseBytes).hasNoOtherValues();
  assertThat(BackendMetrics.latencyMs).hasNoOtherValues();
}

@Override
protected ListenableFuture<ClientHttpResponse> executeInternal(final HttpHeaders headers) throws IOException {
    final SettableListenableFuture<ClientHttpResponse> responseFuture =
            new SettableListenableFuture<ClientHttpResponse>();

    ChannelFutureListener connectionListener = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                Channel channel = future.channel();
                channel.pipeline().addLast(new RequestExecuteHandler(responseFuture));
                FullHttpRequest nettyRequest = createFullHttpRequest(headers);
                channel.writeAndFlush(nettyRequest);
            }
            else {
                responseFuture.setException(future.cause());
            }
        }
    };

    this.bootstrap.connect(this.uri.getHost(), getPort(this.uri)).addListener(connectionListener);

    return responseFuture;
}

private FullHttpRequest createFullHttpRequest(HttpHeaders headers) {
    io.netty.handler.codec.http.HttpMethod nettyMethod =
            io.netty.handler.codec.http.HttpMethod.valueOf(this.method.name());

    FullHttpRequest nettyRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
            nettyMethod, this.uri.toString(), this.body.buffer());

    nettyRequest.headers().set(HttpHeaders.HOST, uri.getHost());
    nettyRequest.headers().set(HttpHeaders.CONNECTION, io.netty.handler.codec.http.HttpHeaders.Values.CLOSE);

    for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
        nettyRequest.headers().add(entry.getKey(), entry.getValue());
    }

    return nettyRequest;
}

/**
 * Tests that the client converts given {@link FullHttpRequest} to bytes, which is sent to the
 * server and reconstructed to a {@link FullHttpRequest} that is equivalent to the original. Then
 * test that the server converts given {@link FullHttpResponse} to bytes, which is sent to the
 * client and reconstructed to a {@link FullHttpResponse} that is equivalent to the original.
 *
 * <p>The request and response equivalences are tested in the same method because the client codec
 * tries to pair the response it receives with the request it sends. Receiving a response without
 * sending a request first will cause the {@link HttpObjectAggregator} to fail to aggregate
 * properly.
 */
private void requestAndRespondWithStatus(HttpResponseStatus status) {
  ByteBuf buffer;
  FullHttpRequest requestSent = makeHttpPostRequest(CONTENT, HOST, PATH);
  // Need to send a copy as the content read index will advance after the request is written to
  // the outbound of client channel, making comparison with requestReceived fail.
  assertThat(channel.writeOutbound(requestSent.copy())).isTrue();
  buffer = channel.readOutbound();
  assertThat(serverChannel.writeInbound(buffer)).isTrue();
  FullHttpRequest requestReceived = serverChannel.readInbound();
  // Verify that the request received is the same as the request sent.
  assertHttpRequestEquivalent(requestSent, requestReceived);

  FullHttpResponse responseSent = makeHttpResponse(CONTENT, status);
  assertThat(serverChannel.writeOutbound(responseSent.copy())).isTrue();
  buffer = serverChannel.readOutbound();
  assertThat(channel.writeInbound(buffer)).isTrue();
  FullHttpResponse responseReceived = channel.readInbound();
  // Verify that the request received is the same as the request sent.
  assertHttpResponseEquivalent(responseSent, responseReceived);
}

private static void sendHttpResponse(
        ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
    // Generate an error page if response getStatus code is not OK (200).
    if (res.status().code() != 200) {
        ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();
        HttpUtil.setContentLength(res, res.content().readableBytes());
    }

    // Send the response and close the connection if necessary.
    ChannelFuture f = ctx.channel().writeAndFlush(res);
    if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
        f.addListener(ChannelFutureListener.CLOSE);
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {

    ByteBuf buf = msg.content();
    byte[] bytes = new byte[buf.readableBytes()];
    buf.getBytes(0, bytes);

    YarRequest yarRequest = YarProtocol.buildRequest(bytes);
    YarResponse yarResponse = process(yarRequest);

    FullHttpResponse response =
            new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(YarProtocol
                    .toProtocolBytes(yarResponse)));
    response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "application/x-www-form-urlencoded");
    response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, response.content().readableBytes());
    if (HttpHeaders.isKeepAlive(msg)) {
        response.headers().set(HttpHeaders.Names.CONNECTION, Values.KEEP_ALIVE);
    }
    ctx.write(response);
    ctx.flush();
    ctx.close();
}

@Test
public void doChannelRead_creates_and_sets_RequestInfo_on_state_and_RequestInfo_is_marked_as_complete_with_all_chunks_if_msg_is_FullHttpRequest() {
    // given
    FullHttpRequest msgMock = mock(FullHttpRequest.class);
    String uri = "/some/url";
    HttpHeaders headers = new DefaultHttpHeaders();
    doReturn(uri).when(msgMock).getUri();
    doReturn(headers).when(msgMock).headers();
    doReturn(headers).when(msgMock).trailingHeaders();
    doReturn(byteBufMock).when(msgMock).content();
    doReturn(false).when(byteBufMock).isReadable();
    doReturn(HttpVersion.HTTP_1_1).when(msgMock).getProtocolVersion();

    // when
    PipelineContinuationBehavior result = handler.doChannelRead(ctxMock, msgMock);

    // then
    ArgumentCaptor<RequestInfo> requestInfoArgumentCaptor = ArgumentCaptor.forClass(RequestInfo.class);
    verify(stateMock).setRequestInfo(requestInfoArgumentCaptor.capture());
    RequestInfo requestInfo = requestInfoArgumentCaptor.getValue();
    assertThat(requestInfo.getUri()).isEqualTo(uri);
    assertThat(requestInfo.isCompleteRequestWithAllChunks()).isTrue();
    assertThat(result).isEqualTo(PipelineContinuationBehavior.CONTINUE);
}

@Before
public void beforeMethod() {
    stateMock = mock(HttpProcessingState.class);
    ctxMock = mock(ChannelHandlerContext.class);
    channelMock = mock(Channel.class);
    pipelineMock = mock(ChannelPipeline.class);
    stateAttrMock = mock(Attribute.class);
    proxyRouterProcessingStateAttrMock = mock(Attribute.class);
    metricsListenerMock = mock(MetricsListener.class);
    msgMockFirstChunkOnly = mock(HttpRequest.class);
    msgMockFullRequest = mock(FullHttpRequest.class);
    msgMockLastChunkOnly = mock(LastHttpContent.class);
    idleChannelTimeoutHandlerMock = mock(IdleChannelTimeoutHandler.class);

    doReturn(channelMock).when(ctxMock).channel();
    doReturn(pipelineMock).when(ctxMock).pipeline();
    doReturn(idleChannelTimeoutHandlerMock).when(pipelineMock).get(IDLE_CHANNEL_TIMEOUT_HANDLER_NAME);
    doReturn(stateAttrMock).when(channelMock).attr(ChannelAttributes.HTTP_PROCESSING_STATE_ATTRIBUTE_KEY);
    doReturn(stateMock).when(stateAttrMock).get();
    doReturn(proxyRouterProcessingStateAttrMock).when(channelMock).attr(ChannelAttributes.PROXY_ROUTER_PROCESSING_STATE_ATTRIBUTE_KEY);

    handler = new RequestStateCleanerHandler(metricsListenerMock, incompleteHttpCallTimeoutMillis);
}

@Override
protected void sendResponse(final ChannelHandlerContext ctx, String streamId, int latency,
        final FullHttpResponse response, final FullHttpRequest request) {
    HttpUtil.setContentLength(response, response.content().readableBytes());
    ctx.executor().schedule(new Runnable() {
        @Override
        public void run() {
            if (isKeepAlive(request)) {
                response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                ctx.writeAndFlush(response);
            } else {
                ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
            }
        }
    }, latency, TimeUnit.MILLISECONDS);
}

public static NettyHttpClientResponse executeRequest(
    FullHttpRequest request, int port, long incompleteCallTimeoutMillis, Consumer<ChannelPipeline> pipelineAdjuster
) throws InterruptedException, TimeoutException, ExecutionException {
    Bootstrap bootstrap = createNettyHttpClientBootstrap();
    try {
        // Connect to the proxyServer.
        Channel ch = connectNettyHttpClientToLocalServer(bootstrap, port);

        try {
            return executeNettyHttpClientCall(ch, request, incompleteCallTimeoutMillis, pipelineAdjuster);
        }
        finally {
            ch.close();
        }
    } finally {
        bootstrap.group().shutdownGracefully();
    }
}

@Test
public void testSuccess_SingleFrame_MultipleInboundMessages() throws Exception {
  // First inbound message is hello.
  channel.readInbound();

  byte[] inputBytes1 = readResourceBytes(getClass(), "testdata/login.xml").read();
  byte[] inputBytes2 = readResourceBytes(getClass(), "testdata/logout.xml").read();

  // Verify inbound messages are as expected.
  assertThat(
          channel.writeInbound(
              Unpooled.wrappedBuffer(
                  getByteBufFromContent(inputBytes1), getByteBufFromContent(inputBytes2))))
      .isTrue();
  assertThat((FullHttpRequest) channel.readInbound()).isEqualTo(makeEppHttpRequest(inputBytes1));
  assertThat((FullHttpRequest) channel.readInbound()).isEqualTo(makeEppHttpRequest(inputBytes2));

  // Nothing more to read.
  assertThat((Object) channel.readInbound()).isNull();
  assertThat(channel.isActive()).isTrue();
}