Java 类io.netty.handler.codec.http.HttpMessage 实例源码

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

private void handleUploadMessage(HttpMessage httpMsg, Message uploadMessage) throws IOException{
    if (httpMsg instanceof HttpContent) { 
           HttpContent chunk = (HttpContent) httpMsg;
           decoder.offer(chunk); 
           try {
               while (decoder.hasNext()) {
                   InterfaceHttpData data = decoder.next();
                   if (data != null) {
                       try { 
                        handleUploadFile(data, uploadMessage);
                       } finally {
                           data.release();
                       }
                   }
               }
           } catch (EndOfDataDecoderException e1) { 
            //ignore
           }

           if (chunk instanceof LastHttpContent) {  
            resetUpload();
           }
       }
}

@Override
protected boolean isContentAlwaysEmpty(HttpMessage httpMessage) {
    if (httpMessage instanceof HttpResponse) {
        // Identify our current request
        identifyCurrentRequest();
    }

    // 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 HttpMethod.HEAD.equals(currentHttpRequest.getMethod()) ?
                true : super.isContentAlwaysEmpty(httpMessage);
    }
}

/**
 * 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);
}

@Test
public void testAddNewViaHeaderToExistingViaHeader() {
    String hostname = "hostname";

    HttpMessage httpMessage = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/endpoint");
    httpMessage.headers().add(HttpHeaders.Names.VIA, "1.1 otherproxy");
    ProxyUtils.addVia(httpMessage, hostname);

    List<String> viaHeaders = httpMessage.headers().getAll(HttpHeaders.Names.VIA);
    assertThat(viaHeaders, hasSize(2));

    assertEquals("1.1 otherproxy", viaHeaders.get(0));

    String expectedViaHeader = "1.1 " + hostname;
    assertEquals(expectedViaHeader, viaHeaders.get(1));
}

/**
 * Adds the Via header to specify that the message has passed through the
 * proxy.
 * 
 * @param msg
 *            The HTTP message.
 */
public static void addVia(final HttpMessage msg) {
    final StringBuilder sb = new StringBuilder();
    sb.append(msg.getProtocolVersion().majorVersion());
    sb.append(".");
    sb.append(msg.getProtocolVersion().minorVersion());
    sb.append(".");
    sb.append(hostName);
    final List<String> vias;
    if (msg.headers().contains(HttpHeaders.Names.VIA)) {
        vias = msg.headers().getAll(HttpHeaders.Names.VIA);
        vias.add(sb.toString());
    } else {
        vias = Arrays.asList(sb.toString());
    }
    msg.headers().set(HttpHeaders.Names.VIA, vias);
}

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

/**
 * Convert the OData Response to Netty Response
 * @param response
 * @param odResponse
 */
static void convertToHttp(final HttpResponse response, final ODataResponse odResponse) {
   response.setStatus(HttpResponseStatus.valueOf(odResponse.getStatusCode()));

   for (Entry<String, List<String>> entry : odResponse.getAllHeaders().entrySet()) {
     for (String headerValue : entry.getValue()) {
       ((HttpMessage)response).headers().add(entry.getKey(), headerValue);
     }
   }

   if (odResponse.getContent() != null) {
     copyContent(odResponse.getContent(), response);
   } else if (odResponse.getODataContent() != null) {
     writeContent(odResponse, response);
   }
 }

/**
 * Adds the Via header to specify that the message has passed through the
 * proxy.
 * 
 * @param msg
 *            The HTTP message.
 */
public static void addVia(final HttpMessage msg) {
    final StringBuilder sb = new StringBuilder();
    sb.append(msg.getProtocolVersion().majorVersion());
    sb.append(".");
    sb.append(msg.getProtocolVersion().minorVersion());
    sb.append(".");
    sb.append(hostName);
    final List<String> vias;
    if (msg.headers().contains(HttpHeaders.Names.VIA)) {
        vias = msg.headers().getAll(HttpHeaders.Names.VIA);
        vias.add(sb.toString());
    } else {
        vias = Arrays.asList(sb.toString());
    }
    msg.headers().set(HttpHeaders.Names.VIA, vias);
}

/**
 * Adds the Via header to specify that the message has passed through the
 * proxy.
 * 
 * @param msg
 *            The HTTP message.
 */
public static void addVia(final HttpMessage msg) {
    final StringBuilder sb = new StringBuilder();
    sb.append(msg.getProtocolVersion().majorVersion());
    sb.append(".");
    sb.append(msg.getProtocolVersion().minorVersion());
    sb.append(".");
    sb.append(hostName);
    final List<String> vias;
    if (msg.headers().contains(HttpHeaders.Names.VIA)) {
        vias = msg.headers().getAll(HttpHeaders.Names.VIA);
        vias.add(sb.toString());
    } else {
        vias = Arrays.asList(sb.toString());
    }
    msg.headers().set(HttpHeaders.Names.VIA, vias);
}

public static SocketAddress resolveClientIpByRemoteAddressHeader(HttpMessage message, String headerName) {
  SocketAddress clientIp = null;
  if (headerName != null && !headerName.trim().isEmpty()) {
    String ip = null;
    try {
      ip = message.headers().get(headerName);
      if (ip != null) {
        ip = ip.split(",")[0]; // to handle multiple proxies case (e.g. X-Forwarded-For: client, proxy1, proxy2)
        clientIp = new InetSocketAddress(InetAddress.getByName(ip), 0);
      }
    } catch (Exception e) {
      log.warn("Failed to parse IP address: {} from http header: {}", ip, headerName);
    }
  }
  return clientIp;
}

@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.
    verifyZeroInteractions(stateMock);
    verifyZeroInteractions(metricsListenerMock);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object e) {
    if (e instanceof HttpMessage) {
        HttpMessage m = (HttpMessage) e;
        // for test there is no Content-Encoding header so just hard
        // coding value
        // for verification
        m.headers().set("X-Original-Content-Encoding", "<original encoding>");
    }
    ctx.fireChannelRead(e);
}

/**
 * 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
public Mono<Void> send() {
    if (markSentHeaderAndBody()) {
        HttpMessage request = newFullEmptyBodyMessage();
        return FutureMono.deferFuture(() -> channel().writeAndFlush(request));
    }
    else {
        return Mono.empty();
    }
}

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

    request.headers()
           .set(requestHeaders.remove(HttpHeaderNames.TRANSFER_ENCODING)
                              .setInt(HttpHeaderNames.CONTENT_LENGTH, 0));
    return request;
}

@Override
protected HttpMessage newFullEmptyBodyMessage() {
    HttpResponse res =
            new DefaultFullHttpResponse(version(), status(), EMPTY_BUFFER);

    if (!HttpMethod.HEAD.equals(method())) {
        res.headers()
           .set(responseHeaders.remove(HttpHeaderNames.TRANSFER_ENCODING)
                               .setInt(HttpHeaderNames.CONTENT_LENGTH, 0));
    }
    else {
        res.headers().set(responseHeaders);
    }
    return res;
}

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

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
        throws Exception {
    if (msg instanceof ByteBuf) {
        offerByteBuf(ctx, msg, promise);
    }
    else if (msg instanceof HttpMessage) {
        offerHttpMessage(msg, promise);
    }
    else {
        super.write(ctx, msg, promise);
    }
}

public NettyOutbound sendHeaders() {
    if (markSentHeaders()) {
        if (HttpUtil.isContentLengthSet(outboundHttpMessage())) {
            outboundHttpMessage().headers()
                                 .remove(HttpHeaderNames.TRANSFER_ENCODING);
        }

        HttpMessage message;
        if (!HttpUtil.isTransferEncodingChunked(outboundHttpMessage())
                && (!HttpUtil.isContentLengthSet(outboundHttpMessage()) ||
        HttpUtil.getContentLength(outboundHttpMessage(), 0) == 0)) {
            if(isKeepAlive() && markSentBody()){
                message = newFullEmptyBodyMessage();
            }
            else {
                markPersistent(false);
                message = outboundHttpMessage();
            }
        }
        else {
            message = outboundHttpMessage();
        }
        return then(FutureMono.deferFuture(() -> {
            if(!channel().isActive()){
                throw new AbortedException();
            }
            return channel().writeAndFlush(message);
        }));
    }
    else {
        return this;
    }
}

private Message decodeHeaders(HttpMessage httpMsg){
    Message msg = new Message();
    Iterator<Entry<String, String>> iter = httpMsg.headers().iterator();
    while (iter.hasNext()) {
        Entry<String, String> e = iter.next();
        if(e.getKey().equalsIgnoreCase(Message.CONTENT_TYPE)){ //encoding and type
            String[] typeInfo = httpContentType(e.getValue());
            msg.setHeader(Message.CONTENT_TYPE, typeInfo[0]); 
            if(msg.getHeader(Message.ENCODING) == null) {
                msg.setHeader(Message.ENCODING, typeInfo[1]);
            }
        } else {
            msg.setHeader(e.getKey().toLowerCase(), e.getValue());
        } 
    }  

    if (httpMsg instanceof HttpRequest) {
        HttpRequest req = (HttpRequest) httpMsg;
        msg.setMethod(req.getMethod().name());
        msg.setUrl(req.getUri());
    } else if (httpMsg instanceof HttpResponse) {
        HttpResponse resp = (HttpResponse) httpMsg;
        int status = resp.getStatus().code();
        msg.setStatus(status);
    }
    return msg;
}

private static String getWebSocketLocation(HttpMessage req, ChannelHandlerContext ctx) {
    String location = req.headers().get(HttpHeaders.Names.HOST) + WEBSOCKET_PATH;
    if (ctx.pipeline().get(SslHandler.class) != null) {
        return "wss://" + location;
    } else {
        return "ws://" + location;
    }
}

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

private void handleHttpHeaders(ChannelHandlerContext ctx, HttpMessage httpMessage) {
  // Figure out if the channel needs to be closed after responding to this request
  if (httpMessage
      .headers()
      .containsValue(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE, true)) {
    ctx.channel().attr(ChannelAttributes.CLOSE_CONNECTION).set(true);
  }
}

@Test
public void testEncodeResponseOfUnknownSize() throws Exception {
  when(ctx.write(any())).thenReturn(mock(ChannelFuture.class));
  TransportBody body = TransportBodySubject.create();
  TransportResponse transportResponse = DefaultTransportResponse.builder().body(body).build();
  encoder.write(ctx, transportResponse, promise);

  verify(ctx)
      .write(
          Mockito.<HttpMessage>argThat(
              httpResponse -> httpResponse.headers().contains(TRANSFER_ENCODING, CHUNKED, true)));
}

@Test
public void testEncodeLargeResponse() throws Exception {
  when(ctx.write(any())).thenReturn(mock(ChannelFuture.class));
  TransportBody body = TransportBodySubject.create(TransportBody.MAX_BUFFERED_BODY_SIZE + 1);
  TransportResponse transportResponse = DefaultTransportResponse.builder().body(body).build();
  encoder.write(ctx, transportResponse, promise);

  verify(ctx)
      .write(
          Mockito.<HttpMessage>argThat(
              httpResponse -> httpResponse.headers().contains(TRANSFER_ENCODING, CHUNKED, true)));
}

@Test
public void testEncodeRequestOfUnknownSize() throws Exception {
  when(ctx.write(any())).thenReturn(mock(ChannelFuture.class));
  TransportBody body = TransportBodySubject.create();
  TransportRequest transportRequest = newTransportRequest(body);
  encoder.write(ctx, transportRequest, promise);

  verify(ctx)
      .write(
          Mockito.<HttpMessage>argThat(
              httpRequest -> httpRequest.headers().contains(TRANSFER_ENCODING, CHUNKED, true)));
}

@Test
public void testEncodeLargeRequest() throws Exception {
  when(ctx.write(any())).thenReturn(mock(ChannelFuture.class));
  TransportBody body = TransportBodySubject.create(TransportBody.MAX_BUFFERED_BODY_SIZE + 1);
  TransportRequest transportRequest = newTransportRequest(body);
  encoder.write(ctx, transportRequest, promise);

  verify(ctx)
      .write(
          Mockito.<HttpMessage>argThat(
              httpRequest -> httpRequest.headers().contains(TRANSFER_ENCODING, CHUNKED, true)));
}

/**
 * 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;
}

@Test
public void testAddNewViaHeader() {
    String hostname = "hostname";

    HttpMessage httpMessage = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/endpoint");
    ProxyUtils.addVia(httpMessage, hostname);

    List<String> viaHeaders = httpMessage.headers().getAll(HttpHeaders.Names.VIA);
    assertThat(viaHeaders, hasSize(1));

    String expectedViaHeader = "1.1 " + hostname;
    assertEquals(expectedViaHeader, viaHeaders.get(0));
}

/**
 *  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()) {
    for (String value : httpMessage.headers().getAll(name)) {
      if (!_headers.containsEntry(name, value)) {
        _headers.put(name, value);
      }
    }
  }
}

/**
 * Add headers from http message and also check if uri is properly set.
 * If not, we need check host header and construct uri using relative path
 * and host name.
 *
 * @param httpMessage netty http message
 * */
@Override
public void addHeaders(HttpMessage httpMessage) {
  super.addHeaders(httpMessage);
  if (_uri == null) {
    String hostName = getHeader(HttpHeaders.Names.HOST);
    if (!Strings.isNullOrEmpty(hostName)) {
      try {
        _uri = new URI(String.format("https://%s%s", hostName, _path));
      } catch (URISyntaxException e) {
        throw new IllegalStateException("Invalid URI in underlying request", e);
      }
    }
  }
}

@Override
protected boolean isContentAlwaysEmpty(HttpMessage httpMessage) {
    if (httpMessage instanceof HttpResponse) {
        // Identify our current request
        identifyCurrentRequest();
    }

    return HttpMethod.HEAD.equals(currentHttpRequest.getMethod()) ?
            true : super.isContentAlwaysEmpty(httpMessage);
}

@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)) {
            SpdyHttpHeaders.setStreamId(httpMsg, 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)) {
        SpdyHttpHeaders.setStreamId(msg, 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(SpdyHttpHeaders.getStreamId((HttpMessage) msg));
        }
    } else if (msg instanceof SpdyRstStreamFrame) {
        ids.remove(((SpdyRstStreamFrame) msg).streamId());
    }

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

@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(RtspHeaders.Names.CONTENT_LENGTH)) {
        return true;
    }
    return empty;
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, final Object msg) throws Exception {
    if (msg instanceof HttpMessage) {
        logger.debug("Adding no cache headers");
        HttpHeaders headers = ((HttpMessage) msg).headers();
        headers.set("Cache-Control", "no-cache, no-store, must-revalidate");
        headers.set("Pragma", "no-cache");
        headers.set("Expires", "0");
    }
    inboundChannel.writeAndFlush(msg);
}