Java 类io.netty.handler.codec.http2.DefaultHttp2Headers 实例源码

/**
 * Converts the specified Armeria HTTP/2 headers into Netty HTTP/2 headers.
 */
public static Http2Headers toNettyHttp2(HttpHeaders in) {
    final Http2Headers out = new DefaultHttp2Headers(false, in.size());
    out.set(in);
    out.remove(HttpHeaderNames.CONNECTION);
    out.remove(HttpHeaderNames.TRANSFER_ENCODING);
    out.remove(HttpHeaderNames.TRAILER);

    if (!out.contains(HttpHeaderNames.COOKIE)) {
        return out;
    }

    // Split up cookies to allow for better compression.
    // https://tools.ietf.org/html/rfc7540#section-8.1.2.5
    final List<CharSequence> cookies = out.getAllAndRemove(HttpHeaderNames.COOKIE);
    for (CharSequence c : cookies) {
        out.add(HttpHeaderNames.COOKIE, COOKIE_SPLITTER.split(c));
    }

    return out;
}

@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);

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

@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().body(Unpooled.EMPTY_BUFFER).build();

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

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  Http2Response responseOut = (Http2Response) responses.remove(0);

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

@Test
public void testFullResponse() throws Exception {
  outputReceived = new CountDownLatch(1);
  Http2Headers headers = new DefaultHttp2Headers().status("200");
  Http2Response responseIn = Http2Response.build(1, headers, true);

  channel.writeInbound(responseIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  Response responseOut = responses.remove(0);

  assertTrue(responseOut != null);
  assertTrue(responseOut instanceof FullResponse);
  assertEquals("h2", responseOut.version());
  assertEquals(OK, responseOut.status());
  assertFalse(responseOut.hasBody());
  assertEquals(1, responseOut.streamId());
}

@Override
public void push(final String method, final String path, final Map<String, Object> headers) {
  ctx.channel().eventLoop().execute(() -> {
    AsciiString streamIdHeader = HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text();
    Http2Connection connection = encoder.connection();
    int nextStreamId = connection.local().incrementAndGetNextStreamId();
    Http2Headers h2headers = new DefaultHttp2Headers()
        .path(path)
        .method(method)
        .authority(authority)
        .scheme(scheme);
    headers.forEach((n, v) -> h2headers.add(n, v.toString()));
    encoder.writePushPromise(ctx, streamId, nextStreamId, h2headers, 0, ctx.newPromise());

    // TODO: Is there another way of handling a push promise?
    DefaultFullHttpRequest pushRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
        HttpMethod.valueOf(method.toUpperCase()), path, Unpooled.EMPTY_BUFFER,
        new DefaultHttpHeaders(false).set(streamIdHeader, nextStreamId),
        EmptyHttpHeaders.INSTANCE);
    ctx.pipeline().fireChannelRead(pushRequest);
    ctx.pipeline().fireChannelReadComplete();
  });
}

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
public void closeBeforeClientHalfCloseShouldSucceed() throws Exception {
  ListMultimap<CharSequence, CharSequence> expectedHeaders =
      ImmutableListMultimap.copyOf(new DefaultHttp2Headers()
          .status(new AsciiString("200"))
          .set(new AsciiString("content-type"), new AsciiString("application/grpc"))
          .set(new AsciiString("grpc-status"), new AsciiString("0")));

  stream().close(Status.OK, new Metadata());

  ArgumentCaptor<SendResponseHeadersCommand> sendHeadersCap =
      ArgumentCaptor.forClass(SendResponseHeadersCommand.class);
  verify(writeQueue).enqueue(sendHeadersCap.capture(), eq(true));
  SendResponseHeadersCommand sendHeaders = sendHeadersCap.getValue();
  assertThat(sendHeaders.stream()).isSameAs(stream.transportState());
  assertThat(ImmutableListMultimap.copyOf(sendHeaders.headers()))
      .containsExactlyEntriesIn(expectedHeaders);
  assertThat(sendHeaders.endOfStream()).isTrue();
  verifyZeroInteractions(serverListener);

  // Sending complete. Listener gets closed()
  stream().transportState().complete();

  verify(serverListener).closed(Status.OK);
  assertNull("no message expected", listenerMessageQueue.poll());
}

@Test
public void closeWithErrorBeforeClientHalfCloseShouldSucceed() throws Exception {
  ListMultimap<CharSequence, CharSequence> expectedHeaders =
      ImmutableListMultimap.copyOf(new DefaultHttp2Headers()
          .status(new AsciiString("200"))
          .set(new AsciiString("content-type"), new AsciiString("application/grpc"))
          .set(new AsciiString("grpc-status"), new AsciiString("1")));

  // Error is sent on wire and ends the stream
  stream().close(Status.CANCELLED, trailers);

  ArgumentCaptor<SendResponseHeadersCommand> sendHeadersCap =
      ArgumentCaptor.forClass(SendResponseHeadersCommand.class);
  verify(writeQueue).enqueue(sendHeadersCap.capture(), eq(true));
  SendResponseHeadersCommand sendHeaders = sendHeadersCap.getValue();
  assertThat(sendHeaders.stream()).isSameAs(stream.transportState());
  assertThat(ImmutableListMultimap.copyOf(sendHeaders.headers()))
      .containsExactlyEntriesIn(expectedHeaders);
  assertThat(sendHeaders.endOfStream()).isTrue();
  verifyZeroInteractions(serverListener);

  // Sending complete. Listener gets closed()
  stream().transportState().complete();
  verify(serverListener).closed(Status.OK);
  assertNull("no message expected", listenerMessageQueue.poll());
}

@Test
public void emptyFramerShouldSendNoPayload() {
  ListMultimap<CharSequence, CharSequence> expectedHeaders =
      ImmutableListMultimap.copyOf(new DefaultHttp2Headers()
          .status(new AsciiString("200"))
          .set(new AsciiString("content-type"), new AsciiString("application/grpc"))
          .set(new AsciiString("grpc-status"), new AsciiString("0")));
  ArgumentCaptor<SendResponseHeadersCommand> cmdCap =
      ArgumentCaptor.forClass(SendResponseHeadersCommand.class);

  stream().close(Status.OK, new Metadata());

  verify(writeQueue).enqueue(cmdCap.capture(), eq(true));
  SendResponseHeadersCommand cmd = cmdCap.getValue();
  assertThat(cmd.stream()).isSameAs(stream.transportState());
  assertThat(ImmutableListMultimap.copyOf(cmd.headers()))
      .containsExactlyEntriesIn(expectedHeaders);
  assertThat(cmd.endOfStream()).isTrue();
}

@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 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(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(false), any(ChannelPromise.class));
}

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

@Test
public void headersWithMissingPathShouldFail() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_METHOD)
      .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC);
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);
  Http2Headers responseHeaders = new DefaultHttp2Headers()
      .set(InternalStatus.CODE_KEY.name(), String.valueOf(Code.UNIMPLEMENTED.value()))
      .set(InternalStatus.MESSAGE_KEY.name(), "Expected path but is missing")
      .status("" + 404)
      .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));
}

@Test
public void headersWithInvalidPathShouldFail() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_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.UNIMPLEMENTED.value()))
      .set(InternalStatus.MESSAGE_KEY.name(), "Expected path to start with /: foo/bar")
      .status("" + 404)
      .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));
}

@Test
public void headersSupportExtensionContentType() throws Exception {
  manualSetUp();
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_METHOD)
      .set(CONTENT_TYPE_HEADER, new AsciiString("application/grpc+json", UTF_8))
      .set(TE_HEADER, TE_TRAILERS)
      .path(new AsciiString("/foo/bar"));
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);

  ArgumentCaptor<NettyServerStream> streamCaptor =
      ArgumentCaptor.forClass(NettyServerStream.class);
  ArgumentCaptor<String> methodCaptor = ArgumentCaptor.forClass(String.class);
  verify(transportListener).streamCreated(streamCaptor.capture(), methodCaptor.capture(),
      any(Metadata.class));
  stream = streamCaptor.getValue();
}

private void createStream() throws Exception {
  Http2Headers headers = new DefaultHttp2Headers()
      .method(HTTP_METHOD)
      .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)
      .set(TE_HEADER, TE_TRAILERS)
      .path(new AsciiString("/foo/bar"));
  ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
  channelRead(headersFrame);

  ArgumentCaptor<NettyServerStream> streamCaptor =
      ArgumentCaptor.forClass(NettyServerStream.class);
  ArgumentCaptor<String> methodCaptor = ArgumentCaptor.forClass(String.class);
  verify(transportListener).streamCreated(streamCaptor.capture(), methodCaptor.capture(),
      any(Metadata.class));
  stream = streamCaptor.getValue();
}

@Test
public void inboundShouldForwardToStream() throws Exception {
  createStream();

  // Read a headers frame first.
  Http2Headers headers = new DefaultHttp2Headers().status(STATUS_OK)
      .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)
      .set(as("magic"), as("value"));
  ByteBuf headersFrame = headersFrame(3, headers);
  channelRead(headersFrame);
  ArgumentCaptor<Metadata> captor = ArgumentCaptor.forClass(Metadata.class);
  verify(streamListener).headersRead(captor.capture());
  assertEquals("value",
      captor.getValue().get(Metadata.Key.of("magic", Metadata.ASCII_STRING_MARSHALLER)));

  streamTransportState.requestMessagesFromDeframer(1);

  // Create a data frame and then trigger the handler to read it.
  ByteBuf frame = grpcDataFrame(3, false, contentAsArray());
  channelRead(frame);
  InputStream message = streamListenerMessageQueue.poll();
  assertArrayEquals(ByteBufUtil.getBytes(content()), ByteStreams.toByteArray(message));
  message.close();
  assertNull("no additional message expected", streamListenerMessageQueue.poll());
}

@Test
public void convertServerHeaders_sanitizes() {
  Metadata metaData = new Metadata();

  // Intentionally being explicit here rather than relying on any pre-defined lists of headers,
  // since the goal of this test is to validate the correctness of such lists in the first place.
  metaData.put(GrpcUtil.CONTENT_TYPE_KEY, "to-be-removed");
  metaData.put(GrpcUtil.TE_HEADER, "to-be-removed");
  metaData.put(GrpcUtil.USER_AGENT_KEY, "to-be-removed");
  metaData.put(userKey, userValue);

  Http2Headers output = Utils.convertServerHeaders(metaData);
  DefaultHttp2Headers headers = new DefaultHttp2Headers();
  for (Map.Entry<CharSequence, CharSequence> entry : output) {
    headers.add(entry.getKey(), entry.getValue());
  }
  // 2 reserved headers, 1 user header
  assertEquals(2 + 1, headers.size());
  assertEquals(Utils.CONTENT_TYPE_GRPC, headers.get(GrpcUtil.CONTENT_TYPE_KEY.name()));
}

private static Http2Headers createResponseHeadersFromRequestHeaders(
        Http2Headers requestHeaders) {
    // Create response headers by echoing request headers.
    Http2Headers responseHeaders = new DefaultHttp2Headers().status(OK.codeAsText());
    for (Map.Entry<CharSequence, CharSequence> header : requestHeaders) {
        if (!header.getKey().toString().startsWith(":")) {
            responseHeaders.add("echo-" + header.getKey(), header.getValue());
        }
    }

    responseHeaders.add("echo-method", requestHeaders.get(":method").toString());
    return responseHeaders;
}

@Override
public void onHeadersRead(ChannelHandlerContext ctx, int streamId,
    Http2Headers headers, int streamDependency, short weight,
    boolean exclusive, int padding, boolean endStream) throws Http2Exception {
  encoder.writeHeaders(ctx, streamId,
    new DefaultHttp2Headers().status(HttpResponseStatus.OK.codeAsText()), 0,
    false, ctx.newPromise());
  encoder.writeData(
    ctx,
    streamId,
    ctx.alloc().buffer()
        .writeBytes("HTTP/2 DTP".getBytes(StandardCharsets.UTF_8)), 0, true,
    ctx.newPromise());
}

/**
 * Sends a "Hello World" DATA frame to the client.
 */
private void sendResponse(ChannelHandlerContext ctx, ByteBuf payload) {
    // Send a frame for the response status
    Http2Headers headers = new DefaultHttp2Headers().status(OK.codeAsText());
    ctx.write(new DefaultHttp2HeadersFrame(headers));
    ctx.writeAndFlush(new DefaultHttp2DataFrame(payload, true));
}

/**
 * Handles the cleartext HTTP upgrade event. If an upgrade occurred, sends a simple response via HTTP/2
 * on stream 1 (the stream specifically reserved for cleartext HTTP upgrade).
 */
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof HttpServerUpgradeHandler.UpgradeEvent) {
        // Write an HTTP/2 response to the upgrade request
        Http2Headers headers =
                new DefaultHttp2Headers().status(OK.codeAsText())
                .set(new AsciiString(UPGRADE_RESPONSE_HEADER), new AsciiString("true"));
        encoder().writeHeaders(ctx, 1, headers, 0, true, ctx.newPromise());
    }
    super.userEventTriggered(ctx, evt);
}

/**
 * Sends a "Hello World" DATA frame to the client.
 */
private void sendResponse(ChannelHandlerContext ctx, int streamId, ByteBuf payload) {
    // Send a frame for the response status
    Http2Headers headers = new DefaultHttp2Headers().status(OK.codeAsText());
    encoder().writeHeaders(ctx, streamId, headers, 0, false, ctx.newPromise());
    encoder().writeData(ctx, streamId, payload, 0, true, ctx.newPromise());
    ctx.flush();
}

/**
 * Sends a "Hello World" DATA frame to the client.
 */
private void sendResponse(ChannelHandlerContext ctx, int streamId, ByteBuf payload) {
    System.out.println("bbb");
    // Send a frame for the response status
    Http2Headers headers = new DefaultHttp2Headers().status(OK.codeAsText());
    encoder.writeHeaders(ctx, streamId, headers, 0, false, ctx.newPromise());
    encoder.writeData(ctx, streamId, payload, 0, true, ctx.newPromise());
    ctx.flush();
}

private void writeErrorResponse(ChannelHandlerContext ctx, int streamId, HttpResponseStatus status) {
    final byte[] content = status.toString().getBytes(StandardCharsets.UTF_8);

    writer.writeHeaders(
            ctx, streamId,
            new DefaultHttp2Headers(false)
                    .status(status.codeAsText())
                    .set(HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8.toString())
                    .setInt(HttpHeaderNames.CONTENT_LENGTH, content.length),
            0, false, ctx.voidPromise());

    writer.writeData(
            ctx, streamId, Unpooled.wrappedBuffer(content), 0, true, ctx.voidPromise());
}

@Test
public void inboundCookiesMustBeMergedForHttp2() {
    final Http2Headers in = new DefaultHttp2Headers();

    in.add(HttpHeaderNames.COOKIE, "a=b; c=d");
    in.add(HttpHeaderNames.COOKIE, "e=f;g=h");
    in.addObject(HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
    in.add(HttpHeaderNames.COOKIE, "i=j");
    in.add(HttpHeaderNames.COOKIE, "k=l;");

    final HttpHeaders out = toArmeria(in);

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

/**
 * This method handles the cleartext HTTP upgrade event. If an upgrade occurred, sends a simple response via HTTP/2
 * on stream 1 (the stream specifically reserved for cleartext HTTP upgrade).
 *
 * @param ctx Channel context
 * @param evt Event
 * @throws Exception Throws when user event trigger has an error
 */
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof HttpServerUpgradeHandler.UpgradeEvent) {
        // Write an HTTP/2 response to the upgrade request
        Http2Headers headers =
                new DefaultHttp2Headers().status(OK.codeAsText())
                        .set(new AsciiString(Constants.UPGRADE_RESPONSE_HEADER), new AsciiString("true"));
        encoder().writeHeaders(ctx, 1, headers, 0, true, ctx.newPromise());
    }
    super.userEventTriggered(ctx, evt);
}

default Http2Headers http2Headers() {
  Http2Headers result = new DefaultHttp2Headers();
  for (Entry<CharSequence, CharSequence> entry : this) {
    result.add(entry.getKey(), entry.getValue());
  }
  return result;
}

@Override
public void handle(Http2Request request, ChannelHandlerContext ctx) {
  Http2Headers headers =
      new DefaultHttp2Headers().status(HttpResponseStatus.NOT_FOUND.codeAsText());
  Http2Response<Http2Headers> not_found = Http2Response.build(request.streamId, headers, true);
  ctx.writeAndFlush(not_found);
}

/**
 * Sends a "Hello World" DATA frame to the client.
 *
 * @param ctx channel handler context
 * @param payload payload
 */
private void sendResponse(ChannelHandlerContext ctx, ByteBuf payload) {
  // Send a frame for the response status
  Http2Headers headers = new DefaultHttp2Headers().status(HttpResponseStatus.OK.codeAsText());
  ctx.write(new DefaultHttp2HeadersFrame(headers));
  ctx.writeAndFlush(new DefaultHttp2DataFrame(payload, true));
}

@Test
public void testFullResponseWithBody() throws Exception {
  outputReceived = new CountDownLatch(2);
  ByteBuf body = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, "response");

  Http2Headers headersIn = new DefaultHttp2Headers().method("GET").path("/");
  Http2Request requestIn = Http2Request.build(1, headersIn, true);
  FullResponse responseIn = ResponseBuilders.newOk().body(body).build();

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

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  Http2Response responseOut = (Http2Response) responses.remove(0);

  assertTrue(responseOut != null);
  assertTrue(responseOut.payload instanceof Http2Headers);
  assertEquals("200", ((Http2Headers) responseOut.payload).status().toString());
  assertFalse(responseOut.eos);
  assertEquals(1, responseOut.streamId);

  Http2Response bodyOut1 = (Http2Response) responses.remove(0);

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

@Test
public void testAddHeaders() {
  Http2Headers other = new DefaultHttp2Headers();
  other.add("header", "value");
  headers.add(other);
  assertEquals(other, headers.delegate());
  assertEquals("value", headers.delegate().get("header"));
}

@Test
public void testEquals() {
  assertTrue(headers.equals(headers));
  assertFalse(headers.equals((Integer) 1));
  Http2HeadersWrapper otherHeaders = new Http2HeadersWrapper(new DefaultHttp2Headers());
  assertTrue(headers.equals(otherHeaders));
  otherHeaders.add("header", "value");
  assertFalse(headers.equals(otherHeaders));
}

@Test
public void testSetHeaders() {
  Http2Headers other = new DefaultHttp2Headers();
  other.set("header", "value");
  headers.set(other);
  assertEquals(other, headers.delegate());
  assertEquals("value", headers.delegate().get("header"));
  other.set("header", "value2");
  headers.set(other);
  assertEquals(other, headers.delegate());
  assertEquals("value2", headers.delegate().get("header"));
}

@Test
public void testSetAll() {
  Http2Headers other = new DefaultHttp2Headers();
  other.set("header", "value");
  headers.setAll(other);
  assertEquals(other, headers.delegate());
  assertEquals("value", headers.delegate().get("header"));
  other.set("header", "value2");
  headers.setAll(other);
  assertEquals(other, headers.delegate());
  assertEquals("value2", headers.delegate().get("header"));
}

/**
 * Checkstyle.
 */
@Benchmark
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
public void defaultHeaders_serverHandler(Blackhole bh) {
  serverHandler(bh, new DefaultHttp2Headers(true, 9));
}

/**
 * Checkstyle.
 */
@Benchmark
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
public void defaultHeaders_clientHandler(Blackhole bh) {
  clientHandler(bh, new DefaultHttp2Headers(true, 2));
}

@Test
public void writeMessageShouldSendResponse() throws Exception {
  ListMultimap<CharSequence, CharSequence> expectedHeaders =
      ImmutableListMultimap.copyOf(new DefaultHttp2Headers()
          .status(Utils.STATUS_OK)
          .set(Utils.CONTENT_TYPE_HEADER, Utils.CONTENT_TYPE_GRPC));

  stream.writeHeaders(new Metadata());

  ArgumentCaptor<SendResponseHeadersCommand> sendHeadersCap =
      ArgumentCaptor.forClass(SendResponseHeadersCommand.class);
  verify(writeQueue).enqueue(sendHeadersCap.capture(), eq(true));
  SendResponseHeadersCommand sendHeaders = sendHeadersCap.getValue();
  assertThat(sendHeaders.stream()).isSameAs(stream.transportState());
  assertThat(ImmutableListMultimap.copyOf(sendHeaders.headers()))
      .containsExactlyEntriesIn(expectedHeaders);
  assertThat(sendHeaders.endOfStream()).isFalse();

  byte[] msg = smallMessage();
  stream.writeMessage(new ByteArrayInputStream(msg));
  stream.flush();

  verify(writeQueue).enqueue(
      eq(new SendGrpcFrameCommand(stream.transportState(), messageFrame(MESSAGE), false)),
      isA(ChannelPromise.class),
      eq(true));
}

@Test
public void closeAfterClientHalfCloseShouldSucceed() throws Exception {
  ListMultimap<CharSequence, CharSequence> expectedHeaders =
      ImmutableListMultimap.copyOf(new DefaultHttp2Headers()
          .status(new AsciiString("200"))
          .set(new AsciiString("content-type"), new AsciiString("application/grpc"))
          .set(new AsciiString("grpc-status"), new AsciiString("0")));

  // Client half-closes. Listener gets halfClosed()
  stream().transportState()
      .inboundDataReceived(new EmptyByteBuf(UnpooledByteBufAllocator.DEFAULT), true);

  verify(serverListener).halfClosed();

  // Server closes. Status sent
  stream().close(Status.OK, trailers);
  assertNull("no message expected", listenerMessageQueue.poll());

  ArgumentCaptor<SendResponseHeadersCommand> cmdCap =
      ArgumentCaptor.forClass(SendResponseHeadersCommand.class);
  verify(writeQueue).enqueue(cmdCap.capture(), eq(true));
  SendResponseHeadersCommand cmd = cmdCap.getValue();
  assertThat(cmd.stream()).isSameAs(stream.transportState());
  assertThat(ImmutableListMultimap.copyOf(cmd.headers()))
      .containsExactlyEntriesIn(expectedHeaders);
  assertThat(cmd.endOfStream()).isTrue();

  // Sending and receiving complete. Listener gets closed()
  stream().transportState().complete();
  verify(serverListener).closed(Status.OK);
  assertNull("no message expected", listenerMessageQueue.poll());
}

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

  Http2Headers headers = new DefaultHttp2Headers(false);
  headers.add(of(":scheme"), of("https")).add(of(":method"), of("GET"))
      .add(of(":path"), of("index.html")).add(of(":authority"), of("foo.grpc.io"))
      .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(":scheme")), decodedHeaders.scheme());
  assertEquals(headers.get(of(":method")), decodedHeaders.method());
  assertEquals(headers.get(of(":path")), decodedHeaders.path());
  assertEquals(headers.get(of(":authority")), decodedHeaders.authority());
  assertEquals(headers.get(of("custom")), decodedHeaders.get(of("custom")));
  assertEquals(headers.size(), decodedHeaders.size());

  String toString = decodedHeaders.toString();
  assertContainsKeyAndValue(toString, ":scheme", decodedHeaders.scheme());
  assertContainsKeyAndValue(toString, ":method", decodedHeaders.method());
  assertContainsKeyAndValue(toString, ":path", decodedHeaders.path());
  assertContainsKeyAndValue(toString, ":authority", decodedHeaders.authority());
  assertContainsKeyAndValue(toString, "custom", decodedHeaders.get(of("custom")));
}