/*
 *  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.
 */

package org.apache.synapse.transport.passthru;

import org.apache.axiom.blob.Blobs;
import org.apache.axiom.blob.OverflowableBlob;
import org.apache.axiom.om.OMOutputFormat;
import org.apache.axis2.AxisFault;
import org.apache.axis2.Constants;
import org.apache.axis2.addressing.AddressingConstants;
import org.apache.axis2.addressing.AddressingHelper;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.description.Parameter;
import org.apache.axis2.description.TransportOutDescription;
import org.apache.axis2.handlers.AbstractHandler;
import org.apache.axis2.transport.MessageFormatter;
import org.apache.axis2.transport.OutTransportInfo;
import org.apache.axis2.transport.TransportSender;
import org.apache.axis2.transport.base.BaseConstants;
import org.apache.axis2.transport.base.threads.NativeThreadFactory;
import org.apache.axis2.transport.base.threads.WorkerPool;
import org.apache.axis2.util.MessageProcessorSelector;
import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.HttpException;
import org.apache.http.HttpStatus;
import org.apache.http.config.ConnectionConfig;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.nio.NHttpClientEventHandler;
import org.apache.http.nio.NHttpServerConnection;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.nio.reactor.IOReactorExceptionHandler;
import org.apache.http.nio.reactor.ssl.SSLSetupHandler;
import org.apache.synapse.commons.jmx.MBeanRegistrar;
import org.apache.synapse.transport.nhttp.NhttpConstants;
import org.apache.synapse.transport.nhttp.util.MessageFormatterDecoratorFactory;
import org.apache.synapse.transport.nhttp.util.NhttpUtil;
import org.apache.synapse.transport.passthru.config.SourceConfiguration;
import org.apache.synapse.transport.passthru.config.TargetConfiguration;
import org.apache.synapse.transport.passthru.connections.TargetConnections;
import org.apache.synapse.transport.passthru.jmx.PassThroughTransportMetricsCollector;
import org.apache.synapse.transport.passthru.jmx.TransportView;
import org.apache.synapse.transport.passthru.util.PassThroughTransportUtils;
import org.apache.synapse.transport.passthru.util.SourceResponseFactory;
import org.apache.synapse.transport.utils.conn.logging.LoggingUtils;

import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.OutputStream;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * PassThroughHttpSender for Synapse based on HttpCore and NIO extensions
 */
public class PassThroughHttpSender extends AbstractHandler implements TransportSender {

    protected Log log = LogFactory.getLog(this.getClass().getName());

    /** IOReactor used to create connections and manage them */
    private DefaultConnectingIOReactor ioReactor;

    /** Delivery agent used for delivering the messages to the servers */
    private DeliveryAgent deliveryAgent;

    /** The configuration of the sender */
    private TargetConfiguration targetConfiguration;

    /** state of the sender */
    private volatile int state = BaseConstants.STOPPED;

    private String namePrefix;

    /** The proxy host */
    private String proxyHost = null;

    /** The proxy port */
    private int proxyPort = 80;

    /** The list of hosts for which the proxy should be bypassed */
    private String[] proxyBypassList = new String[0];

    /** The list of known hosts to bypass proxy */
    private List<String> knownDirectHosts = new ArrayList<String>();

    /** The list of known hosts to go via proxy */
    private List<String> knownProxyHosts = new ArrayList<String>();

    @Override
    public void init(ConfigurationContext configurationContext,
                     TransportOutDescription transportOutDescription) throws AxisFault {

        if (log.isDebugEnabled()) {
            log.debug("Initializing pass-through HTTP/S sender...");
        }

        // is this an SSL Sender?
        SSLContext sslContext = getSSLContext(transportOutDescription);
        SSLSetupHandler sslSetupHandler = getSSLSetupHandler(transportOutDescription);

        // configure proxy settings
        if (sslContext == null) {
            Parameter proxyHostParam = transportOutDescription.getParameter("http.proxyHost");
            if (proxyHostParam != null || System.getProperty("http.proxyHost") != null) {
                if (proxyHostParam != null) {
                    proxyHost = (String) proxyHostParam.getValue();
                } else {
                    proxyHost = System.getProperty("http.proxyHost");
                }

                Parameter proxyPortParam = transportOutDescription.getParameter("http.proxyPort");
                if (proxyPortParam != null) {
                    proxyPort = Integer.parseInt((String) proxyPortParam.getValue());
                } else if (System.getProperty("http.proxyPort") != null) {
                    proxyPort = Integer.parseInt(System.getProperty("http.proxyPort"));
                }

                Parameter bypassList = transportOutDescription.getParameter("http.nonProxyHosts");
                if (bypassList != null) {
                    proxyBypassList = ((String) bypassList.getValue()).split("\\|");
                } else if (System.getProperty("http.nonProxyHosts") != null) {
                    proxyBypassList = (System.getProperty("http.nonProxyHosts")).split("\\|");
                }

                log.info("HTTP sender using Proxy : "
                    + proxyHost + ":" + proxyPort + " bypassing : " + Arrays.toString(proxyBypassList));
            }
        }

        namePrefix = (sslContext == null) ? "HTTP" : "HTTPS";

        WorkerPool workerPool = null;
        Object obj = configurationContext.getProperty(
                PassThroughConstants.PASS_THROUGH_TRANSPORT_WORKER_POOL);
        if (obj != null) {
            workerPool = (WorkerPool) obj;                                   
        }

        targetConfiguration = new TargetConfiguration(configurationContext,
                transportOutDescription, workerPool);
        configurationContext.setProperty(PassThroughConstants.PASS_THROUGH_TRANSPORT_WORKER_POOL,
                targetConfiguration.getWorkerPool());

        PassThroughTransportMetricsCollector metrics = new
                PassThroughTransportMetricsCollector(false, sslContext != null);
        TransportView view = new TransportView(null, this, metrics, null);
        MBeanRegistrar.getInstance().registerMBean(view, "Transport",
                 "passthru-" + namePrefix.toLowerCase() + "-sender");
        targetConfiguration.setMetrics(metrics);

        try {
            String prefix = namePrefix + "-PT-Sender I/O Dispatcher";

            ioReactor = new DefaultConnectingIOReactor(
                            targetConfiguration.getReactorConfig(false),
                            new NativeThreadFactory(new ThreadGroup(prefix + " Thread Group"), prefix));

            ioReactor.setExceptionHandler(new IOReactorExceptionHandler() {

                @Override
                public boolean handle(IOException ioException) {
                    log.warn("System may be unstable: " + namePrefix +
                            " ConnectingIOReactor encountered a checked exception : " +
                            ioException.getMessage(), ioException);
                    return true;
                }

                @Override
                public boolean handle(RuntimeException runtimeException) {
                    log.warn("System may be unstable: " + namePrefix +
                            " ConnectingIOReactor encountered a runtime exception : "
                            + runtimeException.getMessage(), runtimeException);
                    return true;
                }
            });
        } catch (IOReactorException e) {
            handleException("Error starting " + namePrefix + " ConnectingIOReactor", e);
        }

        ConnectCallback connectCallback = new ConnectCallback();
        // manage target connections
        TargetConnections targetConnections =
                new TargetConnections(ioReactor, targetConfiguration, connectCallback);
        targetConfiguration.setConnections(targetConnections);

        // create the delivery agent to hand over messages
        deliveryAgent = new DeliveryAgent(targetConfiguration, targetConnections);
        // we need to set the delivery agent
        connectCallback.setDeliveryAgent(deliveryAgent);        

        TargetHandler handler = new TargetHandler(deliveryAgent, targetConfiguration);
        final IOEventDispatch ioEventDispatch =
                getEventDispatch(handler, sslContext, sslSetupHandler,
                        targetConfiguration.getConnectionConfig(), transportOutDescription);

        // start the sender in a separate thread
        Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    ioReactor.execute(ioEventDispatch);
                } catch (Exception ex) {
                   log.fatal("Exception encountered in the " + namePrefix + " sender. " +
                            "No more connections will be initiated by this transport", ex);
                }
                log.info(namePrefix + " sender shutdown");
            }
        }, "PassThrough" + namePrefix + "Sender");
        t.start();

        state = BaseConstants.STARTED;

        log.info("Pass-through " + namePrefix + " sender started...");
    }

    @Override
    public void cleanup(org.apache.axis2.context.MessageContext messageContext) throws AxisFault {

    }

    @Override
    public void stop() {
        try {
            ioReactor.shutdown();
        } catch (IOException e) {
            log.error("Error shutting down the PassThroughHttpSender", e);
        }
    }

    @Override
    public InvocationResponse invoke(MessageContext msgContext) throws AxisFault {
        // remove unwanted HTTP headers (if any from the current message)
        PassThroughTransportUtils.removeUnwantedHeaders(msgContext, targetConfiguration);

        if (AddressingHelper.isReplyRedirected(msgContext)
                && !msgContext.getReplyTo().hasNoneAddress()) {

            msgContext.setProperty(PassThroughConstants.IGNORE_SC_ACCEPTED, Constants.VALUE_TRUE);
        }

        EndpointReference epr = PassThroughTransportUtils.getDestinationEPR(msgContext);
        if (epr != null) {
            if (!epr.hasNoneAddress()) {
                if (msgContext.getProperty(PassThroughConstants.PASS_THROUGH_PIPE) == null) {
                    Pipe pipe = new Pipe(targetConfiguration.getBufferFactory().getBuffer(),
                            "Test", targetConfiguration);
                    msgContext.setProperty(PassThroughConstants.PASS_THROUGH_PIPE, pipe);
                    msgContext.setProperty(PassThroughConstants.MESSAGE_BUILDER_INVOKED,
                            Boolean.TRUE);
                }
                try {
                    URL url = new URL(epr.getAddress());
                    String host = url.getHost();
                    int port = url.getPort();
                    if (port == -1) {
                        // use default
                        if ("http".equals(url.getProtocol())) {
                            port = 80;
                        } else if ("https".equals(url.getProtocol())) {
                            port = 443;
                        }
                    }

                    if (proxyHost != null) {
                        if (knownProxyHosts.contains(host)) {
                            // this has already been found to be a proxy host
                            host = proxyHost;
                            port = proxyPort;
                        } else if (knownDirectHosts.contains(host)) {
                            // do nothing, let this request go directly bypassing proxy
                        } else {
                            // we are encountering this host:port pair for the first time
                            if (!isBypass(host)) {
                                host = proxyHost;
                                port = proxyPort;
                            }
                        }
                    }

                    deliveryAgent.submit(msgContext, host, port);
                    sendRequestContent(msgContext);
                } catch (MalformedURLException e) {
                    handleException("Malformed URL in the target EPR", e);
                }
            } else {
                handleException("Cannot send message to " + AddressingConstants.Final.WSA_NONE_URI);
            }
        } else {
            if (msgContext.getProperty(Constants.OUT_TRANSPORT_INFO) != null) {
                if (msgContext.getProperty(Constants.OUT_TRANSPORT_INFO) instanceof ServerWorker) {
                    try {
                        submitResponse(msgContext);
                    } catch (Exception e) {
                        handleException("Failed to submit the response", e);
                    }
                } else {
                    //handleException("No valid destination EPR to send message");
                	//should be able to handle sendUsingOutputStream  Ref NHTTP_NIO
                	sendUsingOutputStream(msgContext);
                }
            } else {
                handleException("No valid destination EPR to send message");
            }

            if (msgContext.getOperationContext() != null) {
                msgContext.getOperationContext().setProperty(
                        Constants.RESPONSE_WRITTEN, Constants.VALUE_TRUE);
            }
        }


        return InvocationResponse.CONTINUE;
    }
    
    
    private void sendUsingOutputStream(MessageContext msgContext) throws AxisFault {

        OMOutputFormat format = NhttpUtil.getOMOutputFormat(msgContext);
        MessageFormatter messageFormatter =
                MessageFormatterDecoratorFactory.createMessageFormatterDecorator(msgContext);
        OutputStream out = (OutputStream) msgContext.getProperty(MessageContext.TRANSPORT_OUT);

        if (msgContext.isServerSide()) {
            OutTransportInfo transportInfo =
                (OutTransportInfo) msgContext.getProperty(Constants.OUT_TRANSPORT_INFO);

            if (transportInfo != null) {
                transportInfo.setContentType(
                messageFormatter.getContentType(msgContext, format, msgContext.getSoapAction()));
            } else {
                throw new AxisFault(Constants.OUT_TRANSPORT_INFO + " has not been set");
            }
        }

        try {
            messageFormatter.writeTo(msgContext, format, out, false);
            out.close();
        } catch (IOException e) {
            handleException("IO Error sending response message", e);
        }
    }

	private void sendRequestContent(final MessageContext msgContext) throws AxisFault {
        // NOTE:this a special case where, when the backend service expects content-length but,
        // we don't want the message to be built. If FORCE_HTTP_CONTENT_LENGTH and
        // COPY_CONTENT_LENGTH_FROM_INCOMING, we assume that the content coming from the
        // client side has not changed.
        boolean forceContentLength = msgContext.isPropertyTrue(NhttpConstants.FORCE_HTTP_CONTENT_LENGTH);
        boolean copyContentLength = msgContext.isPropertyTrue(PassThroughConstants.COPY_CONTENT_LENGTH_FROM_INCOMING);

        if (forceContentLength && copyContentLength &&
                msgContext.getProperty(PassThroughConstants.ORIGINAL_CONTENT_LENGTH) != null) {
            long contentLength = Long.parseLong((String) msgContext.getProperty(
                    PassThroughConstants.ORIGINAL_CONTENT_LENGTH));
            msgContext.setProperty(PassThroughConstants.PASS_THROUGH_MESSAGE_LENGTH, contentLength);
        }

		if (Boolean.TRUE.equals(msgContext.getProperty(PassThroughConstants.MESSAGE_BUILDER_INVOKED))) {
			synchronized (msgContext) {
				while (!Boolean.TRUE.equals(msgContext.getProperty(PassThroughConstants.WAIT_BUILDER_IN_STREAM_COMPLETE)) &&
	 				       !Boolean.TRUE.equals(msgContext.getProperty(PassThroughConstants.PASSTHRU_CONNECT_ERROR))) {
					try {
						msgContext.wait();
					} catch (InterruptedException e) {
						log.warn("Interrupted while waiting for message serialization to complete", e);
					}
				}
			}

			if (Boolean.TRUE.equals(msgContext.getProperty(PassThroughConstants.PASSTHRU_CONNECT_ERROR))) {
				return;
			}

			OutputStream out = (OutputStream) msgContext.getProperty(PassThroughConstants.BUILDER_OUTPUT_STREAM);
			if (out != null) {
				String disableChunking = (String) msgContext.getProperty(
                        PassThroughConstants.DISABLE_CHUNKING);
				String forceHttp10 = (String) msgContext.getProperty(
                        PassThroughConstants.FORCE_HTTP_1_0);
				Pipe pipe = (Pipe) msgContext.getProperty(PassThroughConstants.PASS_THROUGH_PIPE);
				
				if ("true".equals(disableChunking) || "true".equals(forceHttp10) ){
					MessageFormatter formatter =  MessageProcessorSelector.getMessageFormatter(
                            msgContext);
					OMOutputFormat format = PassThroughTransportUtils.getOMOutputFormat(msgContext);
                    OverflowableBlob serialized = null;
                    try {
                        serialized = setStreamAsTempData(formatter, msgContext, format);
                        msgContext.setProperty(PassThroughConstants.PASS_THROUGH_MESSAGE_LENGTH,
                                serialized.getSize());
                        serialized.writeTo(out);
                    } catch (IOException e) {
                    	 handleException("I/O error while serializing message", e);
                    } finally {
                        if (serialized != null) {
                            try {
                                serialized.release();
                            } catch (IOException ignored) {
                            }
                        }
                    }
                    pipe.setSerializationComplete(true);
				} else {
					if ((disableChunking == null || !"true".equals(disableChunking)) ||
					    (forceHttp10 == null || !"true".equals(forceHttp10))) {
						MessageFormatter formatter =  MessageProcessorSelector.getMessageFormatter(
                                msgContext);
						OMOutputFormat format = PassThroughTransportUtils.getOMOutputFormat(
                                msgContext);
						formatter.writeTo(msgContext, format, out, false);
					}
					
					if (isCompleteWithoutData(msgContext)) {
                        pipe.setSerializationCompleteWithoutData(true);
					} else {
						pipe.setSerializationComplete(true);
					}
				}
			}
		}
	}

    private boolean isCompleteWithoutData(MessageContext msgContext) {
        if (Boolean.TRUE.equals(msgContext.getProperty(
                PassThroughConstants.REST_GET_DELETE_INVOKE))) {
            return true;
        }

        return Boolean.TRUE.equals(msgContext.getProperty(PassThroughConstants.NO_ENTITY_BODY));
    }

    /**
     * Return the IOEventDispatch implementation to be used. This is overridden by the
     * SSL sender
     *
     * @param handler The passthru target handler instance
     * @param sslContext SSL context used by the sender or null
     * @param sslIOSessionHandler SSL session handler or null
     * @param config ConnectionConfig instance
     * @param trpOut Transport out description
     * @return an IOEventDispatch instance
     * @throws AxisFault on error
     */
    protected IOEventDispatch getEventDispatch(NHttpClientEventHandler handler,
                                               SSLContext sslContext,
                                               SSLSetupHandler sslIOSessionHandler,
                                               ConnectionConfig config,
                                               TransportOutDescription trpOut) throws AxisFault {

        return LoggingUtils.getClientIODispatch(handler, config);
    }

    /**
     * Always return null, as this implementation does not support outgoing SSL
     *
     * @param transportOut The transport out description
     * @return null
     * @throws AxisFault on error
     */
    protected SSLContext getSSLContext(TransportOutDescription transportOut) throws AxisFault {
        return null;
    }

    /**
     * Create the SSL IO Session handler to be used by this listener
     *
     * @param transportOut Transport out description
     * @return always null
     * @throws AxisFault on error
     */
    protected SSLSetupHandler getSSLSetupHandler(TransportOutDescription transportOut)
        throws AxisFault {
        return null;
    }

    public void submitResponse(MessageContext msgContext)
            throws IOException, HttpException {
        SourceConfiguration sourceConfiguration = (SourceConfiguration) msgContext.getProperty(
                        PassThroughConstants.PASS_THROUGH_SOURCE_CONFIGURATION);

        NHttpServerConnection conn = (NHttpServerConnection) msgContext.getProperty(
                PassThroughConstants.PASS_THROUGH_SOURCE_CONNECTION);
        if (conn == null) {
            ServerWorker serverWorker = (ServerWorker) msgContext.getProperty(
                    Constants.OUT_TRANSPORT_INFO);
            if (serverWorker != null) {
                MessageContext requestContext = serverWorker.getRequestContext();
                conn = (NHttpServerConnection) requestContext.getProperty(
                        PassThroughConstants.PASS_THROUGH_SOURCE_CONNECTION);
                sourceConfiguration = (SourceConfiguration) requestContext.getProperty(
                        PassThroughConstants.PASS_THROUGH_SOURCE_CONFIGURATION);
            } else {
                throw new IllegalStateException("Unable to correlate the response to a request");
            }
        }

        SourceRequest sourceRequest = SourceContext.getRequest(conn);

        SourceResponse sourceResponse = SourceResponseFactory.create(msgContext,
                sourceRequest, sourceConfiguration);

        sourceResponse.processChunkingOptions(msgContext);
        SourceContext.setResponse(conn, sourceResponse);

        Boolean noEntityBody = (Boolean) msgContext.getProperty(PassThroughConstants.NO_ENTITY_BODY);
        Pipe pipe = (Pipe) msgContext.getProperty(PassThroughConstants.PASS_THROUGH_PIPE);
        if ((noEntityBody == null || !noEntityBody) || pipe != null) {
            if (pipe == null) {
                pipe = new Pipe(sourceConfiguration.getBufferFactory().getBuffer(),
                        "Test", sourceConfiguration);
                msgContext.setProperty(PassThroughConstants.PASS_THROUGH_PIPE, pipe);
                msgContext.setProperty(PassThroughConstants.MESSAGE_BUILDER_INVOKED, Boolean.TRUE);
            }

            pipe.attachConsumer(conn);
            sourceResponse.connect(pipe);
        }

        Integer errorCode = (Integer) msgContext.getProperty(PassThroughConstants.ERROR_CODE);
        if (errorCode != null) {
            sourceResponse.setStatus(HttpStatus.SC_BAD_GATEWAY);
            SourceContext.get(conn).setShutDown(true);
        }

        ProtocolState state = SourceContext.getState(conn);
        if (state != null && state.compareTo(ProtocolState.REQUEST_DONE) <= 0) {
            // start sending the response
            if (noEntityBody != null && Boolean.TRUE == noEntityBody && pipe != null) {
                OutputStream out = pipe.getOutputStream();
                out.write(new byte[0]);
                pipe.setRawSerializationComplete(true);
                out.close();
            } else if (msgContext.isPropertyTrue(PassThroughConstants.MESSAGE_BUILDER_INVOKED) &&
                    pipe != null) {
                OutputStream out = pipe.getOutputStream();
                if (msgContext.isPropertyTrue(NhttpConstants.SC_ACCEPTED)) {
                    out.write(new byte[0]);
                } else {
                    MessageFormatter formatter = MessageProcessorSelector.getMessageFormatter(
                            msgContext);
                    OMOutputFormat format = PassThroughTransportUtils.getOMOutputFormat(msgContext);
                    formatter.writeTo(msgContext, format, out, false);
                }
                pipe.setSerializationComplete(true);
                out.close();
            }
            conn.requestOutput();
        } else {
            // nothing much to do as we have started the response already
            if (errorCode != null) {
                if (log.isDebugEnabled()) {
                    log.warn("A source connection is closed because of an " +
                            "error in target: " + conn);
                }
            } else {
                log.debug("A source connection is closed, because source handler " +
                        "is already in the process of writing a response while " +
                        "another response is submitted: " + conn);
            }

            SourceContext.updateState(conn, ProtocolState.CLOSED);
            sourceConfiguration.getSourceConnections().shutDownConnection(conn, true);
        }
    }

    public void pause() throws AxisFault {
        if (state != BaseConstants.STARTED) {
            return;
        }
        state = BaseConstants.PAUSED;
        log.info(namePrefix + " Sender Paused");
    }

    public void resume() throws AxisFault {
        if (state != BaseConstants.PAUSED) {
            return;
        }
        state = BaseConstants.STARTED;
        log.info(namePrefix + " Sender Resumed");
    }

    public void maintenanceShutdown(long millis) throws AxisFault {
        if (state != BaseConstants.STARTED) return;
        try {
            long start = System.currentTimeMillis();
            ioReactor.shutdown(millis);
            state = BaseConstants.STOPPED;
            log.info("Sender shutdown in : " + (System.currentTimeMillis() - start) / 1000 + "s");
        } catch (IOException e) {
            handleException("Error shutting down the IOReactor for maintenance", e);
        }
    }

    /**
     * Write the stream to a temporary storage and return a handle to the temporary storage
     *
     * @param messageFormatter Formatter used to serialize the message
     * @param msgContext Message to be serialized
     * @param format Output format
     *
     * @throws IOException if an exception occurred while writing data
     */
    private OverflowableBlob setStreamAsTempData(MessageFormatter messageFormatter,
                                     MessageContext msgContext,
                                     OMOutputFormat format) throws IOException {

        OverflowableBlob serialized = Blobs.createOverflowableBlob(4096, "http-nio_",
                ".dat", FileUtils.getTempDirectory());
        OutputStream out = serialized.getOutputStream();
        try {
            messageFormatter.writeTo(msgContext, format, out, true);
        } finally {
            out.close();
        }
        return serialized;
    }

    private boolean isBypass(String hostName) {
        for (String entry : proxyBypassList) {
            if (hostName.matches(entry)) {
                knownDirectHosts.add(hostName);
                return true;
            }
        }
        knownProxyHosts.add(hostName);
        return false;
    }

    private void handleException(String msg, Exception e) throws AxisFault {
        log.error(msg, e);
        throw new AxisFault(msg, e);
    }

    private void handleException(String msg) throws AxisFault {
        log.error(msg);
        throw new AxisFault(msg);
    }
}
