io.netty.handler.codec.http2.Http2FrameListener #io.netty.handler.codec.http2.Http2Headers源码实例Demo

@Test
public void basicCorrectness() {
  Http2Headers headers = new GrpcHttp2RequestHeaders(1);
  headers.add(of(":method"), of("POST"));
  headers.add(of("content-type"), of("application/grpc+proto"));
  headers.add(of(":path"), of("/google.pubsub.v2.PublisherService/CreateTopic"));
  headers.add(of(":scheme"), of("https"));
  headers.add(of("te"), of("trailers"));
  headers.add(of(":authority"), of("pubsub.googleapis.com"));
  headers.add(of("foo"), of("bar"));

  assertEquals(7, headers.size());
  // Number of headers without the pseudo headers and 'te' header.
  assertEquals(2, ((GrpcHttp2InboundHeaders)headers).numHeaders());

  assertEquals(of("application/grpc+proto"), headers.get(of("content-type")));
  assertEquals(of("/google.pubsub.v2.PublisherService/CreateTopic"), headers.path());
  assertEquals(of("https"), headers.scheme());
  assertEquals(of("POST"), headers.method());
  assertEquals(of("pubsub.googleapis.com"), headers.authority());
  assertEquals(of("trailers"), headers.get(of("te")));
  assertEquals(of("bar"), headers.get(of("foo")));
}
 

private void onHeadersRead(ChannelHandlerContext ctx, Http2HeadersFrame headers) {
    if (headers.isEndStream()) {
        ctx.write(new DefaultHttp2HeadersFrame(headers.headers(), true));
    } else {
        Http2Headers outHeaders = new DefaultHttp2Headers();
        if (headers.headers().contains(EXPECT, CONTINUE)) {
            if (headers.headers().contains(EXPECT_FAIL_HEADER)) {
                outHeaders.status(
                        io.netty.handler.codec.http.HttpResponseStatus.EXPECTATION_FAILED.codeAsText());
                ctx.write(new DefaultHttp2HeadersFrame(outHeaders, true));
                return;
            } else {
                outHeaders.status(io.netty.handler.codec.http.HttpResponseStatus.CONTINUE.codeAsText());
            }
        } else {
            outHeaders.status(io.netty.handler.codec.http.HttpResponseStatus.OK.codeAsText());
        }

        CharSequence contentType = headers.headers().get(CONTENT_TYPE);
        if (contentType != null) {
            outHeaders.add(CONTENT_TYPE, contentType);
        }
        outHeaders.add(HttpHeaderNames.COOKIE, headers.headers().getAll(HttpHeaderNames.COOKIE));
        ctx.write(new DefaultHttp2HeadersFrame(outHeaders));
    }
}
 

/** Test that several data frames will be aggregated into a response. */
@Test
void testOnDataRead_dataAggregated() {
  testHandler = new Http2Handler(mockEncoder, MAX_PAYLOAD, NO_CORS);

  // Create a fake request to aggregate data into.
  XrpcRequest fakeRequest = new XrpcRequest((Http2Headers) null, null, null, channel);
  testHandler.requests.put(STREAM_ID, fakeRequest);

  // Append several data frames.
  testHandler.onDataRead(
      mockContext, STREAM_ID, Unpooled.wrappedBuffer(new byte[] {1}), 0, false);
  testHandler.onDataRead(
      mockContext, STREAM_ID, Unpooled.wrappedBuffer(new byte[] {2}), 0, false);
  testHandler.onDataRead(
      mockContext, STREAM_ID, Unpooled.wrappedBuffer(new byte[] {3}), 0, false);

  // Assert that the request has all the data needed.
  assertEquals(Unpooled.wrappedBuffer(new byte[] {1, 2, 3}), fakeRequest.body());
}
 

@Test
public void keepAliveEnforcer_sendingDataResetsCounters() throws Exception {
  permitKeepAliveWithoutCalls = false;
  permitKeepAliveTimeInNanos = TimeUnit.HOURS.toNanos(1);
  manualSetUp();

  createStream();
  Http2Headers headers = Utils.convertServerHeaders(new Metadata());
  ChannelFuture future = enqueue(
      SendResponseHeadersCommand.createHeaders(stream.transportState(), headers));
  future.get();
  for (int i = 0; i < 10; i++) {
    future = enqueue(
        new SendGrpcFrameCommand(stream.transportState(), content().retainedSlice(), false));
    future.get();
    channel().releaseOutbound();
    channelRead(pingFrame(false /* isAck */, 1L));
  }
  verifyWrite(never()).writeGoAway(eq(ctx()), eq(STREAM_ID),
      eq(Http2Error.ENHANCE_YOUR_CALM.code()), any(ByteBuf.class), any(ChannelPromise.class));
}
 

@Test
public void testFullResponse() throws Exception {
  outputReceived = new CountDownLatch(2);
  Http2Headers headersIn = new DefaultHttp2Headers().method("GET").path("/");
  Http2Request requestIn = Http2Request.build(1, headersIn, true);
  FullResponse responseIn =
      ResponseBuilders.newOk().streamId(1).body(Unpooled.EMPTY_BUFFER).build();

  channel.writeInbound(requestIn);
  channel.runPendingTasks(); // blocks
  channel.writeOutbound(responseIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  Http2Response responseOut = responses.remove(0);

  assertNotNull(responseOut);
  assertTrue(responseOut.payload instanceof Http2Headers);
  assertEquals("200", ((Http2Headers) responseOut.payload).status().toString());
  assertTrue(responseOut.eos);
  assertEquals(1, responseOut.streamId);
}
 

private Http2Headers convertHeaders(HttpHeaders inputHeaders, boolean isTrailersEmpty) {
    final Http2Headers outHeaders = ArmeriaHttpUtil.toNettyHttp2ServerHeaders(inputHeaders);
    if (!isTrailersEmpty && outHeaders.contains(HttpHeaderNames.CONTENT_LENGTH)) {
        // We don't apply chunked encoding when the content-length header is set, which would
        // prevent the trailers from being sent so we go ahead and remove content-length to force
        // chunked encoding.
        outHeaders.remove(HttpHeaderNames.CONTENT_LENGTH);
    }

    if (enableServerHeader && !outHeaders.contains(HttpHeaderNames.SERVER)) {
        outHeaders.add(HttpHeaderNames.SERVER, ArmeriaHttpUtil.SERVER_HEADER);
    }

    if (enableDateHeader && !outHeaders.contains(HttpHeaderNames.DATE)) {
        outHeaders.add(HttpHeaderNames.DATE, HttpTimestampSupplier.currentTime());
    }
    return outHeaders;
}
 

@Test
public void testFullRequest() throws Exception {
  outputReceived = new CountDownLatch(1);

  Http2Headers headers = new DefaultHttp2Headers().method("GET").path("/");
  Http2Request requestIn = Http2Request.build(1, headers, true);

  channel.writeInbound(requestIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  Request requestOut = requests.remove(0);

  assertNotNull(requestOut);
  assertTrue(requestOut instanceof FullRequest);
  assertEquals("h2", requestOut.version());
  assertEquals(HttpMethod.GET, requestOut.method());
  assertEquals("/", requestOut.path());
  assertFalse(requestOut.hasBody());
  assertNotNull(requestOut.body());
  assertEquals(0, requestOut.body().readableBytes());
  assertEquals(1, requestOut.streamId());
}
 

private void respondWithHttpError(
    ChannelHandlerContext ctx, int streamId, int code, Status.Code statusCode, String msg) {
  Metadata metadata = new Metadata();
  metadata.put(InternalStatus.CODE_KEY, statusCode.toStatus());
  metadata.put(InternalStatus.MESSAGE_KEY, msg);
  byte[][] serialized = InternalMetadata.serialize(metadata);

  Http2Headers headers = new DefaultHttp2Headers(true, serialized.length / 2)
      .status("" + code)
      .set(CONTENT_TYPE_HEADER, "text/plain; encoding=utf-8");
  for (int i = 0; i < serialized.length; i += 2) {
    headers.add(new AsciiString(serialized[i], false), new AsciiString(serialized[i + 1], false));
  }
  encoder().writeHeaders(ctx, streamId, headers, 0, false, ctx.newPromise());
  ByteBuf msgBuf = ByteBufUtil.writeUtf8(ctx.alloc(), msg);
  encoder().writeData(ctx, streamId, msgBuf, 0, true, ctx.newPromise());
}
 

@Test
void toArmeriaRequestHeaders() {
    final Http2Headers in = new DefaultHttp2Headers().set("a", "b");

    final InetSocketAddress socketAddress = new InetSocketAddress(36462);
    final Channel channel = mock(Channel.class);
    when(channel.localAddress()).thenReturn(socketAddress);

    final ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
    when(ctx.channel()).thenReturn(channel);

    in.set(HttpHeaderNames.METHOD, "GET")
      .set(HttpHeaderNames.PATH, "/");
    // Request headers without pseudo headers.
    final RequestHeaders headers =
            ArmeriaHttpUtil.toArmeriaRequestHeaders(ctx, in, false, "https", serverConfig());
    assertThat(headers.scheme()).isEqualTo("https");
    assertThat(headers.authority()).isEqualTo("foo:36462");
}
 

/**
 * Writes the given response body as "text/plain" to the given stream. Marks the response status
 * metric. Closes the stream after writing the response.
 */
private void writeResponse(
    ChannelHandlerContext ctx, int streamId, HttpResponseStatus status, ByteBuf body) {

  Preconditions.checkArgument(body != null, "body must not be null");

  markResponseStatus(ctx, status);

  Http2Headers headers = new DefaultHttp2Headers(true);
  // TODO(jkinkead): This should honor accept headers; we shouldn't send text/plain if the client
  // doesn't want it.
  headers.set(CONTENT_TYPE, "text/plain");
  headers.setInt(CONTENT_LENGTH, body.readableBytes());
  headers.status(status.codeAsText());

  writeResponse(ctx, streamId, headers, Optional.of(body));
}
 

@Test
public void headersWithInvalidContentTypeShouldFail() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_METHOD)
      .set(CONTENT_TYPE_HEADER, new AsciiString("application/bad", UTF_8))
      .set(TE_HEADER, TE_TRAILERS)
      .path(new AsciiString("/foo/bar"));
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);
  Http2Headers responseHeaders = new DefaultHttp2Headers()
      .set(InternalStatus.CODE_KEY.name(), String.valueOf(Code.INTERNAL.value()))
      .set(InternalStatus.MESSAGE_KEY.name(), "Content-Type 'application/bad' is not supported")
      .status("" + 415)
      .set(CONTENT_TYPE_HEADER, "text/plain; encoding=utf-8");

  verifyWrite()
      .writeHeaders(
          eq(ctx()),
          eq(STREAM_ID),
          eq(responseHeaders),
          eq(0),
          eq(false),
          any(ChannelPromise.class));
}
 

public static Http2Headers convertClientHeaders(Metadata headers,
    AsciiString scheme,
    AsciiString defaultPath,
    AsciiString authority,
    AsciiString method,
    AsciiString userAgent) {
  Preconditions.checkNotNull(defaultPath, "defaultPath");
  Preconditions.checkNotNull(authority, "authority");
  Preconditions.checkNotNull(method, "method");

  // Discard any application supplied duplicates of the reserved headers
  headers.discardAll(CONTENT_TYPE_KEY);
  headers.discardAll(GrpcUtil.TE_HEADER);
  headers.discardAll(GrpcUtil.USER_AGENT_KEY);

  return GrpcHttp2OutboundHeaders.clientRequestHeaders(
      toHttp2Headers(headers),
      authority,
      defaultPath,
      method,
      scheme,
      userAgent);
}
 

public static Http2Headers convertClientHeaders(Metadata headers,
    AsciiString scheme,
    AsciiString defaultPath,
    AsciiString authority,
    AsciiString method,
    AsciiString userAgent) {
  Preconditions.checkNotNull(defaultPath, "defaultPath");
  Preconditions.checkNotNull(authority, "authority");
  Preconditions.checkNotNull(method, "method");

  // Discard any application supplied duplicates of the reserved headers
  headers.discardAll(CONTENT_TYPE_KEY);
  headers.discardAll(GrpcUtil.TE_HEADER);
  headers.discardAll(GrpcUtil.USER_AGENT_KEY);

  return GrpcHttp2OutboundHeaders.clientRequestHeaders(
      toHttp2Headers(headers),
      authority,
      defaultPath,
      method,
      scheme,
      userAgent);
}
 

@Test
public void binaryHeadersShouldBeBase64Decoded() {
  Http2Headers headers = new GrpcHttp2RequestHeaders(1);

  byte[] data = new byte[100];
  new Random().nextBytes(data);
  headers.add(of("foo-bin"), of(BASE64_ENCODING_OMIT_PADDING.encode(data)));

  assertEquals(1, headers.size());

  byte[][] namesAndValues = ((GrpcHttp2InboundHeaders)headers).namesAndValues();

  assertEquals(of("foo-bin"), new AsciiString(namesAndValues[0]));
  assertNotSame(data, namesAndValues[1]);
  assertArrayEquals(data, namesAndValues[1]);
}
 

@Test
public void decode_responseHeaders() throws Http2Exception {
  Http2HeadersDecoder decoder = new GrpcHttp2ClientHeadersDecoder(DEFAULT_MAX_HEADER_LIST_SIZE);
  Http2HeadersEncoder encoder =
      new DefaultHttp2HeadersEncoder(NEVER_SENSITIVE);

  Http2Headers headers = new DefaultHttp2Headers(false);
  headers.add(of(":status"), of("200")).add(of("custom"), of("header"));
  encodedHeaders = Unpooled.buffer();
  encoder.encodeHeaders(1 /* randomly chosen */, headers, encodedHeaders);

  Http2Headers decodedHeaders = decoder.decodeHeaders(3 /* randomly chosen */, encodedHeaders);
  assertEquals(headers.get(of(":status")), decodedHeaders.get(of(":status")));
  assertEquals(headers.get(of("custom")), decodedHeaders.get(of("custom")));
  assertEquals(headers.size(), decodedHeaders.size());

  String toString = decodedHeaders.toString();
  assertContainsKeyAndValue(toString, ":status", decodedHeaders.get(of(":status")));
  assertContainsKeyAndValue(toString, "custom", decodedHeaders.get(of("custom")));
}
 

@Test
public void dupBinHeadersWithComma() {
  Key<byte[]> key = Key.of("bytes-bin", BINARY_BYTE_MARSHALLER);
  Http2Headers http2Headers = new GrpcHttp2RequestHeaders(2);
  http2Headers.add(AsciiString.of("bytes-bin"), AsciiString.of("BaS,e6,,4+,padding=="));
  http2Headers.add(AsciiString.of("bytes-bin"), AsciiString.of("more"));
  http2Headers.add(AsciiString.of("bytes-bin"), AsciiString.of(""));
  Metadata recoveredHeaders = Utils.convertHeaders(http2Headers);
  byte[][] values = Iterables.toArray(recoveredHeaders.getAll(key), byte[].class);

  assertTrue(Arrays.deepEquals(
      new byte[][] {
          BaseEncoding.base64().decode("BaS"),
          BaseEncoding.base64().decode("e6"),
          BaseEncoding.base64().decode(""),
          BaseEncoding.base64().decode("4+"),
          BaseEncoding.base64().decode("padding"),
          BaseEncoding.base64().decode("more"),
          BaseEncoding.base64().decode("")},
      values));
}
 

@Test
public void headersWithInvalidMethodShouldFail() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_FAKE_METHOD)
      .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)
      .path(new AsciiString("/foo/bar"));
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);
  Http2Headers responseHeaders = new DefaultHttp2Headers()
      .set(InternalStatus.CODE_KEY.name(), String.valueOf(Code.INTERNAL.value()))
      .set(InternalStatus.MESSAGE_KEY.name(), "Method 'FAKE' is not supported")
      .status("" + 405)
      .set(CONTENT_TYPE_HEADER, "text/plain; encoding=utf-8");

  verifyWrite().writeHeaders(eq(ctx()), eq(STREAM_ID), eq(responseHeaders), eq(0),
      eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(false), any(ChannelPromise.class));
}
 

@Override
public void onHeadersRead(ChannelHandlerContext ctx, int streamId,
                          Http2Headers headers, int padding, boolean endOfStream) {
    if (endOfStream) {
        ByteBuf content = ctx.alloc().buffer();
        content.writeBytes(RESPONSE_BYTES.duplicate());
        ByteBufUtil.writeAscii(content, " - via HTTP/2");
        sendResponse(ctx, streamId, content);
    }
}
 

@Override
public void onHeadersRead(ChannelHandlerContext ctx,
    int streamId,
    Http2Headers headers,
    int streamDependency,
    short weight,
    boolean exclusive,
    int padding,
    boolean endStream) throws Http2Exception {
  NettyClientHandler.this.onHeadersRead(streamId, headers, endStream);
}
 

private static Http2Headers http1HeadersToHttp2Headers(FullHttpRequest request) {
    CharSequence host = request.headers().get(HttpHeaderNames.HOST);
    Http2Headers http2Headers = new DefaultHttp2Headers()
            .method(HttpMethod.GET.asciiName())
            .path(request.uri())
            .scheme(HttpScheme.HTTP.name());
    if (host != null) {
        http2Headers.authority(host);
    }
    return http2Headers;
}
 

private Response wrapHeaders(Http2Headers headers, int streamId, boolean eos) {
  if (eos) {
    return new FullHttp2Response(headers, streamId);
  } else {
    return new SegmentedHttp2Response(headers, streamId);
  }
}
 

/**
 * Converts the specified Armeria HTTP/2 request headers into Netty HTTP/2 headers.
 *
 * @param inputHeaders the HTTP/2 request headers to convert.
 */
public static Http2Headers toNettyHttp2ClientHeader(HttpHeaders inputHeaders) {
    final int headerSizeHint = inputHeaders.size() + 3; // User_Agent, :scheme and :authority.
    final Http2Headers outputHeaders = new DefaultHttp2Headers(false, headerSizeHint);
    toNettyHttp2Client(inputHeaders, outputHeaders, false);
    return outputHeaders;
}
 

private void parseHttp2Request(Http2Headers headers, SofaRequest sofaRequest) {
    String targetApp = StringUtils.toString(headers.get(RemotingConstants.HEAD_TARGET_APP));
    sofaRequest.setTargetAppName(targetApp);
    // 获取序列化类型
    byte serializeType;
    CharSequence codeName = headers.get(RemotingConstants.HEAD_SERIALIZE_TYPE);
    if (codeName != null) {
        serializeType = HttpTransportUtils.getSerializeTypeByName(codeName.toString());
    } else {
        String contentType = StringUtils.toString(headers.get(HttpHeaderNames.CONTENT_TYPE));
        serializeType = HttpTransportUtils.getSerializeTypeByContentType(contentType);
    }
    sofaRequest.setSerializeType(serializeType);

    // 解析trace信息
    Map<String, String> traceMap = new HashMap<String, String>(8);
    Iterator<Map.Entry<CharSequence, CharSequence>> it = headers.iterator();
    while (it.hasNext()) {
        Map.Entry<CharSequence, CharSequence> entry = it.next();
        String key = entry.getKey().toString();
        if (HttpTracerUtils.isTracerKey(key)) {
            HttpTracerUtils.parseTraceKey(traceMap, key, StringUtils.toString(entry.getValue()));
        } else if (!key.startsWith(":")) {
            sofaRequest.addRequestProp(key, StringUtils.toString(entry.getValue()));
        }
    }
    if (!traceMap.isEmpty()) {
        sofaRequest.addRequestProp(RemotingConstants.RPC_TRACE_NAME, traceMap);
    }
}
 

private void onHeadersRead(int streamId, Http2Headers headers, boolean endStream) {
  // Stream 1 is reserved for the Upgrade response, so we should ignore its headers here:
  if (streamId != Http2CodecUtil.HTTP_UPGRADE_STREAM_ID) {
    NettyClientStream.TransportState stream = clientStream(requireHttp2Stream(streamId));
    PerfMark.event("NettyClientHandler.onHeadersRead", stream.tag());
    stream.transportHeadersReceived(headers, endStream);
  }

  if (keepAliveManager != null) {
    keepAliveManager.onDataReceived();
  }
}
 

@Test
public void testFullRequestWithBody() throws Exception {
  outputReceived = new CountDownLatch(1);
  ByteBuf body = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, "body");
  FullRequest requestIn = RequestBuilders.newPost("/").host("localhost").body(body).build();

  channel.writeOutbound(requestIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  Http2Request requestOut = (Http2Request) requests.remove(0);

  assertTrue(requestOut != null);
  assertTrue(requestOut.payload instanceof Http2Headers);
  assertEquals("POST", ((Http2Headers) requestOut.payload).method().toString());
  assertEquals("/", ((Http2Headers) requestOut.payload).path());
  assertFalse(requestOut.eos);

  Http2Request contentOut = (Http2Request) requests.remove(0);

  assertTrue(contentOut != null);
  assertTrue(contentOut.payload instanceof Http2DataFrame);
  assertEquals(body, ((Http2DataFrame) contentOut.payload).content());
  assertTrue(contentOut.eos);
}
 

@Test
void outboundCookiesMustBeSplitForHttp2() {
    final HttpHeaders in = HttpHeaders.builder()
                                      .add(HttpHeaderNames.COOKIE, "a=b; c=d")
                                      .add(HttpHeaderNames.COOKIE, "e=f;g=h")
                                      .addObject(HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8)
                                      .add(HttpHeaderNames.COOKIE, "i=j")
                                      .add(HttpHeaderNames.COOKIE, "k=l;")
                                      .build();

    final Http2Headers out = toNettyHttp2ClientHeader(in);
    assertThat(out.getAll(HttpHeaderNames.COOKIE))
            .containsExactly("a=b", "c=d", "e=f", "g=h", "i=j", "k=l");
}
 

@Override
public void onHeadersRead(ChannelHandlerContext ctx, int streamId,
                          Http2Headers headers, int padding, boolean endOfStream) {
    if (endOfStream) {
        ByteBuf content = ctx.alloc().buffer();
        content.writeBytes(RESPONSE_BYTES.duplicate());
        ByteBufUtil.writeAscii(content, " - via HTTP/2");
        sendResponse(ctx, streamId, content);
    }
}
 

public Http2PriorityHeadersEvent(int streamId, Http2Headers headers, int padding, boolean endOfStream,
                                 int streamDependency, short weight, boolean exclusive) {
    super(Http2FrameTypes.HEADERS, streamId);
    this.headers = headers;
    this.padding = padding;
    this.endOfStream = endOfStream;

    this.streamDependency = streamDependency;
    this.weight = weight;
    this.exclusive = exclusive;
}
 

static GrpcHttp2OutboundHeaders serverResponseHeaders(byte[][] serializedMetadata) {
  AsciiString[] preHeaders = new AsciiString[] {
      Http2Headers.PseudoHeaderName.STATUS.value(), Utils.STATUS_OK,
      Utils.CONTENT_TYPE_HEADER, Utils.CONTENT_TYPE_GRPC,
  };
  return new GrpcHttp2OutboundHeaders(preHeaders, serializedMetadata);
}
 

@Override
public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
    int streamDependency, short weight, boolean exclusive, int padding, boolean endStream,
    ChannelPromise promise) {
  keepAliveEnforcer.resetCounters();
  return super.writeHeaders(ctx, streamId, headers, streamDependency, weight, exclusive,
      padding, endStream, promise);
}