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

/**
 * Replaces the entity body of the message with the specified contents. Encodes the message contents according to charset in the message's
 * Content-Type header, or uses {@link BrowserMobHttpUtil#DEFAULT_HTTP_CHARSET} if none is specified.
 * <b>Note:</b> If the charset of the message is not supported on this platform, this will throw an {@link java.nio.charset.UnsupportedCharsetException}.
 *
 * TODO: Currently this method only works for FullHttpMessages, since it must modify the Content-Length header; determine if this may be applied to chunked messages as well
 *
 * @param message the HTTP message to manipulate
 * @param newContents the new entity body contents
 * @throws java.nio.charset.UnsupportedCharsetException if the charset in the message is not supported on this platform
 */
public static void replaceTextHttpEntityBody(FullHttpMessage message, String newContents) {
    // get the content type for this message so we can encode the newContents into a byte stream appropriately
    String contentTypeHeader = message.headers().get(HttpHeaders.Names.CONTENT_TYPE);

    Charset messageCharset;
    try {
        messageCharset = BrowserMobHttpUtil.readCharsetInContentTypeHeader(contentTypeHeader);
    } catch (UnsupportedCharsetException e) {
        java.nio.charset.UnsupportedCharsetException cause = e.getUnsupportedCharsetExceptionCause() ;
        log.error("Found unsupported character set in Content-Type header '{}' while attempting to replace contents of HTTP message.", contentTypeHeader, cause);

        throw cause;
    }

    if (messageCharset == null) {
        messageCharset = BrowserMobHttpUtil.DEFAULT_HTTP_CHARSET;
        log.warn("No character set declared in HTTP message. Replacing text using default charset {}.", messageCharset);
    }

    byte[] contentBytes = newContents.getBytes(messageCharset);

    replaceBinaryHttpEntityBody(message, contentBytes);
}

/**
 * Extracts the entity body from a FullHttpMessage, according to the character set in the message's Content-Type header. If the Content-Type
 * header is not present or does not specify a charset, assumes the ISO-8859-1 character set (see {@link BrowserMobHttpUtil#DEFAULT_HTTP_CHARSET}).
 *
 * @param httpMessage HTTP message to extract entity body from
 * @return String representation of the entity body
 * @throws java.nio.charset.UnsupportedCharsetException if there is a charset specified in the content-type header, but it is not supported
 */
public static String extractHttpEntityBody(FullHttpMessage httpMessage) {
    Charset charset;
    try {
        charset = getCharsetFromMessage(httpMessage);
    } catch (UnsupportedCharsetException e) {
        // the declared character set is not supported, so it is impossible to decode the contents of the message. log an error and throw an exception
        // to alert the client code.
        java.nio.charset.UnsupportedCharsetException cause = e.getUnsupportedCharsetExceptionCause();

        String contentTypeHeader = HttpHeaders.getHeader(httpMessage, HttpHeaders.Names.CONTENT_TYPE);
        log.error("Cannot retrieve text contents of message because HTTP message declares a character set that is not supported on this platform. Content type header: {}.", contentTypeHeader, cause);

        throw cause;
    }

    return extractHttpEntityBody(httpMessage, charset);
}

/**
 * Provides the Observer with a new item to observe.
 * <p>
 * The {@link com.caricah.iotracah.core.modules.Worker} may call this method 0 or more times.
 * <p>
 * The {@code Observable} will not call this method again after it calls either {@link #onCompleted} or
 * {@link #onError}.
 *
 * @param ioTMessage the item emitted by the Observable
 */
@Override
public void onNext(IOTMessage ioTMessage) {

    if(null == ioTMessage || !Protocol.HTTP.equals(ioTMessage.getProtocol())){
        return;
    }

    log.debug(" HttpServer onNext : message outbound {}", ioTMessage);


    FullHttpMessage mqttMessage = toServerMessage(ioTMessage);

    if(null == mqttMessage){
        log.debug(" HttpServer onNext : ignoring outbound message {}", ioTMessage);
    }else {
        serverImpl.pushToClient(ioTMessage.getConnectionId(), mqttMessage);
    }
    serverImpl.postProcess(ioTMessage);
}

/**
 * Replaces an HTTP entity body with the specified binary contents.
 * TODO: Currently this method only works for FullHttpMessages, since it must modify the Content-Length header; determine if this may be applied to chunked messages as well
 *
 * @param message the HTTP message to manipulate
 * @param newBinaryContents the new entity body contents
 */
public static void replaceBinaryHttpEntityBody(FullHttpMessage message, byte[] newBinaryContents) {
    message.content().resetWriterIndex();
    // resize the buffer if needed, since the new message may be longer than the old one
    message.content().ensureWritable(newBinaryContents.length, true);
    message.content().writeBytes(newBinaryContents);

    // update the Content-Length header, since the size may have changed
    message.headers().set(HttpHeaders.Names.CONTENT_LENGTH, newBinaryContents.length);
}

/**
 * Creates a new instance with the specified parameters.
 *
 * @param version the protocol version
 * @param maxContentLength the maximum length of the message content.
 *        If the length of the message content exceeds this value,
 *        a {@link TooLongFrameException} will be raised.
 * @param messageMap the {@link Map} used to hold partially received messages.
 * @param validateHeaders {@code true} if http headers should be validated
 */
protected SpdyHttpDecoder(SpdyVersion version, int maxContentLength, Map<Integer,
        FullHttpMessage> messageMap, boolean validateHeaders) {
    if (version == null) {
        throw new NullPointerException("version");
    }
    if (maxContentLength <= 0) {
        throw new IllegalArgumentException(
                "maxContentLength must be a positive integer: " + maxContentLength);
    }
    spdyVersion = version.getVersion();
    this.maxContentLength = maxContentLength;
    this.messageMap = messageMap;
    this.validateHeaders = validateHeaders;
}

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

@SuppressWarnings("unchecked")
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    if (msg instanceof FullHttpMessage) {
        FullHttpMessage fullHttpMessage = (FullHttpMessage) msg;
        sourceReqCmsg = setupCarbonMessage(fullHttpMessage);
        notifyRequestListener(sourceReqCmsg, ctx);
        ByteBuf content = ((FullHttpMessage) msg).content();
        sourceReqCmsg.addHttpContent(new DefaultLastHttpContent(content));
        if (handlerExecutor != null) {
            handlerExecutor.executeAtSourceRequestSending(sourceReqCmsg);
        }

    } else if (msg instanceof HttpRequest) {
        HttpRequest httpRequest = (HttpRequest) msg;
        sourceReqCmsg = setupCarbonMessage(httpRequest);
        notifyRequestListener(sourceReqCmsg, ctx);
    } else {
        if (sourceReqCmsg != null) {
            if (msg instanceof HttpContent) {
                HttpContent httpContent = (HttpContent) msg;
                sourceReqCmsg.addHttpContent(httpContent);
                if (Util.isLastHttpContent(httpContent)) {
                    if (handlerExecutor != null) {
                        handlerExecutor.executeAtSourceRequestSending(sourceReqCmsg);
                    }
                }
            }
        }
    }
}

private void processHttpRequest(HttpRequest message, ChannelHandlerContext ctx) throws Exception {

        if (LOGGER.isDebugEnabled()) {
            debugHeadersAndCookies(message);
        }

        // Expect: 100-continue should be handled by HttpObjectAggregator.
        ByteBuf buf = ((FullHttpMessage) message).content();
        m_totalContentLength.addAndGet(buf.readableBytes());
        m_server.processHttpRequest(message, ctx.channel());
    }

@Override
    protected void decode(final ChannelHandlerContext ctx, final HttpRequest msg, final List<Object> out) throws Exception {
        final String uri = msg.uri();
        log.info("-----------------------> URI [{}]", uri);
        if(uri.endsWith("/favicon.ico")) {
            final DefaultFullHttpResponse resp = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, favicon);
            resp.headers().set(HttpHeaders.CONTENT_TYPE, "image/x-icon");
            resp.headers().setInt(HttpHeaders.CONTENT_LENGTH, favSize);             
            ctx.writeAndFlush(resp);
            return;
        }           
        ReferenceCountUtil.retain(msg);
        final ChannelPipeline p = ctx.pipeline();

        final int index = uri.indexOf("/api/");
        final String endpoint = index==-1 ? "" : uri.substring(5);
        if(index != -1 && pureJsonEndPoints.contains(endpoint) ) {
            log.info("Switching to PureJSON handler");
            p.addLast(eventExecutorGroup, "httpToJson", httpToJson);
//          p.addLast("jsonLogger", loggingHandler);
            p.addLast("jsonDecoder", new JsonObjectDecoder(true));
//          p.addLast("jsonLogger", loggingHandler);
            p.addLast("traceHandler", traceHandler);
            p.remove(this);
            if(msg instanceof FullHttpMessage) {
                out.add(msg);
            }

        } else {
            log.info("Switching to Http Request Manager");
            out.add(msg);
            p.addLast(eventExecutorGroup, "requestManager", HttpRequestManager.getInstance());
            p.remove(this);         
        }
    }

/**
 * Creates a new instance with the specified parameters.
 *
 * @param version the protocol version
 * @param maxContentLength the maximum length of the message content.
 *        If the length of the message content exceeds this value,
 *        a {@link TooLongFrameException} will be raised.
 * @param messageMap the {@link Map} used to hold partially received messages.
 */
protected SpdyHttpDecoder(SpdyVersion version, int maxContentLength, Map<Integer, FullHttpMessage> messageMap) {
    if (version == null) {
        throw new NullPointerException("version");
    }
    if (maxContentLength <= 0) {
        throw new IllegalArgumentException(
                "maxContentLength must be a positive integer: " + maxContentLength);
    }
    spdyVersion = version.getVersion();
    this.maxContentLength = maxContentLength;
    this.messageMap = messageMap;
}

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

/**
 * Creates a new instance with the specified parameters.
 *
 * @param version the protocol version
 * @param maxContentLength the maximum length of the message content.
 *        If the length of the message content exceeds this value,
 *        a {@link TooLongFrameException} will be raised.
 * @param messageMap the {@link Map} used to hold partially received messages.
 */
protected SpdyHttpDecoder(SpdyVersion version, int maxContentLength, Map<Integer, FullHttpMessage> messageMap) {
    if (version == null) {
        throw new NullPointerException("version");
    }
    if (maxContentLength <= 0) {
        throw new IllegalArgumentException(
                "maxContentLength must be a positive integer: " + maxContentLength);
    }
    spdyVersion = version.getVersion();
    this.maxContentLength = maxContentLength;
    this.messageMap = messageMap;
}

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

public HttpMessageContents(FullHttpMessage httpMessage) {
    this.httpMessage = httpMessage;
}

protected FullHttpMessage putMessage(int streamId, FullHttpMessage message) {
    return messageMap.put(streamId, message);
}

protected FullHttpMessage getMessage(int streamId) {
    return messageMap.get(streamId);
}

protected FullHttpMessage removeMessage(int streamId) {
    return messageMap.remove(streamId);
}

@Override
public boolean acceptOutboundMessage(Object msg) throws Exception {
    return msg instanceof FullHttpMessage;
}

public HttpServerImpl(Server<FullHttpMessage> internalServer) {
    super(internalServer);
}

@Override
protected ServerInitializer<FullHttpMessage> getServerInitializer(ServerImpl<FullHttpMessage> serverImpl, int connectionTimeout) {
    return new HttpServerInitializer(serverImpl, connectionTimeout);
}

@Override
protected ServerInitializer<FullHttpMessage> getServerInitializer(ServerImpl<FullHttpMessage> serverImpl, int connectionTimeout, SSLHandler sslHandler) {
    return new HttpServerInitializer(serverImpl, connectionTimeout,sslHandler);
}

public HttpServerInitializer(ServerImpl<FullHttpMessage> serverImpl,  int connectionTimeout) {
    super(serverImpl,  connectionTimeout);
}

public HttpServerInitializer(ServerImpl<FullHttpMessage> serverImpl, int connectionTimeout, SSLHandler sslHandler) {
    super(serverImpl, connectionTimeout, sslHandler);
}

public JetstreamServletInputStream(HttpRequest req) {
    m_buf = ((FullHttpMessage) req).content();
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpMessage msg) {
    // Request header 처리.
    if (msg instanceof HttpRequest) {
        this.request = (HttpRequest) msg;

        if (HttpHeaders.is100ContinueExpected(request)) {
            send100Continue(ctx);
        }

        HttpHeaders headers = request.headers();
        if (!headers.isEmpty()) {
            for (Map.Entry<String, String> h : headers) {
                String key = h.getKey();
                if (usingHeader.contains(key)) {
                    reqData.put(key, h.getValue());
                }
            }
        }

        reqData.put("REQUEST_URI", request.getUri());
        reqData.put("REQUEST_METHOD", request.getMethod().name());
    }

    // Request content 처리.
    if (msg instanceof HttpContent) {
        if (msg instanceof LastHttpContent) {
            logger.debug("LastHttpContent message received!!" + request.getUri());

            LastHttpContent trailer = (LastHttpContent) msg;

            readPostData();

            ApiRequest service = ServiceDispatcher.dispatch(reqData);

            try {
                service.executeService();

                apiResult = service.getApiResult();
            }
            finally {
                reqData.clear();
            }

            if (!writeResponse(trailer, ctx)) {
                // If keep-alive is off, close the connection once the
                // content is fully written.
                ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
            }
            reset();
        }
    }
}

protected FullHttpMessage putMessage(int streamId, FullHttpMessage message) {
    return messageMap.put(streamId, message);
}

protected FullHttpMessage getMessage(int streamId) {
    return messageMap.get(streamId);
}

protected FullHttpMessage removeMessage(int streamId) {
    return messageMap.remove(streamId);
}

@Override
public boolean acceptOutboundMessage(Object msg) throws Exception {
    return msg instanceof FullHttpMessage;
}

protected FullHttpMessage putMessage(int streamId, FullHttpMessage message) {
    return messageMap.put(streamId, message);
}

protected FullHttpMessage getMessage(int streamId) {
    return messageMap.get(streamId);
}

protected FullHttpMessage removeMessage(int streamId) {
    return messageMap.remove(streamId);
}

@Override
public boolean acceptOutboundMessage(Object msg) throws Exception {
    return msg instanceof FullHttpMessage;
}

/**
 * Compares two {@link FullHttpMessage} for equivalency.
 *
 * <p>This method is needed because an HTTP message decoded and aggregated from inbound {@link
 * ByteBuf} is of a different class than the one written to the outbound {@link ByteBuf}, and The
 * {@link ByteBuf} implementations that hold the content of the HTTP messages are different, even
 * though the actual content, headers, etc are the same.
 *
 * <p>This method is not type-safe, msg1 & msg2 can be a request and a response, respectively. Do
 * not use this method directly.
 */
private static void assertHttpMessageEquivalent(FullHttpMessage msg1, FullHttpMessage msg2) {
  assertThat(msg1.protocolVersion()).isEqualTo(msg2.protocolVersion());
  assertThat(msg1.content()).isEqualTo(msg2.content());
  assertThat(msg1.headers()).isEqualTo(msg2.headers());
}

/**
 * Creates a new instance.
 *
 * @param version the protocol version
 * @param maxContentLength the maximum length of the message content.
 *        If the length of the message content exceeds this value,
 *        a {@link TooLongFrameException} will be raised.
 */
public SpdyHttpDecoder(SpdyVersion version, int maxContentLength) {
    this(version, maxContentLength, new HashMap<Integer, FullHttpMessage>(), true);
}

/**
 * Creates a new instance.
 *
 * @param version the protocol version
 * @param maxContentLength the maximum length of the message content.
 *        If the length of the message content exceeds this value,
 *        a {@link TooLongFrameException} will be raised.
 * @param validateHeaders {@code true} if http headers should be validated
 */
public SpdyHttpDecoder(SpdyVersion version, int maxContentLength, boolean validateHeaders) {
    this(version, maxContentLength, new HashMap<Integer, FullHttpMessage>(), validateHeaders);
}

/**
 * Creates a new instance with the specified parameters.
 *
 * @param version the protocol version
 * @param maxContentLength the maximum length of the message content.
 *        If the length of the message content exceeds this value,
 *        a {@link TooLongFrameException} will be raised.
 * @param messageMap the {@link Map} used to hold partially received messages.
 */
protected SpdyHttpDecoder(SpdyVersion version, int maxContentLength, Map<Integer, FullHttpMessage> messageMap) {
    this(version, maxContentLength, messageMap, true);
}

/**
  * Implementation is expected to transform a server specific message
  * to an internal message that the iotracah workers can handle.
  * <p>
  * Everything that goes beyond the server to workers and eventers
  * or the other way round.
  *
  * @param serverMessage
  * @return
  */
 @Override
 protected IOTMessage toIOTMessage(FullHttpMessage serverMessage) {
    return httpIOTTransformer.toIOTMessage(serverMessage);
}

/**
 * Implementation transforms the internal message to a server specific message
 * that the server now knows how to handle.
 * <p>
 * At the risk of making iotracah create so many unwanted objects,
 * This would be the best way to just ensure the appropriate plugin separation
 * is maintained.
 *
 * @param internalMessage
 * @return
 */
@Override
protected FullHttpMessage toServerMessage(IOTMessage internalMessage) {
    return iotHttpTransformer.toServerMessage(internalMessage);
}

/**
 * Creates a new instance.
 *
 * @param version the protocol version
 * @param maxContentLength the maximum length of the message content.
 *        If the length of the message content exceeds this value,
 *        a {@link TooLongFrameException} will be raised.
 */
public SpdyHttpDecoder(SpdyVersion version, int maxContentLength) {
    this(version, maxContentLength, new HashMap<Integer, FullHttpMessage>());
}