@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
    Integer streamId = msg.headers().getInt(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
    if (streamId == null) {
        System.err.println("HttpResponseHandler unexpected message received: " + msg);
        return;
    }

    Entry<ChannelFuture, ChannelPromise> entry = streamidPromiseMap.get(streamId);
    if (entry == null) {
        System.err.println("Message received for unknown stream id " + streamId);
    } else {
        // Do stuff with the message (for now just print it)
        ByteBuf content = msg.content();
        if (content.isReadable()) {
            int contentLength = content.readableBytes();
            byte[] arr = new byte[contentLength];
            content.readBytes(arr);
            System.out.println(new String(arr, 0, contentLength, CharsetUtil.UTF_8));
        }

        entry.getValue().setSuccess();
    }
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
    Integer streamId = msg.headers().getInt(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
    if (streamId == null) {
        log.error("Http2ResponseHandler unexpected message received: " + msg);
        return;
    }

    Entry<ChannelFuture, ChannelPromise> entry = streamidPromiseMap.get(streamId);
    if (entry == null) {
        log.error("Message received for unknown stream id " + streamId);
    } else {
        // Do stuff with the message (for now just print it)
        ByteBuf content = msg.content();
        if (content.isReadable()) {
            int contentLength = content.readableBytes();
            byte[] arr = new byte[contentLength];
            content.readBytes(arr);
            log.debug(new String(arr, 0, contentLength, CharsetUtil.UTF_8));
        }
        entry.getValue().setSuccess();
    }
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
    HttpHeaders headers = msg.headers();
    Integer streamId = headers.getInt(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
    if (streamId == null) {
        if (LOGGER.isWarnEnabled()) {
            LOGGER.warn("HttpResponseHandler unexpected message received: {}, data is {}", msg.toString(),
                NettyHelper.toString(msg.content()));
        }
        return;
    }

    Entry<ChannelFuture, AbstractHttpClientHandler> entry = removePromise(streamId);
    if (entry == null) {
        if (LOGGER.isWarnEnabled()) {
            LOGGER.warn("Message received for unknown stream id {}, msg is {}, data is {}", streamId,
                msg.toString(), NettyHelper.toString(msg.content()));
        }
    } else {
        final AbstractHttpClientHandler callback = entry.getValue();
        callback.receiveHttpResponse(msg);
    }
}
 

protected FullHttpRequest convertToHttpRequest(SofaRequest request) {
    HttpScheme scheme = SslContextBuilder.SSL ? HttpScheme.HTTPS : HttpScheme.HTTP;
    AsciiString hostName = new AsciiString(providerInfo.getHost() + ':' + providerInfo.getPort());
    String url = "/" + request.getTargetServiceUniqueName() + "/" + request.getMethodName();
    if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("send request to url :{}", url);
    }

    // Create a simple POST request with a body.
    FullHttpRequest httpRequest = new DefaultFullHttpRequest(HTTP_1_1, POST, url,
        wrappedBuffer(request.getData().array()));
    HttpHeaders headers = httpRequest.headers();
    addToHeader(headers, HttpHeaderNames.HOST, hostName);
    addToHeader(headers, HttpConversionUtil.ExtensionHeaderNames.SCHEME.text(), scheme.name());
    addToHeader(headers, HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP);
    addToHeader(headers, HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.DEFLATE);
    addToHeader(headers, RemotingConstants.HEAD_SERIALIZE_TYPE,
        SerializerFactory.getAliasByCode(request.getSerializeType()));
    addToHeader(headers, RemotingConstants.HEAD_TARGET_APP, request.getTargetAppName());
    Map<String, Object> requestProps = request.getRequestProps();
    if (requestProps != null) {
        // <String, Object> 转扁平化 <String, String>
        flatCopyTo("", requestProps, headers);
    }
    return httpRequest;
}
 

/**
 * Writes the given HTTP/1 response to the given stream. Marks the response status metric. Closes
 * the stream after writing the response.
 */
private void writeResponse(ChannelHandlerContext ctx, int streamId, HttpResponse h1Response) {
  markResponseStatus(ctx, h1Response.status());

  // Convert and validate headers.
  Http2Headers headers = HttpConversionUtil.toHttp2Headers(h1Response, true);

  Optional<ByteBuf> body = Optional.empty();
  if (h1Response instanceof FullHttpResponse) {
    ByteBuf content = ((FullHttpResponse) h1Response).content();
    if (content.readableBytes() > 0) {
      body = Optional.of(content);
    }
  }

  writeResponse(ctx, streamId, headers, body);
}
 

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

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

    if (msg instanceof HttpResponse) {
        HttpResponse response = (HttpResponse) msg;
        if (!response.headers().contains(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text())) {
            if (streams.isEmpty()) {
                throw new IllegalStateException("No active streams");
            }
            response.headers().add(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text(),
                                   streams.poll());
        }
    }

    ctx.write(msg, promise);
}
 

@Test(description = "Test API invocation with an HTTP/1.1 request via insecure connection sending to HTTP/2.0 BE")
public void testHTTP1RequestsViaInsecureConnectionWithHTTP2BE() throws Exception {
    Map<String, String> headers = new HashMap<>();
    headers.put(HttpHeaderNames.AUTHORIZATION.toString(), "Bearer " + jwtTokenProd);
    headers.put(HttpHeaderNames.HOST.toString(), "127.0.0.1:9590");
    headers.put(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text().toString(), HttpScheme.HTTP.toString());
    headers.put(HttpHeaderNames.ACCEPT_ENCODING.toString(), HttpHeaderValues.GZIP.toString());
    headers.put(HttpHeaderNames.ACCEPT_ENCODING.toString(), HttpHeaderValues.DEFLATE.toString());
    org.wso2.micro.gateway.tests.util.HttpResponse response = HttpClientRequest
            .doGet(getServiceURLHttp("/pizzashack/1.0.0/menu"), headers);
    log.info("Response: " + response.getResponseMessage() + " , " + response.getResponseCode());
}
 

@Test(description = "Test API invocation with an HTTP/1.1 request via secure connection sending to HTTP/2.0 BE")
public void testHTTP1RequestsViaSecureConnectionWithHTTP2BE() throws Exception {
    Map<String, String> headers = new HashMap<>();
    headers.put(HttpHeaderNames.AUTHORIZATION.toString(), "Bearer " + jwtTokenProd);
    headers.put(HttpHeaderNames.HOST.toString(), "127.0.0.1:9595");
    headers.put(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text().toString(), HttpScheme.HTTP.toString());
    headers.put(HttpHeaderNames.ACCEPT_ENCODING.toString(), HttpHeaderValues.GZIP.toString());
    headers.put(HttpHeaderNames.ACCEPT_ENCODING.toString(), HttpHeaderValues.DEFLATE.toString());
    org.wso2.micro.gateway.tests.util.HttpResponse response = HttpClientRequest
            .doGet(getServiceURLHttp("/pizzashack/1.0.0/menu"), headers);
    log.info("Response: " + response.getResponseMessage() + " , " + response.getResponseCode());
}
 

private FullHttpRequest buildHttpRequest(String url) {
    // Create a simple POST request with a body.
    EchoRequest request = EchoRequest.newBuilder().setGroup(Group.A).setName("xxx").build();
    FullHttpRequest httpRequest = new DefaultFullHttpRequest(HTTP_1_1, POST, url,
        wrappedBuffer(request.toByteArray()));
    HttpHeaders headers = httpRequest.headers();
    addToHeader(headers, HttpHeaderNames.HOST, "127.0.0.1");
    addToHeader(headers, HttpConversionUtil.ExtensionHeaderNames.SCHEME.text(), "HTTP");
    addToHeader(headers, HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP);
    addToHeader(headers, HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.DEFLATE);
    return httpRequest;
}
 

private void onHeadersRead(Http2HeadersFrame headersFrame, ChannelHandlerContext ctx) throws Http2Exception {

        HttpResponse httpResponse = HttpConversionUtil.toHttpResponse(headersFrame.stream().id(), headersFrame.headers(), true);
        ctx.fireChannelRead(httpResponse);

        if (HttpStatusFamily.of(httpResponse.status().code()) == HttpStatusFamily.SERVER_ERROR) {
            fireConnectionExceptionForServerError(ctx);
        }
    }
 

@Test
public void adapt_h2Request_addsSchemeExtension() {
    SdkHttpRequest request = SdkHttpRequest.builder()
            .uri(URI.create("http://localhost:12345/foo/bar/baz"))
            .putRawQueryParameter("foo", "bar")
            .putRawQueryParameter("bar", "baz")
            .putHeader("header1", "header1val")
            .putHeader("header2", "header2val")
            .method(SdkHttpMethod.GET)
            .build();

    HttpRequest adapted = h2Adapter.adapt(request);

    assertThat(adapted.headers().getAll(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text())).containsExactly("http");
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof Http2Settings) {
        settingsPromise.setSuccess(null);
        return;
    }

    if (msg instanceof FullHttpResponse) {
        final FullHttpResponse res = (FullHttpResponse) msg;
        final Integer streamId = res.headers().getInt(
                HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
        if (streamId == null) {
            responsePromise.tryFailure(new AssertionError("message without stream ID: " + msg));
            return;
        }

        if (streamId == 1) {
            // Response to the upgrade request, which is OK to ignore.
            return;
        }

        if (streamId != 3) {
            responsePromise.tryFailure(new AssertionError("unexpected stream ID: " + msg));
            return;
        }

        responsePromise.setSuccess(res.content().retain());
        return;
    }

    throw new IllegalStateException("unexpected message type: " + msg.getClass().getName());
}
 

/**
 * Throws a RuntimeException if the underlying status cannot be converted to an HttpResponseStatus
 */
@Override
public HttpResponseStatus status() {
  try {
    return HttpConversionUtil.parseStatus(delegate.status());
  } catch (Http2Exception e) {
    throw new RuntimeException(e);
  }
}
 

public HttpResponseStatus status() {
  try {
    return HttpConversionUtil.parseStatus(delegate.status());
  } catch (Http2Exception e) {
    throw new RuntimeException(e);
  }
}
 

/** Return an Http1 Headers object based on the values in the underlying Http2Headers object. */
@Override
public HttpHeaders http1Headers(boolean isTrailer, boolean isRequest) {
  try {
    HttpHeaders headers = new DefaultHttpHeaders();
    HttpConversionUtil.addHttp2ToHttpHeaders(
        -1, delegate, headers, HttpVersion.HTTP_1_1, isTrailer, isRequest);
    return headers;
  } catch (Http2Exception e) {
    throw new RuntimeException(e);
  }
}
 

public static FullHttpRequest createGetRequest(String host, int port) {
    FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.valueOf("HTTP/2.0"), HttpMethod.GET, "/", Unpooled.EMPTY_BUFFER);
    request.headers()
        .add(HttpHeaderNames.HOST, new String(host + ":" + port));
    request.headers()
        .add(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text(), HttpScheme.HTTPS);
    request.headers()
        .add(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP);
    request.headers()
        .add(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.DEFLATE);
    return request;
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
    Integer streamId = msg.headers()
        .getInt(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
    if (streamId == null) {
        logger.error("HttpResponseHandler unexpected message received: " + msg);
        return;
    }

    MapValues value = streamidMap.get(streamId);

    if (value == null) {
        logger.error("Message received for unknown stream id " + streamId);
        ctx.close();
    } else {
        ByteBuf content = msg.content();
        if (content.isReadable()) {
            int contentLength = content.readableBytes();
            byte[] arr = new byte[contentLength];
            content.readBytes(arr);
            String response = new String(arr, 0, contentLength, CharsetUtil.UTF_8);
            logger.info("Response from Server: "+ (response));
            value.setResponse(response);
        }
        
        value.getPromise()
            .setSuccess();
    }
}
 

private static String streamId(FullHttpRequest request) {
    return request.headers().get(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
}
 

private void writeRequest(HttpToHttp2OutboundAdapter adapter, ChannelPromise promise) {
    FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "/", Unpooled.wrappedBuffer(new byte[16]));
    request.headers().add(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text(), "http");
    adapter.write(ctx, request, promise);
}
 

final boolean isHttp2() {
	return requestHeaders().contains(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text());
}
 

@Override
public Http2Headers http2Headers() {
  return HttpConversionUtil.toHttp2Headers(delegate, true);
}
 

private HttpResponse buildHttpResponse(final HttpResponseMessage zuulResp) {
    final HttpRequestInfo zuulRequest = zuulResp.getInboundRequest();
    HttpVersion responseHttpVersion;
    final String inboundProtocol = zuulRequest.getProtocol();
    if (inboundProtocol.startsWith("HTTP/1")) {
        responseHttpVersion = HttpVersion.valueOf(inboundProtocol);
    }
    else {
        // Default to 1.1. We do this to cope with HTTP/2 inbound requests.
        responseHttpVersion = HttpVersion.HTTP_1_1;
    }

    // Create the main http response to send, with body.
    final DefaultHttpResponse nativeResponse = new DefaultHttpResponse(responseHttpVersion,
            HttpResponseStatus.valueOf(zuulResp.getStatus()), false, false);

    // Now set all of the response headers - note this is a multi-set in keeping with HTTP semantics
    final HttpHeaders nativeHeaders = nativeResponse.headers();
    for (Header entry : zuulResp.getHeaders().entries()) {
        nativeHeaders.add(entry.getKey(), entry.getValue());
    }

    // Netty does not automatically add Content-Length or Transfer-Encoding: chunked. So we add here if missing.
    if (! HttpUtil.isContentLengthSet(nativeResponse) && ! HttpUtil.isTransferEncodingChunked(nativeResponse)) {
        nativeResponse.headers().add(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
    }

    final HttpRequest nativeReq = (HttpRequest) zuulResp.getContext().get(CommonContextKeys.NETTY_HTTP_REQUEST);
    if (!closeConnection && HttpUtil.isKeepAlive(nativeReq)) {
        HttpUtil.setKeepAlive(nativeResponse, true);
    } else {
        // Send a Connection: close response header (only needed for HTTP/1.0 but no harm in doing for 1.1 too).
        nativeResponse.headers().set("Connection", "close");
    }

    // TODO - temp hack for http/2 handling.
    if (nativeReq.headers().contains(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text())) {
        String streamId = nativeReq.headers().get(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
        nativeResponse.headers().set(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text(), streamId);
    }

    return nativeResponse;
}