io.netty.handler.codec.http.HttpVersion # HTTP_1_1 ( ) 源码实例Demo

private void testModelsInvokeMultipart(Channel channel)
        throws InterruptedException, HttpPostRequestEncoder.ErrorDataEncoderException,
                IOException {
    result = null;
    latch = new CountDownLatch(1);
    DefaultFullHttpRequest req =
            new DefaultFullHttpRequest(
                    HttpVersion.HTTP_1_1, HttpMethod.POST, "/models/noop/invoke");

    HttpPostRequestEncoder encoder = new HttpPostRequestEncoder(req, true);
    MemoryFileUpload body =
            new MemoryFileUpload("data", "test.txt", "text/plain", null, null, 4);
    body.setContent(Unpooled.copiedBuffer("test", StandardCharsets.UTF_8));
    encoder.addBodyHttpData(body);

    channel.writeAndFlush(encoder.finalizeRequest());
    if (encoder.isChunked()) {
        channel.writeAndFlush(encoder).sync();
    }

    latch.await();

    Assert.assertEquals(result, "OK");
}
 

private boolean writeResponse(HttpObject currentObj, ChannelHandlerContext ctx) {
    // Decide whether to close the connection or not.
    boolean keepAlive = HttpUtil.isKeepAlive(request);
    // Build the response object.
    FullHttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1,
            currentObj.decoderResult().isSuccess() ? HttpResponseStatus.OK : HttpResponseStatus.BAD_REQUEST,
            Unpooled.EMPTY_BUFFER);

    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");

    if (keepAlive) {
        // Add 'Content-Length' header only for a keep-alive connection.
        response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
        // Add keep alive header as per:
        // - http://www.w3.org/Protocols/HTTP/1.1/draft-ietf-http-v11-spec-01.html#Connection
        response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
    }

    // Write the response.
    ctx.write(response);

    return keepAlive;
}
 

public void sendChunkedHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
	
       HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
       HttpUtil.setTransferEncodingChunked(response, true);
       response.headers().set(HttpHeaderNames.CONTENT_TYPE, res.headers().get(HttpHeaderNames.CONTENT_TYPE));

       if(HttpUtil.isKeepAlive(req)) {
          response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
       }

       ctx.write(response);
       ctx.write(new ChunkedStream(new ByteBufInputStream(res.content())));
       ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
       if(!HttpUtil.isKeepAlive(req)) {
          future.addListener(ChannelFutureListener.CLOSE);
       }		
       metrics.incHTTPResponseCounter(className, res.status().code());
}
 

@Override
protected void encode(ChannelHandlerContext ctx, Response msg, List<Object> out) throws Exception {

    String content = msg.getContent();
    int byteBufLen = 0;
    if (content != null && content.length() > 0) {
        byteBufLen = content.length();
    }
    ByteBuf buf;
    if (byteBufLen > 0) {
        buf = ctx.alloc().buffer(byteBufLen);
        buf.writeBytes(content.getBytes());
    } else {
        buf = Unpooled.EMPTY_BUFFER;
    }

    DefaultFullHttpResponse httpResponse
            = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, msg.getStatus(), buf);
    httpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, String.format("%s;charset=%s",msg.getContentType(),msg.getCharset()));
    httpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, httpResponse.content().toString());
    if(msg.isKeepAlive()) {
        httpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
    }
    out.add(httpResponse);
}
 

@Test(
        alwaysRun = true,
        dependsOnMethods = {"testPredictionsJson"})
public void testInvocationsJson() throws InterruptedException {
    Channel channel = TestUtils.getInferenceChannel(configManager);
    TestUtils.setResult(null);
    TestUtils.setLatch(new CountDownLatch(1));
    DefaultFullHttpRequest req =
            new DefaultFullHttpRequest(
                    HttpVersion.HTTP_1_1, HttpMethod.POST, "/invocations?model_name=noop");
    req.content().writeCharSequence("{\"data\": \"test\"}", CharsetUtil.UTF_8);
    HttpUtil.setContentLength(req, req.content().readableBytes());
    req.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
    channel.writeAndFlush(req);
    TestUtils.getLatch().await();

    Assert.assertEquals(TestUtils.getResult(), "OK");
}
 

/**
 * Creates a {@link NettyRequest} with the given parameters.
 * @param httpMethod the {@link HttpMethod} desired.
 * @param uri the URI desired.
 * @param headers {@link HttpHeaders} that need to be a part of the request.
 * @param channel the {@link Channel} that the request arrived over.
 * @return {@link NettyRequest} encapsulating a {@link HttpRequest} with the given parameters.
 * @throws RestServiceException if the {@code httpMethod} is not recognized by {@link NettyRequest}.
 */
private NettyRequest createNettyRequest(HttpMethod httpMethod, String uri, HttpHeaders headers, Channel channel)
    throws RestServiceException {
  MetricRegistry metricRegistry = new MetricRegistry();
  RestRequestMetricsTracker.setDefaults(metricRegistry);
  HttpRequest httpRequest = new DefaultHttpRequest(HttpVersion.HTTP_1_1, httpMethod, uri, false);
  if (headers != null) {
    httpRequest.headers().set(headers);
  }
  NettyRequest nettyRequest =
      new NettyRequest(httpRequest, channel, new NettyMetrics(metricRegistry), BLACKLISTED_QUERY_PARAM_SET);
  assertEquals("Auto-read is in an invalid state",
      (!httpMethod.equals(HttpMethod.POST) && !httpMethod.equals(HttpMethod.PUT))
          || NettyRequest.bufferWatermark <= 0, channel.config().isAutoRead());
  return nettyRequest;
}
 

@Test(expected = JsonMappingException.class)
public void testLookupPostWithNoSession() throws Exception {
    decoder = new TestHttpQueryDecoder(config);
    DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST,
            "/api/search/lookup");
    decoder.decode(null, request, results);
}
 

private void testInvalidModelsMethod() throws InterruptedException {
    Channel channel = connect(true);
    Assert.assertNotNull(channel);

    HttpRequest req =
            new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "/models");
    channel.writeAndFlush(req).sync();
    channel.closeFuture().sync();

    ErrorResponse resp = JsonUtils.GSON.fromJson(result, ErrorResponse.class);

    Assert.assertEquals(resp.getCode(), HttpResponseStatus.METHOD_NOT_ALLOWED.code());
    Assert.assertEquals(resp.getMessage(), ERROR_METHOD_NOT_ALLOWED);
}
 

@Test
public void testMultipartRequestWithFileInvalidCharset() throws Exception {
    final String boundary = "dLV9Wyq26L_-JQxk6ferf-RT153LhOO";
    final DefaultFullHttpRequest req = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST,
        "http://localhost");
    req.headers().add(HttpHeaderNames.CONTENT_TYPE, "multipart/form-data; boundary=" + boundary);
    // Force to use memory-based data.
    final DefaultHttpDataFactory inMemoryFactory = new DefaultHttpDataFactory(false);
    final String data = "asdf";
    final String filename = "tmp;0.txt";
    final String body =
        "--" + boundary + "\r\n" +
            "Content-Disposition: form-data; name=\"file\"; filename=\"" + filename + "\"\r\n" +
            "Content-Type: image/gif; charset=ABCD\r\n" +
            "\r\n" +
            data + "\r\n" +
            "--" + boundary + "--\r\n";

    req.content().writeBytes(body.getBytes(CharsetUtil.UTF_8));
    // Create decoder instance to test.
    try {
        new HttpPostRequestDecoder(inMemoryFactory, req);
        fail("Was expecting an ErrorDataDecoderException");
    } catch (HttpPostRequestDecoder.ErrorDataDecoderException e) {
        assertTrue(e.getCause() instanceof UnsupportedCharsetException);
    } finally {
        req.release();
    }
}
 

private void handleHttpRequest(ChannelHandlerContext ctx, HttpRequest msg) throws UnsupportedEncodingException {
    String uri = msg.uri();
    QueryStringDecoder decoder = new QueryStringDecoder(uri);
    logger.debug("Got Request for " + uri);
    HttpResponseStatus status = fail ? HttpResponseStatus.BAD_REQUEST : HttpResponseStatus.OK;

    ByteArrayOutputStream out = new ByteArrayOutputStream();
    try {
        JsonGenerator generator = jsonFactory.createGenerator(out, JsonEncoding.UTF8);
        generator.writeStartObject();
        for (Map.Entry<String, List<String>> entry : decoder.parameters().entrySet()) {
            if (entry.getValue().size() == 1) {
                generator.writeStringField(entry.getKey(), URLDecoder.decode(entry.getValue().get(0), "UTF-8"));
            } else {
                generator.writeArrayFieldStart(entry.getKey());
                for (String value : entry.getValue()) {
                    generator.writeString(URLDecoder.decode(value, "UTF-8"));
                }
                generator.writeEndArray();
            }
        }
        generator.writeEndObject();
        generator.close();

    } catch (Exception ex) {
        status = HttpResponseStatus.INTERNAL_SERVER_ERROR;
    }
    ByteBuf byteBuf = Unpooled.wrappedBuffer(out.toByteArray());
    responses.add(out.toString(StandardCharsets.UTF_8.name()));

    DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, byteBuf);
    ChannelFuture future = ctx.channel().writeAndFlush(response);
    future.addListener(ChannelFutureListener.CLOSE);
}
 

@Test
public void testIOsRedirect() throws Exception {
	FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/ios/run");
	replay();
	
	FullHttpResponse response = handler.respond(request, mockContext());
	assertRedirectTo(response, config.getAppleStoreUrl());
}
 

@Test
public void testUnauthenticatedWebLaunchWithExtraPathAndQueryParams() throws Exception {
	FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/web/launch/extra/context/here?random=text&param=assert");
	expectGetUnauthenticatedClient();
	replay();
	
	FullHttpResponse response = handler.respond(request, mockContext());
	assertRedirectTo(response, config.getWebUrl() + "/extra/context/here?random=text&param=assert");
}
 

@Test(expected = IllegalArgumentException.class)
public void testSuggestURIWithInvalidTypeAnonAccess() throws Exception {
    decoder = new TestHttpQueryDecoder(anonConfig);
    DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET,
            "/api/suggest?type=foo");
    decoder.decode(null, request, results);
    Assert.assertEquals(1, results.size());
    Assert.assertEquals(SuggestRequest.class, results.iterator().next().getClass());
    SuggestRequest suggest = (SuggestRequest) results.iterator().next();
    Assert.assertEquals("foo", suggest.getType());
    Assert.assertFalse(suggest.getQuery().isPresent());
    Assert.assertEquals(25, suggest.getMax());
}
 

private FullHttpRequest createRequest(HttpMethod method, String uri, String requestBody) {
    DefaultFullHttpRequest rawRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, method, uri);
    rawRequest.headers().set("tenantId", TENANT);
    if (!requestBody.equals(""))
        rawRequest.content().writeBytes(Unpooled.copiedBuffer(requestBody.getBytes()));
    return HttpRequestWithDecodedQueryParams.create(rawRequest);
}
 

@Override
public HttpResponse marshallResponseHeaders(final boolean rawStream) {
  final HttpResponse res = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
  res.headers().add(CONNECTION, KEEP_ALIVE);
  res.headers().add(CONTENT_TYPE.toLowerCase(), "text/event-stream");
  res.headers().add(TRANSFER_ENCODING, "UTF-8");
  res.headers().add(CONTENT_ENCODING, "UTF-8");
  res.headers().add(ACCESS_CONTROL_ALLOW_ORIGIN, "*");
  return res;
}
 

public static FullHttpResponse createResponse(HttpResponseStatus status, String body) {
    byte[] bytes = FileUtils.toBytes(body);
    ByteBuf bodyBuf = Unpooled.copiedBuffer(bytes);
    DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, bodyBuf);
    response.headers().set(HttpHeaderNames.CONTENT_LENGTH, bytes.length);
    return response;
}
 

@Test
public void shouldHandleRequestsAndResponses() {
    inboundChannel.pipeline().addLast(handler);

    DefaultFullHttpRequest req = new DefaultFullHttpRequest(
            HttpVersion.HTTP_1_1, HttpMethod.GET, "/");

    // First request
    inboundChannel.write(req.retain());

    assertEquals(1, inboundChannel.outboundMessages().size());
    assertTrue(inboundChannel.outboundMessages().poll() instanceof ByteBuf);

    // First response
    DefaultFullHttpResponse resp = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    assertFalse(inboundChannel.writeInbound(resp));

    assertEquals(1, outboundChannel.outboundMessages().size());
    assertEquals(resp, outboundChannel.outboundMessages().poll());

    // Second request
    inboundChannel.write(req);

    assertEquals(1, inboundChannel.outboundMessages().size());
    assertTrue(inboundChannel.outboundMessages().poll() instanceof ByteBuf);

    // Second response
    assertFalse(inboundChannel.writeInbound(resp));

    assertEquals(1, outboundChannel.outboundMessages().size());
    assertEquals(resp, outboundChannel.outboundMessages().poll());

    resp.release();
}
 

/**
 * Provided a cause, returns an error response with the right status and error message.
 * @param cause the cause of the error.
 * @return a {@link FullHttpResponse} with the error message that can be sent to the client.
 */
private FullHttpResponse getErrorResponse(Throwable cause) {
  HttpResponseStatus status;
  RestServiceErrorCode restServiceErrorCode = null;
  String errReason = null;
  if (cause instanceof RestServiceException) {
    RestServiceException restServiceException = (RestServiceException) cause;
    restServiceErrorCode = restServiceException.getErrorCode();
    errorResponseStatus = ResponseStatus.getResponseStatus(restServiceErrorCode);
    status = getHttpResponseStatus(errorResponseStatus);
    if (shouldSendFailureReason(status, restServiceException)) {
      errReason = new String(
          Utils.getRootCause(cause).getMessage().replaceAll("[\n\t\r]", " ").getBytes(StandardCharsets.US_ASCII),
          StandardCharsets.US_ASCII);
    }
  } else if (Utils.isPossibleClientTermination(cause)) {
    nettyMetrics.clientEarlyTerminationCount.inc();
    status = HttpResponseStatus.INTERNAL_SERVER_ERROR;
    errorResponseStatus = ResponseStatus.InternalServerError;
  } else {
    nettyMetrics.internalServerErrorCount.inc();
    status = HttpResponseStatus.INTERNAL_SERVER_ERROR;
    errorResponseStatus = ResponseStatus.InternalServerError;
  }
  logger.trace("Constructed error response for the client - [{} - {}]", status, errReason);
  FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status);
  response.headers().set(HttpHeaderNames.DATE, new GregorianCalendar().getTime());
  HttpUtil.setContentLength(response, 0);
  if (errReason != null) {
    response.headers().set(FAILURE_REASON_HEADER, errReason);
  }
  if (restServiceErrorCode != null && HttpStatusClass.CLIENT_ERROR.contains(status.code())) {
    response.headers().set(ERROR_CODE_HEADER, restServiceErrorCode.name());
  }
  response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
  // if there is an ALLOW header in the response so far constructed, copy it
  if (responseMetadata.headers().contains(HttpHeaderNames.ALLOW)) {
    response.headers().set(HttpHeaderNames.ALLOW, responseMetadata.headers().get(HttpHeaderNames.ALLOW));
  } else if (errorResponseStatus == ResponseStatus.MethodNotAllowed) {
    logger.warn("Response is {} but there is no value for {}", ResponseStatus.MethodNotAllowed,
        HttpHeaderNames.ALLOW);
  }
  copyTrackingHeaders(responseMetadata, response);
  HttpUtil.setKeepAlive(response, shouldKeepAlive(status));
  return response;
}
 

public static FullHttpResponse unauthorized() {
   return new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.UNAUTHORIZED);
}
 

@Override
public FullHttpResponse respond(FullHttpRequest req, ChannelHandlerContext ctx) throws Exception {
   return new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NO_CONTENT);
}