io.netty.handler.codec.http2.Http2ServerUpgradeCodec #io.netty.handler.codec.http.HttpMessage源码实例Demo

@Override
protected HttpMessage createMessage(final String[] initialLine)
        throws Exception {
    // If the first element of the initial line is a version string then
    // this is a response
    if (versionPattern.matcher(initialLine[0]).matches()) {
        isDecodingRequest = false;
        return new DefaultHttpResponse(RtspVersions.valueOf(initialLine[0]),
            new HttpResponseStatus(Integer.parseInt(initialLine[1]),
                                   initialLine[2]),
            validateHeaders);
    } else {
        isDecodingRequest = true;
        return new DefaultHttpRequest(RtspVersions.valueOf(initialLine[2]),
                RtspMethods.valueOf(initialLine[0]),
                initialLine[1],
                validateHeaders);
    }
}
 

@Override
protected void encodeInitialLine(final ByteBuf buf, final HttpMessage message)
       throws Exception {
    if (message instanceof HttpRequest) {
        HttpRequest request = (HttpRequest) message;
        ByteBufUtil.copy(request.method().asciiName(), buf);
        buf.writeByte(SP);
        buf.writeCharSequence(request.uri(), CharsetUtil.UTF_8);
        buf.writeByte(SP);
        buf.writeCharSequence(request.protocolVersion().toString(), CharsetUtil.US_ASCII);
        ByteBufUtil.writeShortBE(buf, CRLF_SHORT);
    } else if (message instanceof HttpResponse) {
        HttpResponse response = (HttpResponse) message;
        buf.writeCharSequence(response.protocolVersion().toString(), CharsetUtil.US_ASCII);
        buf.writeByte(SP);
        ByteBufUtil.copy(response.status().codeAsText(), buf);
        buf.writeByte(SP);
        buf.writeCharSequence(response.status().reasonPhrase(), CharsetUtil.US_ASCII);
        ByteBufUtil.writeShortBE(buf, CRLF_SHORT);
    } else {
        throw new UnsupportedMessageTypeException("Unsupported type "
                            + StringUtil.simpleClassName(message));
    }
}
 

/**
 * Converts the given HTTP/1.x headers into HTTP/2 headers.
 * The following headers are only used if they can not be found in from the {@code HOST} header or the
 * {@code Request-Line} as defined by <a href="https://tools.ietf.org/html/rfc7230">rfc7230</a>
 * <ul>
 * <li>{@link ExtensionHeaderNames#SCHEME}</li>
 * </ul>
 * {@link ExtensionHeaderNames#PATH} is ignored and instead extracted from the {@code Request-Line}.
 */
public static Http2Headers toHttp2Headers(HttpMessage in, boolean validateHeaders) {
    HttpHeaders inHeaders = in.headers();
    final Http2Headers out = new DefaultHttp2Headers(validateHeaders, inHeaders.size());
    if (in instanceof HttpRequest) {
        HttpRequest request = (HttpRequest) in;
        URI requestTargetUri = URI.create(request.uri());
        out.path(toHttp2Path(requestTargetUri));
        out.method(request.method().asciiName());
        setHttp2Scheme(inHeaders, requestTargetUri, out);

        if (!isOriginForm(requestTargetUri) && !isAsteriskForm(requestTargetUri)) {
            // Attempt to take from HOST header before taking from the request-line
            String host = inHeaders.getAsString(HttpHeaderNames.HOST);
            setHttp2Authority((host == null || host.isEmpty()) ? requestTargetUri.getAuthority() : host, out);
        }
    } else if (in instanceof HttpResponse) {
        HttpResponse response = (HttpResponse) in;
        out.status(response.status().codeAsText());
    }

    // Add the HTTP headers which have not been consumed above
    toHttp2Headers(inHeaders, out);
    return out;
}
 

/**
 * Adds the Via header to specify that the message has passed through the proxy. The specified alias will be
 * appended to the Via header line. The alias may be the hostname of the machine proxying the request, or a
 * pseudonym. From RFC 7230, section 5.7.1:
 * <pre>
 * The received-by portion of the field value is normally the host and
 * optional port number of a recipient server or client that
 * subsequently forwarded the message.  However, if the real host is
 * considered to be sensitive information, a sender MAY replace it with
 * a pseudonym.
 * </pre>
 *
 * @param httpMessage HTTP message to add the Via header to
 * @param alias       the alias to provide in the Via header for this proxy
 */
public static void addVia(HttpMessage httpMessage, String alias) {
    String newViaHeader = new StringBuilder()
            .append(httpMessage.getProtocolVersion().majorVersion())
            .append('.')
            .append(httpMessage.getProtocolVersion().minorVersion())
            .append(' ')
            .append(alias)
            .toString();

    final List<String> vias;
    if (httpMessage.headers().contains(HttpHeaders.Names.VIA)) {
        List<String> existingViaHeaders = httpMessage.headers().getAll(HttpHeaders.Names.VIA);
        vias = new ArrayList<String>(existingViaHeaders);
        vias.add(newViaHeader);
    } else {
        vias = Collections.singletonList(newViaHeader);
    }

    httpMessage.headers().set(HttpHeaders.Names.VIA, vias);
}
 

@Override
public void channelRead(final ChannelHandlerContext ctx, final Object obj)
                        throws Exception {
    if (obj instanceof HttpMessage && !isWebSocket((HttpMessage) obj)) {
        ChannelPipeline pipeline = ctx.pipeline();
        ChannelHandler authHandler = pipeline.get(HTTP_AUTH);
        if (authHandler != null) {
            authHandler = pipeline.remove(HTTP_AUTH);
        } else {
            authHandler = new HttpBasicAuthHandler(
                          this.authenticator,  this.authenticationSettings);
        }
        pipeline.addAfter(AUTHENTICATOR, HTTP_AUTH, authHandler);
        ctx.fireChannelRead(obj);
    } else {
        super.channelRead(ctx, obj);
    }
}
 

@Test
public void argsAreEligibleForLinkingAndUnlinkingDistributedTracingInfo_only_returns_true_for_HttpRequest_or_LastHttpContent() {
    // given
    Object httpRequestMsg = mock(HttpRequest.class);
    Object lastHttpContentMsg = mock(LastHttpContent.class);
    Object httpMessageMsg = mock(HttpMessage.class);

    // expect
    assertThat(handlerSpy.argsAreEligibleForLinkingAndUnlinkingDistributedTracingInfo(
        DO_CHANNEL_READ, ctxMock, httpRequestMsg, null)
    ).isTrue();
    assertThat(handlerSpy.argsAreEligibleForLinkingAndUnlinkingDistributedTracingInfo(
        DO_CHANNEL_READ, ctxMock, lastHttpContentMsg, null)
    ).isTrue();
    assertThat(handlerSpy.argsAreEligibleForLinkingAndUnlinkingDistributedTracingInfo(
        DO_CHANNEL_READ, ctxMock, httpMessageMsg, null)
    ).isFalse();
}
 

@Override
protected HttpMessage newFullBodyMessage(ByteBuf body) {
	HttpResponse res =
			new DefaultFullHttpResponse(version(), status(), body);

	responseHeaders.remove(HttpHeaderNames.TRANSFER_ENCODING);
	if (!HttpResponseStatus.NOT_MODIFIED.equals(status())) {

		if (HttpUtil.getContentLength(nettyResponse, -1) == -1) {
			responseHeaders.setInt(HttpHeaderNames.CONTENT_LENGTH, body.readableBytes());
		}
	}

	res.headers().set(responseHeaders);
	return res;
}
 

@SuppressWarnings("FutureReturnValueIgnored")
private void doTestStatus(HttpResponseStatus status) {
	EmbeddedChannel channel = new EmbeddedChannel();
	HttpServerOperations ops = new HttpServerOperations(
			Connection.from(channel),
			ConnectionObserver.emptyListener(),
			null,
			new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/"),
			null,
			ServerCookieEncoder.STRICT,
			ServerCookieDecoder.STRICT);
	ops.status(status);
	HttpMessage response = ops.newFullBodyMessage(Unpooled.EMPTY_BUFFER);
	assertThat(((FullHttpResponse) response).status().reasonPhrase()).isEqualTo(status.reasonPhrase());
	// "FutureReturnValueIgnored" is suppressed deliberately
	channel.close();
}
 

@Override
public void channelRead(final ChannelHandlerContext ctx, final Object obj) {
    final ChannelPipeline pipeline = ctx.pipeline();
    if (obj instanceof HttpMessage && !WebSocketHandlerUtil.isWebSocket((HttpMessage)obj)) {
        if (null != pipeline.get(PIPELINE_AUTHENTICATOR)) {
            pipeline.remove(PIPELINE_REQUEST_HANDLER);
            final ChannelHandler authenticator = pipeline.get(PIPELINE_AUTHENTICATOR);
            pipeline.remove(PIPELINE_AUTHENTICATOR);
            pipeline.addAfter(PIPELINE_HTTP_RESPONSE_ENCODER, PIPELINE_AUTHENTICATOR, authenticator);
            pipeline.addAfter(PIPELINE_AUTHENTICATOR, PIPELINE_REQUEST_HANDLER, this.httpGremlinEndpointHandler);
        } else {
            pipeline.remove(PIPELINE_REQUEST_HANDLER);
            pipeline.addAfter(PIPELINE_HTTP_RESPONSE_ENCODER, PIPELINE_REQUEST_HANDLER, this.httpGremlinEndpointHandler);
        }
    }
    ctx.fireChannelRead(obj);
}
 

@Override
protected boolean isContentAlwaysEmpty(HttpMessage httpMessage) {
    // The current HTTP Request can be null when this proxy is
    // negotiating a CONNECT request with a chained proxy
    // while it is running as a MITM. Since the response to a
    // CONNECT request does not have any content, we return true.
    if(currentHttpRequest == null) {
        return true;
    } else {
        return ProxyUtils.isHEAD(currentHttpRequest) || super.isContentAlwaysEmpty(httpMessage);
    }
}
 

/**
 * Derives the charset from the Content-Type header in the HttpMessage. If the Content-Type header is not present or does not contain
 * a character set, this method returns the ISO-8859-1 character set. See {@link BrowserUpHttpUtil#readCharsetInContentTypeHeader(String)}
 * for more details.
 *
 * @param httpMessage HTTP message to extract charset from
 * @return the charset associated with the HTTP message, or the default charset if none is present
 * @throws UnsupportedCharsetException if there is a charset specified in the content-type header, but it is not supported
 */
public static Charset getCharsetFromMessage(HttpMessage httpMessage) throws UnsupportedCharsetException {
    String contentTypeHeader = HttpHeaders.getHeader(httpMessage, HttpHeaderNames.CONTENT_TYPE);

    Charset charset = BrowserUpHttpUtil.readCharsetInContentTypeHeader(contentTypeHeader);
    if (charset == null) {
        return BrowserUpHttpUtil.DEFAULT_HTTP_CHARSET;
    }

    return charset;
}
 

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
    if (acceptOutboundMessage(msg)) {
        HttpMessage httpMsg = (HttpMessage) msg;
        if (!httpMsg.headers().contains(SpdyHttpHeaders.Names.STREAM_ID)) {
            httpMsg.headers().setInt(Names.STREAM_ID, currentStreamId);
            // Client stream IDs are always odd
            currentStreamId += 2;
        }
    }
    ctx.write(msg, promise);
}
 

/**
 * Checks if the given HTTP message should be considered as a last SPDY frame.
 *
 * @param httpMessage check this HTTP message
 * @return whether the given HTTP message should generate a <em>last</em> SPDY frame.
 */
private static boolean isLast(HttpMessage httpMessage) {
    if (httpMessage instanceof FullHttpMessage) {
        FullHttpMessage fullMessage = (FullHttpMessage) httpMessage;
        if (fullMessage.trailingHeaders().isEmpty() && !fullMessage.content().isReadable()) {
            return true;
        }
    }

    return false;
}
 

@Override
protected void encode(ChannelHandlerContext ctx, HttpMessage msg, List<Object> out) throws Exception {
    Integer id = ids.poll();
    if (id != null && id.intValue() != NO_ID && !msg.headers().contains(SpdyHttpHeaders.Names.STREAM_ID)) {
        msg.headers().setInt(Names.STREAM_ID, id);
    }

    out.add(ReferenceCountUtil.retain(msg));
}
 

@Override
protected void decode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {
    if (msg instanceof HttpMessage) {
        boolean contains = ((HttpMessage) msg).headers().contains(SpdyHttpHeaders.Names.STREAM_ID);
        if (!contains) {
            ids.add(NO_ID);
        } else {
            ids.add(((HttpMessage) msg).headers().getInt(Names.STREAM_ID));
        }
    } else if (msg instanceof SpdyRstStreamFrame) {
        ids.remove(((SpdyRstStreamFrame) msg).streamId());
    }

    out.add(ReferenceCountUtil.retain(msg));
}
 

@Override
protected HttpMessage createInvalidMessage() {
    if (isDecodingRequest) {
        return new DefaultFullHttpRequest(RtspVersions.RTSP_1_0,
                   RtspMethods.OPTIONS, "/bad-request", validateHeaders);
    } else {
        return new DefaultFullHttpResponse(RtspVersions.RTSP_1_0,
                                           UNKNOWN_STATUS,
                                           validateHeaders);
    }
}
 

@Override
protected boolean isContentAlwaysEmpty(HttpMessage msg) {
    // Unlike HTTP, RTSP always assumes zero-length body if Content-Length
    // header is absent.
    boolean empty = super.isContentAlwaysEmpty(msg);
    if (empty) {
        return true;
    }
    if (!msg.headers().contains(RtspHeaderNames.CONTENT_LENGTH)) {
        return true;
    }
    return empty;
}
 

private Http2Headers toHttp2Headers(final HttpMessage msg) {
    if (msg instanceof HttpRequest) {
        msg.headers().set(
                HttpConversionUtil.ExtensionHeaderNames.SCHEME.text(),
                scheme.name());
    }

    return HttpConversionUtil.toHttp2Headers(msg, validateHeaders);
}
 

/**
 * Derives the charset from the Content-Type header in the HttpMessage. If the Content-Type header is not present or does not contain
 * a character set, this method returns the ISO-8859-1 character set. See {@link BrowserMobHttpUtil#readCharsetInContentTypeHeader(String)}
 * for more details.
 *
 * @param httpMessage HTTP message to extract charset from
 * @return the charset associated with the HTTP message, or the default charset if none is present
 * @throws UnsupportedCharsetException if there is a charset specified in the content-type header, but it is not supported
 */
public static Charset getCharsetFromMessage(HttpMessage httpMessage) throws UnsupportedCharsetException {
    String contentTypeHeader = HttpHeaders.getHeader(httpMessage, HttpHeaders.Names.CONTENT_TYPE);

    Charset charset = BrowserMobHttpUtil.readCharsetInContentTypeHeader(contentTypeHeader);
    if (charset == null) {
        return BrowserMobHttpUtil.DEFAULT_HTTP_CHARSET;
    }

    return charset;
}
 

@Override
protected boolean isContentAlwaysEmpty(HttpMessage httpMessage) {
    // The current HTTP Request can be null when this proxy is
    // negotiating a CONNECT request with a chained proxy
    // while it is running as a MITM. Since the response to a
    // CONNECT request does not have any content, we return true.
    if(currentHttpRequest == null) {
        return true;
    } else {
        return ProxyUtils.isHEAD(currentHttpRequest) || super.isContentAlwaysEmpty(httpMessage);
    }
}
 

/**
 * Returns true if the HTTP message cannot contain an entity body, according to the HTTP spec. This code is taken directly
 * from {@link io.netty.handler.codec.http.HttpObjectDecoder#isContentAlwaysEmpty(HttpMessage)}.
 *
 * @param msg HTTP message
 * @return true if the HTTP message is always empty, false if the message <i>may</i> have entity content.
 */
public static boolean isContentAlwaysEmpty(HttpMessage msg) {
    if (msg instanceof HttpResponse) {
        HttpResponse res = (HttpResponse) msg;
        int code = res.getStatus().code();

        // Correctly handle return codes of 1xx.
        //
        // See:
        //     - http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html Section 4.4
        //     - https://github.com/netty/netty/issues/222
        if (code >= 100 && code < 200) {
            // According to RFC 7231, section 6.1, 1xx responses have no content (https://tools.ietf.org/html/rfc7231#section-6.2):
            //   1xx responses are terminated by the first empty line after
            //   the status-line (the empty line signaling the end of the header
            //        section).

            // Hixie 76 websocket handshake responses contain a 16-byte body, so their content is not empty; but Hixie 76
            // was a draft specification that was superceded by RFC 6455. Since it is rarely used and doesn't conform to
            // RFC 7231, we do not support or make special allowance for Hixie 76 responses.
            return true;
        }

        switch (code) {
            case 204:
            case 205:
            case 304:
                return true;
        }
    }
    return false;
}
 

@Override
protected HttpMessage createMessage(String[] initialLine) throws Exception {

    return isDecodingRequest() ? new DefaultHttpRequest(
            HttpVersion.valueOf(initialLine[2]),
            HttpMethod.valueOf(initialLine[0]), initialLine[1], validateHeaders) :
            new DefaultHttpResponse(
                    HttpVersion.valueOf(initialLine[0]),
                    HttpResponseStatus.valueOf(Integer.parseInt(initialLine[1]), initialLine[2]), validateHeaders);
}
 

@Override
protected boolean isContentAlwaysEmpty(HttpMessage httpMessage) {
    // The current HTTP Request can be null when this proxy is
    // negotiating a CONNECT request with a chained proxy
    // while it is running as a MITM. Since the response to a
    // CONNECT request does not have any content, we return true.
    if(currentHttpRequest == null) {
        return true;
    } else {
        return ProxyUtils.isHEAD(currentHttpRequest) || super.isContentAlwaysEmpty(httpMessage);
    }
}
 

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    LOGGER.info("[Client ({})] => [Server ({})] : {}",
                connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                msg);
    if (msg instanceof FullHttpRequest) {
        HttpMessage httpMessage = (HttpRequest) msg;
        httpMessage.headers().add(ExtensionHeaderNames.SCHEME.text(), "https");
    } else if (msg instanceof HttpObject) {
        throw new IllegalStateException("Cannot handle message: " + msg.getClass());
    }

    ctx.writeAndFlush(msg, promise);
}
 

@Test
public void channelRead_does_nothing_if_msg_is_not_HttpRequest_or_LastHttpContent() throws Exception {
    // given
    HttpMessage ignoredMsgMock = mock(HttpMessage.class);

    // when
    handler.channelRead(ctxMock, ignoredMsgMock);

    // then
    verify(ctxMock).fireChannelRead(ignoredMsgMock); // the normal continuation behavior from the super class.
    verifyNoMoreInteractions(ctxMock); // nothing else should have happened related to the ctx.
    verifyNoInteractions(stateMock);
    verifyNoInteractions(metricsListenerMock);
}
 

/**
 * Derives the charset from the Content-Type header in the HttpMessage. If the Content-Type header is not present or does not contain
 * a character set, this method returns the ISO-8859-1 character set. See {@link BrowserMobHttpUtil#readCharsetInContentTypeHeader(String)}
 * for more details.
 *
 * @param httpMessage HTTP message to extract charset from
 * @return the charset associated with the HTTP message, or the default charset if none is present
 * @throws UnsupportedCharsetException if there is a charset specified in the content-type header, but it is not supported
 */
public static Charset getCharsetFromMessage(HttpMessage httpMessage) throws UnsupportedCharsetException {
    String contentTypeHeader = HttpHeaders.getHeader(httpMessage, HttpHeaders.Names.CONTENT_TYPE);

    Charset charset = BrowserMobHttpUtil.readCharsetInContentTypeHeader(contentTypeHeader);
    if (charset == null) {
        return BrowserMobHttpUtil.DEFAULT_HTTP_CHARSET;
    }

    return charset;
}
 

@Override
protected HttpMessage newFullBodyMessage(ByteBuf body) {
	HttpRequest request = new DefaultFullHttpRequest(version(), method(), uri(), body);

	requestHeaders.setInt(HttpHeaderNames.CONTENT_LENGTH, body.readableBytes());
	requestHeaders.remove(HttpHeaderNames.TRANSFER_ENCODING);

	request.headers()
	       .set(requestHeaders);
	return request;
}
 

final Mono<Void> send() {
	if (!channel().isActive()) {
		return Mono.error(AbortedException.beforeSend());
	}
	if (markSentHeaderAndBody()) {
		HttpMessage request = newFullBodyMessage(Unpooled.EMPTY_BUFFER);
		return FutureMono.deferFuture(() -> channel().writeAndFlush(request));
	}
	else {
		return Mono.empty();
	}
}
 

@Override
public Mono<Void> send() {
	if (markSentHeaderAndBody()) {
		HttpMessage response = newFullBodyMessage(EMPTY_BUFFER);
		return FutureMono.deferFuture(() -> channel().writeAndFlush(response));
	}
	else {
		return Mono.empty();
	}
}
 

/**
 *  Extract headers from {@link io.netty.handler.codec.http.HttpMessage} and put in temporary
 *  headers. Headers are stored as multi-map because given the same key, it can have more than
 *  one values.
 *  @param httpMessage netty http message
 * */
public void addHeaders(HttpMessage httpMessage) {
  if (httpMessage.headers() == null) {
    return;
  }
  for (String name : httpMessage.headers().names()) {
    _headers.putAll(name, httpMessage.headers().getAll(name));
  }
}