<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.bootstrap.ServerBootstrap;
import io.reactivex.netty.ConnectionHandler;
import io.reactivex.netty.ObservableConnection;
import io.reactivex.netty.pipeline.PipelineConfigurator;
import io.reactivex.netty.pipeline.PipelineConfiguratorComposite;
import io.reactivex.netty.server.RxServer;
import rx.Observable;
import rx.util.functions.Action0;
import rx.util.functions.Func1;

/**
 * @author <NAME>
 */
public class HttpServer<I, O> extends RxServer<HttpRequest<I>, HttpResponse<O>> {

    public HttpServer(ServerBootstrap bootstrap, int port,
                      PipelineConfigurator<HttpRequest<I>, HttpResponse<O>> pipelineConfigurator,
                      RequestHandler<I, O> requestHandler) {
        super(bootstrap, port, addRequiredConfigurator(pipelineConfigurator), new HttpConnectionHandler<I, O>(requestHandler));
    }

    HttpServer(ServerBootstrap bootstrap, int port,
               PipelineConfigurator<HttpRequest<I>, HttpResponse<O>> pipelineConfigurator,
               HttpConnectionHandler<I, O> connectionHandler) {
        super(bootstrap, port, addRequiredConfigurator(pipelineConfigurator), connectionHandler);
    }

    private static <I, O> PipelineConfigurator<HttpRequest<I>, HttpResponse<O>> addRequiredConfigurator(
            PipelineConfigurator<HttpRequest<I>, HttpResponse<O>> pipelineConfigurator) {
        return new PipelineConfiguratorComposite<HttpRequest<I>, HttpResponse<O>>(pipelineConfigurator,
                                                                                  new ServerRequiredConfigurator<I, O>());
    }

    static class HttpConnectionHandler<I, O> implements ConnectionHandler<HttpRequest<I>, HttpResponse<O>> {

        private final RequestHandler<I, O> requestHandler;

        public HttpConnectionHandler(RequestHandler<I, O> requestHandler) {
            this.requestHandler = requestHandler;
        }

        @Override
        public Observable<Void> handle(final ObservableConnection<HttpRequest<I>, HttpResponse<O>> newConnection) {

            return newConnection.getInput().flatMap(new Func1<HttpRequest<I>, Observable<Void>>() {
                @Override
                public Observable<Void> call(HttpRequest<I> newRequest) {
                    final HttpResponse<O> response = new HttpResponse<O>(newConnection.getChannelHandlerContext(),
                                                                   newRequest.getHttpVersion());
                    return requestHandler.handle(newRequest, response).onErrorResumeNext(new Func1<Throwable, Observable<Void>>() {
                        @Override
                        public Observable<Void> call(Throwable throwable) {
                            System.out.println("Error occured in request handling. Error: ");
                            return null;
                        }
                    }).finallyDo(new Action0() {
                        @Override
                        public void call() {
                            response.close();
                        }
                    });
                }
            });
        }<commit_msg>Adding exception stack of request handler error.<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.bootstrap.ServerBootstrap;
import io.reactivex.netty.ConnectionHandler;
import io.reactivex.netty.ObservableConnection;
import io.reactivex.netty.pipeline.PipelineConfigurator;
import io.reactivex.netty.pipeline.PipelineConfiguratorComposite;
import io.reactivex.netty.server.RxServer;
import rx.Observable;
import rx.util.functions.Action0;
import rx.util.functions.Func1;

/**
 * @author <NAME>
 */
public class HttpServer<I, O> extends RxServer<HttpRequest<I>, HttpResponse<O>> {

    public HttpServer(ServerBootstrap bootstrap, int port,
                      PipelineConfigurator<HttpRequest<I>, HttpResponse<O>> pipelineConfigurator,
                      RequestHandler<I, O> requestHandler) {
        super(bootstrap, port, addRequiredConfigurator(pipelineConfigurator), new HttpConnectionHandler<I, O>(requestHandler));
    }

    HttpServer(ServerBootstrap bootstrap, int port,
               PipelineConfigurator<HttpRequest<I>, HttpResponse<O>> pipelineConfigurator,
               HttpConnectionHandler<I, O> connectionHandler) {
        super(bootstrap, port, addRequiredConfigurator(pipelineConfigurator), connectionHandler);
    }

    private static <I, O> PipelineConfigurator<HttpRequest<I>, HttpResponse<O>> addRequiredConfigurator(
            PipelineConfigurator<HttpRequest<I>, HttpResponse<O>> pipelineConfigurator) {
        return new PipelineConfiguratorComposite<HttpRequest<I>, HttpResponse<O>>(pipelineConfigurator,
                                                                                  new ServerRequiredConfigurator<I, O>());
    }

    static class HttpConnectionHandler<I, O> implements ConnectionHandler<HttpRequest<I>, HttpResponse<O>> {

        private final RequestHandler<I, O> requestHandler;

        public HttpConnectionHandler(RequestHandler<I, O> requestHandler) {
            this.requestHandler = requestHandler;
        }

        @Override
        public Observable<Void> handle(final ObservableConnection<HttpRequest<I>, HttpResponse<O>> newConnection) {

            return newConnection.getInput().flatMap(new Func1<HttpRequest<I>, Observable<Void>>() {
                @Override
                public Observable<Void> call(HttpRequest<I> newRequest) {
                    final HttpResponse<O> response = new HttpResponse<O>(newConnection.getChannelHandlerContext(),
                                                                   newRequest.getHttpVersion());
                    return requestHandler.handle(newRequest, response).onErrorResumeNext(new Func1<Throwable, Observable<Void>>() {
                        @Override
                        public Observable<Void> call(Throwable throwable) {
                            System.err.println("Error occured in request handling. Error: ");
                            throwable.printStackTrace(System.err);
                            return null;
                        }
                    }).finallyDo(new Action0() {
                        @Override
                        public void call() {
                            response.close();
                        }
                    });
                }
            });
        }
{'fraction_non_alphanumeric': 0.08717148061410357, 'fraction_numerical': 0.0033827738745771533, 'mean_word_length': 2.47670737222976, 'pattern_counts': {'":': 0, '<': 93, '<?xml version=': 0, '>': 93, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2562310', 'n_tokens_mistral': 1878, 'n_tokens_neox': 1815, 'n_words': 457}