/*
 * ====================================================================
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of the Apache Software Foundation.  For more
 * information on the Apache Software Foundation, please see
 * <http://www.apache.org/>.
 *
 */
package org.apache.http.examples.nio;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.Locale;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.http.ConnectionReuseStrategy;
import org.apache.http.HttpEntityEnclosingRequest;
import org.apache.http.HttpException;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequest;
import org.apache.http.HttpRequestInterceptor;
import org.apache.http.HttpResponse;
import org.apache.http.HttpResponseInterceptor;
import org.apache.http.HttpStatus;
import org.apache.http.HttpVersion;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.DefaultConnectionReuseStrategy;
import org.apache.http.impl.EnglishReasonPhraseCatalog;
import org.apache.http.impl.nio.DefaultHttpClientIODispatch;
import org.apache.http.impl.nio.DefaultHttpServerIODispatch;
import org.apache.http.impl.nio.pool.BasicNIOConnPool;
import org.apache.http.impl.nio.pool.BasicNIOPoolEntry;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.DefaultListeningIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHttpEntityEnclosingRequest;
import org.apache.http.message.BasicHttpRequest;
import org.apache.http.message.BasicHttpResponse;
import org.apache.http.nio.ContentDecoder;
import org.apache.http.nio.ContentEncoder;
import org.apache.http.nio.IOControl;
import org.apache.http.nio.NHttpClientConnection;
import org.apache.http.nio.NHttpConnection;
import org.apache.http.nio.NHttpServerConnection;
import org.apache.http.nio.entity.NStringEntity;
import org.apache.http.nio.pool.NIOConnFactory;
import org.apache.http.nio.protocol.BasicAsyncResponseProducer;
import org.apache.http.nio.protocol.HttpAsyncRequestExecutor;
import org.apache.http.nio.protocol.HttpAsyncRequestConsumer;
import org.apache.http.nio.protocol.HttpAsyncRequester;
import org.apache.http.nio.protocol.HttpAsyncRequestHandler;
import org.apache.http.nio.protocol.HttpAsyncRequestHandlerRegistry;
import org.apache.http.nio.protocol.HttpAsyncRequestHandlerResolver;
import org.apache.http.nio.protocol.HttpAsyncRequestProducer;
import org.apache.http.nio.protocol.HttpAsyncResponseConsumer;
import org.apache.http.nio.protocol.HttpAsyncResponseProducer;
import org.apache.http.nio.protocol.HttpAsyncExchange;
import org.apache.http.nio.protocol.HttpAsyncService;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.reactor.ListeningIOReactor;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.http.params.CoreProtocolPNames;
import org.apache.http.params.HttpParams;
import org.apache.http.params.SyncBasicHttpParams;
import org.apache.http.pool.PoolStats;
import org.apache.http.protocol.ExecutionContext;
import org.apache.http.protocol.HttpContext;
import org.apache.http.protocol.HttpProcessor;
import org.apache.http.protocol.ImmutableHttpProcessor;
import org.apache.http.protocol.RequestConnControl;
import org.apache.http.protocol.RequestContent;
import org.apache.http.protocol.RequestExpectContinue;
import org.apache.http.protocol.RequestTargetHost;
import org.apache.http.protocol.RequestUserAgent;
import org.apache.http.protocol.ResponseConnControl;
import org.apache.http.protocol.ResponseContent;
import org.apache.http.protocol.ResponseDate;
import org.apache.http.protocol.ResponseServer;

/**
 * Asynchronous, fully streaming HTTP/1.1 reverse proxy.
 */
public class NHttpReverseProxy {

    public static void main(String[] args) throws Exception {
        if (args.length < 1) {
            System.out.println("Usage: NHttpReverseProxy <hostname> [port]");
            System.exit(1);
        }
        URI uri = new URI(args[0]);
        int port = 8080;
        if (args.length > 1) {
            port = Integer.parseInt(args[1]);
        }

        // Target host
        HttpHost targetHost = new HttpHost(
                uri.getHost(),
                uri.getPort() > 0 ? uri.getPort() : 80,
                uri.getScheme() != null ? uri.getScheme() : "http");

        System.out.println("Reverse proxy to " + targetHost);

        HttpParams params = new SyncBasicHttpParams();
        params
            .setIntParameter(CoreConnectionPNames.SO_TIMEOUT, 30000)
            .setIntParameter(CoreConnectionPNames.SOCKET_BUFFER_SIZE, 8 * 1024)
            .setParameter(CoreProtocolPNames.ORIGIN_SERVER, "Test/1.1")
            .setParameter(CoreProtocolPNames.USER_AGENT, "Test/1.1");

        IOReactorConfig config = new IOReactorConfig();
        config.setIoThreadCount(1);
        final ConnectingIOReactor connectingIOReactor = new DefaultConnectingIOReactor(config);
        final ListeningIOReactor listeningIOReactor = new DefaultListeningIOReactor(config);

        // Set up HTTP protocol processor for incoming connections
        HttpProcessor inhttpproc = new ImmutableHttpProcessor(
                new HttpResponseInterceptor[] {
                        new ResponseDate(),
                        new ResponseServer(),
                        new ResponseContent(),
                        new ResponseConnControl()
         });

        // Set up HTTP protocol processor for outgoing connections
        HttpProcessor outhttpproc = new ImmutableHttpProcessor(
                new HttpRequestInterceptor[] {
                        new RequestContent(),
                        new RequestTargetHost(),
                        new RequestConnControl(),
                        new RequestUserAgent(),
                        new RequestExpectContinue()
        });

        ProxyClientProtocolHandler clientHandler = new ProxyClientProtocolHandler();
        HttpAsyncRequester executor = new HttpAsyncRequester(
                outhttpproc, new ProxyOutgoingConnectionReuseStrategy(), params);

        ProxyConnPool connPool = new ProxyConnPool(connectingIOReactor, params);
        connPool.setMaxTotal(100);
        connPool.setDefaultMaxPerRoute(20);

        HttpAsyncRequestHandlerRegistry handlerRegistry = new HttpAsyncRequestHandlerRegistry();
        handlerRegistry.register("*", new ProxyRequestHandler(targetHost, executor, connPool));

        ProxyServiceHandler serviceHandler = new ProxyServiceHandler(
                inhttpproc, new ProxyIncomingConnectionReuseStrategy(), handlerRegistry, params);

        final IOEventDispatch connectingEventDispatch = new DefaultHttpClientIODispatch(
                clientHandler, params);

        final IOEventDispatch listeningEventDispatch = new DefaultHttpServerIODispatch(
                serviceHandler, params);

        Thread t = new Thread(new Runnable() {

            public void run() {
                try {
                    connectingIOReactor.execute(connectingEventDispatch);
                } catch (InterruptedIOException ex) {
                    System.err.println("Interrupted");
                } catch (IOException ex) {
                    ex.printStackTrace();
                } finally {
                    try {
                        listeningIOReactor.shutdown();
                    } catch (IOException ex2) {
                        ex2.printStackTrace();
                    }
                }
            }

        });
        t.start();
        try {
            listeningIOReactor.listen(new InetSocketAddress(port));
            listeningIOReactor.execute(listeningEventDispatch);
        } catch (InterruptedIOException ex) {
            System.err.println("Interrupted");
        } catch (IOException ex) {
            ex.printStackTrace();
        } finally {
            try {
                connectingIOReactor.shutdown();
            } catch (IOException ex2) {
                ex2.printStackTrace();
            }
        }
    }

    static class ProxyHttpExchange {

        private final ByteBuffer inBuffer;
        private final ByteBuffer outBuffer;

        private volatile String id;
        private volatile HttpHost target;
        private volatile HttpAsyncExchange responseTrigger;
        private volatile IOControl originIOControl;
        private volatile IOControl clientIOControl;
        private volatile HttpRequest request;
        private volatile boolean requestReceived;
        private volatile HttpResponse response;
        private volatile boolean responseReceived;
        private volatile Exception ex;

        public ProxyHttpExchange() {
            super();
            this.inBuffer = ByteBuffer.allocateDirect(10240);
            this.outBuffer = ByteBuffer.allocateDirect(10240);
        }

        public ByteBuffer getInBuffer() {
            return this.inBuffer;
        }

        public ByteBuffer getOutBuffer() {
            return this.outBuffer;
        }

        public String getId() {
            return this.id;
        }

        public void setId(final String id) {
            this.id = id;
        }

        public HttpHost getTarget() {
            return this.target;
        }

        public void setTarget(final HttpHost target) {
            this.target = target;
        }

        public HttpRequest getRequest() {
            return this.request;
        }

        public void setRequest(final HttpRequest request) {
            this.request = request;
        }

        public HttpResponse getResponse() {
            return this.response;
        }

        public void setResponse(final HttpResponse response) {
            this.response = response;
        }

        public HttpAsyncExchange getResponseTrigger() {
            return this.responseTrigger;
        }

        public void setResponseTrigger(final HttpAsyncExchange responseTrigger) {
            this.responseTrigger = responseTrigger;
        }

        public IOControl getClientIOControl() {
            return this.clientIOControl;
        }

        public void setClientIOControl(final IOControl clientIOControl) {
            this.clientIOControl = clientIOControl;
        }

        public IOControl getOriginIOControl() {
            return this.originIOControl;
        }

        public void setOriginIOControl(final IOControl originIOControl) {
            this.originIOControl = originIOControl;
        }

        public boolean isRequestReceived() {
            return this.requestReceived;
        }

        public void setRequestReceived() {
            this.requestReceived = true;
        }

        public boolean isResponseReceived() {
            return this.responseReceived;
        }

        public void setResponseReceived() {
            this.responseReceived = true;
        }

        public Exception getException() {
            return this.ex;
        }

        public void setException(final Exception ex) {
            this.ex = ex;
        }

        public void reset() {
            this.inBuffer.clear();
            this.outBuffer.clear();
            this.target = null;
            this.id = null;
            this.responseTrigger = null;
            this.clientIOControl = null;
            this.originIOControl = null;
            this.request = null;
            this.requestReceived = false;
            this.response = null;
            this.responseReceived = false;
            this.ex = null;
        }

    }

    static class ProxyRequestHandler implements HttpAsyncRequestHandler<ProxyHttpExchange> {

        private final HttpHost target;
        private final HttpAsyncRequester executor;
        private final BasicNIOConnPool connPool;
        private final AtomicLong counter;

        public ProxyRequestHandler(
                final HttpHost target,
                final HttpAsyncRequester executor,
                final BasicNIOConnPool connPool) {
            super();
            this.target = target;
            this.executor = executor;
            this.connPool = connPool;
            this.counter = new AtomicLong(1);
        }

        public HttpAsyncRequestConsumer<ProxyHttpExchange> processRequest(
                final HttpRequest request,
                final HttpContext context) {
            ProxyHttpExchange httpExchange = (ProxyHttpExchange) context.getAttribute("http-exchange");
            if (httpExchange == null) {
                httpExchange = new ProxyHttpExchange();
                context.setAttribute("http-exchange", httpExchange);
            }
            synchronized (httpExchange) {
                httpExchange.reset();
                String id = String.format("%08X", this.counter.getAndIncrement());
                httpExchange.setId(id);
                httpExchange.setTarget(this.target);
                return new ProxyRequestConsumer(httpExchange, this.executor, this.connPool);
            }
        }

        public void handle(
                final ProxyHttpExchange httpExchange,
                final HttpAsyncExchange responseTrigger,
                final HttpContext context) throws HttpException, IOException {
            synchronized (httpExchange) {
                Exception ex = httpExchange.getException();
                if (ex != null) {
                    System.out.println("[client<-proxy] " + httpExchange.getId() + " " + ex);
                    int status = HttpStatus.SC_INTERNAL_SERVER_ERROR;
                    HttpResponse response = new BasicHttpResponse(HttpVersion.HTTP_1_0, status,
                            EnglishReasonPhraseCatalog.INSTANCE.getReason(status, Locale.US));
                    String message = ex.getMessage();
                    if (message == null) {
                        message = "Unexpected error";
                    }
                    response.setEntity(new NStringEntity(message, ContentType.DEFAULT_TEXT));
                    responseTrigger.submitResponse(new BasicAsyncResponseProducer(response));
                    System.out.println("[client<-proxy] " + httpExchange.getId() + " error response triggered");
                }
                HttpResponse response = httpExchange.getResponse();
                if (response != null) {
                    responseTrigger.submitResponse(new ProxyResponseProducer(httpExchange));
                    System.out.println("[client<-proxy] " + httpExchange.getId() + " response triggered");
                }
                // No response yet.
                httpExchange.setResponseTrigger(responseTrigger);
            }
        }

    }

    static class ProxyRequestConsumer implements HttpAsyncRequestConsumer<ProxyHttpExchange> {

        private final ProxyHttpExchange httpExchange;
        private final HttpAsyncRequester executor;
        private final BasicNIOConnPool connPool;

        private volatile boolean completed;

        public ProxyRequestConsumer(
                final ProxyHttpExchange httpExchange,
                final HttpAsyncRequester executor,
                final BasicNIOConnPool connPool) {
            super();
            this.httpExchange = httpExchange;
            this.executor = executor;
            this.connPool = connPool;
        }

        public void close() throws IOException {
        }

        public void requestReceived(final HttpRequest request) {
            synchronized (this.httpExchange) {
                System.out.println("[client->proxy] " + this.httpExchange.getId() + " " + request.getRequestLine());
                this.httpExchange.setRequest(request);
                this.executor.execute(
                        new ProxyRequestProducer(this.httpExchange),
                        new ProxyResponseConsumer(this.httpExchange),
                        this.connPool);
            }
        }

        public void consumeContent(
                final ContentDecoder decoder, final IOControl ioctrl) throws IOException {
            synchronized (this.httpExchange) {
                this.httpExchange.setClientIOControl(ioctrl);
                // Receive data from the client
                ByteBuffer buf = this.httpExchange.getInBuffer();
                int n = decoder.read(buf);
                System.out.println("[client->proxy] " + this.httpExchange.getId() + " " + n + " bytes read");
                if (decoder.isCompleted()) {
                    System.out.println("[client->proxy] " + this.httpExchange.getId() + " content fully read");
                }
                // If the buffer is full, suspend client input until there is free
                // space in the buffer
                if (!buf.hasRemaining()) {
                    ioctrl.suspendInput();
                    System.out.println("[client->proxy] " + this.httpExchange.getId() + " suspend client input");
                }
                // If there is some content in the input buffer make sure origin
                // output is active
                if (buf.position() > 0) {
                    if (this.httpExchange.getOriginIOControl() != null) {
                        this.httpExchange.getOriginIOControl().requestOutput();
                        System.out.println("[client->proxy] " + this.httpExchange.getId() + " request origin output");
                    }
                }
            }
        }

        public void requestCompleted(final HttpContext context) {
            synchronized (this.httpExchange) {
                this.completed = true;;
                System.out.println("[client->proxy] " + this.httpExchange.getId() + " request completed");
                this.httpExchange.setRequestReceived();
                if (this.httpExchange.getOriginIOControl() != null) {
                    this.httpExchange.getOriginIOControl().requestOutput();
                    System.out.println("[client->proxy] " + this.httpExchange.getId() + " request origin output");
                }
            }
        }

        public Exception getException() {
            return null;
        }

        public ProxyHttpExchange getResult() {
            return this.httpExchange;
        }

        public boolean isDone() {
            return this.completed;
        }

        public void failed(final Exception ex) {
            System.out.println("[client->proxy] " + ex.toString());
        }

    }

    static class ProxyRequestProducer implements HttpAsyncRequestProducer {

        private final ProxyHttpExchange httpExchange;

        public ProxyRequestProducer(final ProxyHttpExchange httpExchange) {
            super();
            this.httpExchange = httpExchange;
        }

        public void close() throws IOException {
        }

        public HttpHost getTarget() {
            synchronized (this.httpExchange) {
                return this.httpExchange.getTarget();
            }
        }

        public HttpRequest generateRequest() {
            synchronized (this.httpExchange) {
                HttpRequest request = this.httpExchange.getRequest();
                System.out.println("[proxy->origin] " + this.httpExchange.getId() + " " + request.getRequestLine());
                // Rewrite request!!!!
                if (request instanceof HttpEntityEnclosingRequest) {
                    BasicHttpEntityEnclosingRequest r = new BasicHttpEntityEnclosingRequest(
                            request.getRequestLine());
                    r.setEntity(((HttpEntityEnclosingRequest) request).getEntity());
                    return r;
                } else {
                    return new BasicHttpRequest(request.getRequestLine());
                }
            }
        }

        public void produceContent(
                final ContentEncoder encoder, final IOControl ioctrl) throws IOException {
            synchronized (this.httpExchange) {
                this.httpExchange.setOriginIOControl(ioctrl);
                // Send data to the origin server
                ByteBuffer buf = this.httpExchange.getInBuffer();
                buf.flip();
                int n = encoder.write(buf);
                buf.compact();
                System.out.println("[proxy->origin] " + this.httpExchange.getId() + " " + n + " bytes written");
                // If there is space in the buffer and the message has not been
                // transferred, make sure the client is sending more data
                if (buf.hasRemaining() && !this.httpExchange.isRequestReceived()) {
                    if (this.httpExchange.getClientIOControl() != null) {
                        this.httpExchange.getClientIOControl().requestInput();
                        System.out.println("[proxy->origin] " + this.httpExchange.getId() + " request client input");
                    }
                }
                if (buf.position() == 0) {
                    if (this.httpExchange.isRequestReceived()) {
                        encoder.complete();
                        System.out.println("[proxy->origin] " + this.httpExchange.getId() + " content fully written");
                    } else {
                        // Input buffer is empty. Wait until the client fills up
                        // the buffer
                        ioctrl.suspendOutput();
                        System.out.println("[proxy->origin] " + this.httpExchange.getId() + " suspend origin output");
                    }
                }
            }
        }

        public void requestCompleted(final HttpContext context) {
            synchronized (this.httpExchange) {
                System.out.println("[proxy->origin] " + this.httpExchange.getId() + " request completed");
            }
        }

        public boolean isRepeatable() {
            return false;
        }

        public void resetRequest() {
        }

        public void failed(final Exception ex) {
            System.out.println("[proxy->origin] " + ex.toString());
        }

    }

    static class ProxyResponseConsumer implements HttpAsyncResponseConsumer<ProxyHttpExchange> {

        private final ProxyHttpExchange httpExchange;

        private volatile boolean completed;

        public ProxyResponseConsumer(final ProxyHttpExchange httpExchange) {
            super();
            this.httpExchange = httpExchange;
        }

        public void close() throws IOException {
        }

        public void responseReceived(final HttpResponse response) {
            synchronized (this.httpExchange) {
                System.out.println("[proxy<-origin] " + this.httpExchange.getId() + " " + response.getStatusLine());
                this.httpExchange.setResponse(response);
                HttpAsyncExchange responseTrigger = this.httpExchange.getResponseTrigger();
                if (responseTrigger != null && !responseTrigger.isCompleted()) {
                    System.out.println("[client<-proxy] " + this.httpExchange.getId() + " response triggered");
                    responseTrigger.submitResponse(new ProxyResponseProducer(this.httpExchange));
                }
            }
        }

        public void consumeContent(
                final ContentDecoder decoder, final IOControl ioctrl) throws IOException {
            synchronized (this.httpExchange) {
                this.httpExchange.setOriginIOControl(ioctrl);
                // Receive data from the origin
                ByteBuffer buf = this.httpExchange.getOutBuffer();
                int n = decoder.read(buf);
                System.out.println("[proxy<-origin] " + this.httpExchange.getId() + " " + n + " bytes read");
                if (decoder.isCompleted()) {
                    System.out.println("[proxy<-origin] " + this.httpExchange.getId() + " content fully read");
                }
                // If the buffer is full, suspend origin input until there is free
                // space in the buffer
                if (!buf.hasRemaining()) {
                    ioctrl.suspendInput();
                    System.out.println("[proxy<-origin] " + this.httpExchange.getId() + " suspend origin input");
                }
                // If there is some content in the input buffer make sure client
                // output is active
                if (buf.position() > 0) {
                    if (this.httpExchange.getClientIOControl() != null) {
                        this.httpExchange.getClientIOControl().requestOutput();
                        System.out.println("[proxy<-origin] " + this.httpExchange.getId() + " request client output");
                    }
                }
            }
        }

        public void responseCompleted(final HttpContext context) {
            synchronized (this.httpExchange) {
                if (this.completed) {
                    return;
                }
                this.completed = true;
                System.out.println("[proxy<-origin] " + this.httpExchange.getId() + " response completed");
                this.httpExchange.setResponseReceived();
                if (this.httpExchange.getClientIOControl() != null) {
                    this.httpExchange.getClientIOControl().requestOutput();
                    System.out.println("[proxy<-origin] " + this.httpExchange.getId() + " request client output");
                }
            }
        }

        public void failed(final Exception ex) {
            synchronized (this.httpExchange) {
                if (this.completed) {
                    return;
                }
                this.completed = true;
                this.httpExchange.setException(ex);
                HttpAsyncExchange responseTrigger = this.httpExchange.getResponseTrigger();
                if (responseTrigger != null && !responseTrigger.isCompleted()) {
                    System.out.println("[client<-proxy] " + this.httpExchange.getId() + " " + ex);
                    int status = HttpStatus.SC_INTERNAL_SERVER_ERROR;
                    HttpResponse response = new BasicHttpResponse(HttpVersion.HTTP_1_0, status,
                            EnglishReasonPhraseCatalog.INSTANCE.getReason(status, Locale.US));
                    String message = ex.getMessage();
                    if (message == null) {
                        message = "Unexpected error";
                    }
                    response.setEntity(new NStringEntity(message, ContentType.DEFAULT_TEXT));
                    responseTrigger.submitResponse(new BasicAsyncResponseProducer(response));
                }
            }
        }

        public boolean cancel() {
            synchronized (this.httpExchange) {
                if (this.completed) {
                    return false;
                }
                failed(new InterruptedIOException("Cancelled"));
                return true;
            }
        }

        public ProxyHttpExchange getResult() {
            return this.httpExchange;
        }

        public Exception getException() {
            return null;
        }

        public boolean isDone() {
            return this.completed;
        }

    }

    static class ProxyResponseProducer implements HttpAsyncResponseProducer {

        private final ProxyHttpExchange httpExchange;

        public ProxyResponseProducer(final ProxyHttpExchange httpExchange) {
            super();
            this.httpExchange = httpExchange;
        }

        public void close() throws IOException {
            this.httpExchange.reset();
        }

        public HttpResponse generateResponse() {
            synchronized (this.httpExchange) {
                HttpResponse response = this.httpExchange.getResponse();
                System.out.println("[client<-proxy] " + this.httpExchange.getId() + " " + response.getStatusLine());
                // Rewrite response!!!!
                BasicHttpResponse r = new BasicHttpResponse(response.getStatusLine());
                r.setEntity(response.getEntity());
                return r;
            }
        }

        public void produceContent(
                final ContentEncoder encoder, final IOControl ioctrl) throws IOException {
            synchronized (this.httpExchange) {
                this.httpExchange.setClientIOControl(ioctrl);
                // Send data to the client
                ByteBuffer buf = this.httpExchange.getOutBuffer();
                buf.flip();
                int n = encoder.write(buf);
                buf.compact();
                System.out.println("[client<-proxy] " + this.httpExchange.getId() + " " + n + " bytes written");
                // If there is space in the buffer and the message has not been
                // transferred, make sure the origin is sending more data
                if (buf.hasRemaining() && !this.httpExchange.isResponseReceived()) {
                    if (this.httpExchange.getOriginIOControl() != null) {
                        this.httpExchange.getOriginIOControl().requestInput();
                        System.out.println("[client<-proxy] " + this.httpExchange.getId() + " request origin input");
                    }
                }
                if (buf.position() == 0) {
                    if (this.httpExchange.isResponseReceived()) {
                        encoder.complete();
                        System.out.println("[client<-proxy] " + this.httpExchange.getId() + " content fully written");
                    } else {
                        // Input buffer is empty. Wait until the origin fills up
                        // the buffer
                        ioctrl.suspendOutput();
                        System.out.println("[client<-proxy] " + this.httpExchange.getId() + " suspend client output");
                    }
                }
            }
        }

        public void responseCompleted(final HttpContext context) {
            synchronized (this.httpExchange) {
                System.out.println("[client<-proxy] " + this.httpExchange.getId() + " response completed");
            }
        }

        public void failed(final Exception ex) {
            System.out.println("[client<-proxy] " + ex.toString());
        }

    }

    static class ProxyIncomingConnectionReuseStrategy extends DefaultConnectionReuseStrategy {

        @Override
        public boolean keepAlive(final HttpResponse response, final HttpContext context) {
            NHttpConnection conn = (NHttpConnection) context.getAttribute(
                    ExecutionContext.HTTP_CONNECTION);
            boolean keepAlive = super.keepAlive(response, context);
            if (keepAlive) {
                System.out.println("[client->proxy] connection kept alive " + conn);
            }
            return keepAlive;
        }

    };

    static class ProxyOutgoingConnectionReuseStrategy extends DefaultConnectionReuseStrategy {

        @Override
        public boolean keepAlive(final HttpResponse response, final HttpContext context) {
            NHttpConnection conn = (NHttpConnection) context.getAttribute(
                    ExecutionContext.HTTP_CONNECTION);
            boolean keepAlive = super.keepAlive(response, context);
            if (keepAlive) {
                System.out.println("[proxy->origin] connection kept alive " + conn);
            }
            return keepAlive;
        }

    };

    static class ProxyServiceHandler extends HttpAsyncService {

        public ProxyServiceHandler(
                final HttpProcessor httpProcessor,
                final ConnectionReuseStrategy reuseStrategy,
                final HttpAsyncRequestHandlerResolver handlerResolver,
                final HttpParams params) {
            super(httpProcessor, reuseStrategy, handlerResolver, params);
        }

        @Override
        protected void log(final Exception ex) {
            ex.printStackTrace();
        }

        @Override
        public void connected(final NHttpServerConnection conn) {
            System.out.println("[client->proxy] connection open " + conn);
            super.connected(conn);
        }

        @Override
        public void closed(final NHttpServerConnection conn) {
            System.out.println("[client->proxy] connection closed " + conn);
            super.closed(conn);
        }

    }

    static class ProxyClientProtocolHandler extends HttpAsyncRequestExecutor {

        public ProxyClientProtocolHandler() {
            super();
        }

        @Override
        protected void log(final Exception ex) {
            ex.printStackTrace();
        }

        @Override
        public void connected(final NHttpClientConnection conn,
                final Object attachment) throws IOException, HttpException {
            System.out.println("[proxy->origin] connection open " + conn);
            super.connected(conn, attachment);
        }

        @Override
        public void closed(final NHttpClientConnection conn) {
            System.out.println("[proxy->origin] connection closed " + conn);
            super.closed(conn);
        }

    }

    static class ProxyConnPool extends BasicNIOConnPool {

        public ProxyConnPool(final ConnectingIOReactor ioreactor, final HttpParams params) {
            super(ioreactor, params);
        }

        public ProxyConnPool(
                final ConnectingIOReactor ioreactor,
                final NIOConnFactory<HttpHost, NHttpClientConnection> connFactory,
                final HttpParams params) {
            super(ioreactor, connFactory, params);
        }

        @Override
        public void release(final BasicNIOPoolEntry entry, boolean reusable) {
            System.out.println("[proxy->origin] connection released " + entry.getConnection());
            super.release(entry, reusable);
            StringBuilder buf = new StringBuilder();
            PoolStats totals = getTotalStats();
            buf.append("[total kept alive: ").append(totals.getAvailable()).append("; ");
            buf.append("total allocated: ").append(totals.getLeased() + totals.getAvailable());
            buf.append(" of ").append(totals.getMax()).append("]");
            System.out.println("[proxy->origin] " + buf.toString());
        }

    }

}