<commit_before>/*
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.reactivex.netty.protocol.http.server;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.DefaultHttpContent;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpContent;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.util.ReferenceCountUtil;
import io.reactivex.netty.metrics.Clock;
import io.reactivex.netty.metrics.MetricEventsSubject;
import io.reactivex.netty.protocol.http.UnicastContentSubject;
import io.reactivex.netty.server.ServerMetricsEvent;

import java.util.concurrent.TimeUnit;

/**
 * A channel handler for {@link HttpServer} to convert netty's http request/response objects to {@link HttpServer}'s
 * request/response objects. It handles the following message types:
 *
 * <h2>Reading Objects</h2>
 * <ul>
 <li>{@link HttpRequest}: Converts it to {@link HttpServerRequest } </li>
 <li>{@link HttpContent}: Converts it to the content of the previously generated
{@link HttpServerRequest }</li>
 <li>{@link FullHttpRequest}: Converts it to a {@link HttpServerRequest } with pre-populated content observable.</li>
 <li>Any other object: Assumes that it is a transformed HTTP content & pass it through to the content observable.</li>
 </ul>
 *
 * <h2>Writing Objects</h2>
 * <ul>
 <li>{@link HttpServerResponse}: Converts it to a {@link HttpResponse}</li>
 <li>{@link ByteBuf} to an {@link HttpContent}</li>
 <li>Pass through any other message type.</li>
 </ul>
 *
 * @author <NAME>
 */
public class ServerRequestResponseConverter extends ChannelDuplexHandler {

    private final MetricEventsSubject<ServerMetricsEvent<?>> eventsSubject;
    private final long requestContentSubscriptionTimeoutMs;
    private RequestState currentRequestState;

    public ServerRequestResponseConverter(MetricEventsSubject<ServerMetricsEvent<?>> eventsSubject,
                                          long requestContentSubscriptionTimeoutMs) {
        this.eventsSubject = eventsSubject;
        this.requestContentSubscriptionTimeoutMs = requestContentSubscriptionTimeoutMs;
        currentRequestState = new RequestState();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Class<?> recievedMsgClass = msg.getClass();

        final RequestState stateToUse = currentRequestState;
        boolean isHttpRequest = false;

        if (HttpRequest.class.isAssignableFrom(recievedMsgClass)) {
            eventsSubject.onEvent(HttpServerMetricsEvent.REQUEST_HEADERS_RECEIVED);
            stateToUse.createRxRequest(ctx, (HttpRequest) msg); // Update the state to use.
            super.channelRead(ctx, stateToUse.rxRequest);
            isHttpRequest = true;
        }

        if (HttpContent.class.isAssignableFrom(recievedMsgClass)) {// This will be executed if the incoming message is a FullHttpRequest or only HttpContent.
            ByteBuf content = ((ByteBufHolder) msg).content();
            eventsSubject.onEvent(HttpServerMetricsEvent.REQUEST_CONTENT_RECEIVED);
            invokeContentOnNext(content, stateToUse.contentSubject);
            if (LastHttpContent.class.isAssignableFrom(recievedMsgClass)) {
                long durationInMs = stateToUse.rxRequest.onProcessingEnd();
                eventsSubject.onEvent(HttpServerMetricsEvent.REQUEST_RECEIVE_COMPLETE, durationInMs);
                stateToUse.contentSubject.onCompleted();
                currentRequestState = new RequestState(); // Reset the current state for the next request to arrive on this connection
            }<commit_msg>Fixes #243 (HttpServerListener requestReadTimes metric value is incorrect)<commit_after>/*
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.reactivex.netty.protocol.http.server;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.DefaultHttpContent;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpContent;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.util.ReferenceCountUtil;
import io.reactivex.netty.metrics.Clock;
import io.reactivex.netty.metrics.MetricEventsSubject;
import io.reactivex.netty.protocol.http.UnicastContentSubject;
import io.reactivex.netty.server.ServerMetricsEvent;

import java.util.concurrent.TimeUnit;

/**
 * A channel handler for {@link HttpServer} to convert netty's http request/response objects to {@link HttpServer}'s
 * request/response objects. It handles the following message types:
 *
 * <h2>Reading Objects</h2>
 * <ul>
 <li>{@link HttpRequest}: Converts it to {@link HttpServerRequest } </li>
 <li>{@link HttpContent}: Converts it to the content of the previously generated
{@link HttpServerRequest }</li>
 <li>{@link FullHttpRequest}: Converts it to a {@link HttpServerRequest } with pre-populated content observable.</li>
 <li>Any other object: Assumes that it is a transformed HTTP content & pass it through to the content observable.</li>
 </ul>
 *
 * <h2>Writing Objects</h2>
 * <ul>
 <li>{@link HttpServerResponse}: Converts it to a {@link HttpResponse}</li>
 <li>{@link ByteBuf} to an {@link HttpContent}</li>
 <li>Pass through any other message type.</li>
 </ul>
 *
 * @author <NAME>
 */
public class ServerRequestResponseConverter extends ChannelDuplexHandler {

    private final MetricEventsSubject<ServerMetricsEvent<?>> eventsSubject;
    private final long requestContentSubscriptionTimeoutMs;
    private RequestState currentRequestState;

    public ServerRequestResponseConverter(MetricEventsSubject<ServerMetricsEvent<?>> eventsSubject,
                                          long requestContentSubscriptionTimeoutMs) {
        this.eventsSubject = eventsSubject;
        this.requestContentSubscriptionTimeoutMs = requestContentSubscriptionTimeoutMs;
        currentRequestState = new RequestState();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Class<?> recievedMsgClass = msg.getClass();

        final RequestState stateToUse = currentRequestState;
        boolean isHttpRequest = false;

        if (HttpRequest.class.isAssignableFrom(recievedMsgClass)) {
            eventsSubject.onEvent(HttpServerMetricsEvent.REQUEST_HEADERS_RECEIVED);
            stateToUse.createRxRequest(ctx, (HttpRequest) msg); // Update the state to use.
            stateToUse.rxRequest.onProcessingStart(Clock.newStartTimeMillis());
            super.channelRead(ctx, stateToUse.rxRequest);
            isHttpRequest = true;
        }

        if (HttpContent.class.isAssignableFrom(recievedMsgClass)) {// This will be executed if the incoming message is a FullHttpRequest or only HttpContent.
            ByteBuf content = ((ByteBufHolder) msg).content();
            eventsSubject.onEvent(HttpServerMetricsEvent.REQUEST_CONTENT_RECEIVED);
            invokeContentOnNext(content, stateToUse.contentSubject);
            if (LastHttpContent.class.isAssignableFrom(recievedMsgClass)) {
                long durationInMs = stateToUse.rxRequest.onProcessingEnd();
                eventsSubject.onEvent(HttpServerMetricsEvent.REQUEST_RECEIVE_COMPLETE, durationInMs);
                stateToUse.contentSubject.onCompleted();
                currentRequestState = new RequestState(); // Reset the current state for the next request to arrive on this connection
            }
{'fraction_non_alphanumeric': 0.08956682227771143, 'fraction_numerical': 0.0029312778199978287, 'mean_word_length': 4.083885209713024, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6082824', 'n_tokens_mistral': 2454, 'n_tokens_neox': 2420, 'n_words': 682}