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

import org.apache.axiom.util.UIDGenerator;
import org.apache.axis2.AxisFault;
import org.apache.axis2.Constants;
import org.apache.axis2.builder.Builder;
import org.apache.axis2.builder.BuilderUtil;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.description.Parameter;
import org.apache.axis2.engine.AxisEngine;
import org.apache.axis2.transport.RequestResponseTransport;
import org.apache.axis2.transport.TransportUtils;
import org.apache.axis2.transport.base.MetricsCollector;
import org.apache.axis2.transport.base.endpoint.URLEndpoint;
import org.apache.axis2.transport.base.endpoint.URLEndpointsConfiguration;
import org.apache.axis2.transport.http.HTTPTransportUtils;
import org.apache.axis2.util.MessageContextBuilder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.*;
import org.apache.http.nio.NHttpServerConnection;
import org.apache.http.protocol.HTTP;
import org.apache.synapse.transport.nhttp.util.NhttpUtil;
import org.apache.synapse.transport.nhttp.util.RESTUtil;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.util.Comparator;
import java.util.Map;
import java.util.TreeMap;

/**
 * Processes an incoming request through Axis2. An instance of this class would be created to
 * process each unique request
 */
public class ServerWorker implements Runnable {

    private static final Log log = LogFactory.getLog(ServerWorker.class);

    /** the incoming message to be processed */
    private MessageContext msgContext = null;
    /** the Axis2 configuration context */
    private ConfigurationContext cfgCtx = null;
    /** the message handler to be used */
    private ServerHandler serverHandler = null;
    /** the underlying http connection */
    private NHttpServerConnection conn = null;
    /** is this https? */
    private boolean isHttps = false;
    /** the http request */
    private HttpRequest request = null;
    /** the http response message (which the this would be creating) */
    private HttpResponse response = null;
    /** the input stream to read the incoming message body */
    private InputStream is = null;
    /** the output stream to write the response message body */
    private OutputStream os = null;
    /** the metrics collector */
    private MetricsCollector metrics = null;
    /** Weather we should do rest dispatching or not */
    private boolean isRestDispatching = true;
    /** WSDL processor for Get requests */
    private HttpGetRequestProcessor httpGetRequestProcessor = null;

    private URLEndpointsConfiguration endpointsConfiguration = null;
    
    private static final String SOAPACTION   = "SOAPAction";
    private static final String LOCATION     = "Location";
    private static final String CONTENT_TYPE = "Content-Type";
    private static final String TEXT_HTML    = "text/html";
    private static final String TEXT_XML     = "text/xml";
    /**
     * Save requesting user IP address for logging - even during response processing when
     * the connection may be closed
     */
    private String remoteAddress = null;

    /**
     * Create a new server side worker to process an incoming message and optionally begin creating
     * its output. This however does not force the processor to write a response back as the
     * traditional servlet service() method, but creates the background required to write the
     * response, if one would be created.
     *
     * @param listenerContext the listener configuration
     * @param conn the underlying http connection
     * @param serverHandler the handler of the server side messages
     * @param request the http request received (might still be in the process of being streamed)
     * @param is the stream input stream to read the request body
     * @param response the response to be populated if applicable
     * @param os the output stream to write the response body if one is applicable
     */
    public ServerWorker(ListenerContext listenerContext,
                        final NHttpServerConnection conn,
        final ServerHandler serverHandler,
        final HttpRequest request, final InputStream is,
        final HttpResponse response, final OutputStream os) {

        this.cfgCtx = listenerContext.getCfgCtx();
        this.conn = conn;
        this.isHttps = listenerContext.isSsl();
        this.metrics = listenerContext.getMetrics();
        this.serverHandler = serverHandler;
        this.request = request;
        this.response = response;
        this.is = is;
        this.os = os;
        this.msgContext = createMessageContext(request);
        this.isRestDispatching = listenerContext.isRestDispatching();
        this.httpGetRequestProcessor = listenerContext.getHttpGetRequestProcessor();
        this.endpointsConfiguration = listenerContext.getEndpoints();
    }

    /**
     * Create an Axis2 message context for the given http request. The request may be in the
     * process of being streamed
     * @param request the http request to be used to create the corresponding Axis2 message context
     * @return the Axis2 message context created
     */
    private MessageContext createMessageContext(HttpRequest request) {

        MessageContext msgContext = new MessageContext();
        msgContext.setMessageID(UIDGenerator.generateURNString());

        // There is a discrepency in what I thought, Axis2 spawns a new threads to
        // send a message if this is TRUE - and I want it to be the other way
        msgContext.setProperty(MessageContext.CLIENT_API_NON_BLOCKING, Boolean.FALSE);
        msgContext.setConfigurationContext(cfgCtx);
        if (isHttps) {
            msgContext.setTransportOut(cfgCtx.getAxisConfiguration()
                .getTransportOut(Constants.TRANSPORT_HTTPS));
            msgContext.setTransportIn(cfgCtx.getAxisConfiguration()
                .getTransportIn(Constants.TRANSPORT_HTTPS));
            msgContext.setIncomingTransportName(Constants.TRANSPORT_HTTPS);
        } else {
            msgContext.setTransportOut(cfgCtx.getAxisConfiguration()
                .getTransportOut(Constants.TRANSPORT_HTTP));
            msgContext.setTransportIn(cfgCtx.getAxisConfiguration()
                .getTransportIn(Constants.TRANSPORT_HTTP));
            msgContext.setIncomingTransportName(Constants.TRANSPORT_HTTP);
        }
        msgContext.setProperty(Constants.OUT_TRANSPORT_INFO, this);
        // the following statement causes the soap session services to be failing - ruwan        
        // msgContext.setServiceGroupContextId(UUIDGenerator.getUUID());
        msgContext.setServerSide(true);
        msgContext.setProperty(
            Constants.Configuration.TRANSPORT_IN_URL, request.getRequestLine().getUri());

        // http transport header names are case insensitive 
        Map<String, String> headers = new TreeMap<String, String>(new Comparator<String>() {
            @Override
            public int compare(String o1, String o2) {
                return o1.compareToIgnoreCase(o2);
            }
        });
        
        for (Header header : request.getAllHeaders()) {
            headers.put(header.getName(), header.getValue());
        }
        msgContext.setProperty(MessageContext.TRANSPORT_HEADERS, headers);

        // find the remote party IP address and set it to the message context
        if (conn instanceof HttpInetConnection) {
            HttpInetConnection inetConn = (HttpInetConnection) conn;
            InetAddress remoteAddr = inetConn.getRemoteAddress();
            if (remoteAddr != null) {
                msgContext.setProperty(
                        MessageContext.REMOTE_ADDR, remoteAddr.getHostAddress());
                msgContext.setProperty(
                        NhttpConstants.REMOTE_HOST, NhttpUtil.getHostName(remoteAddr));
                remoteAddress = remoteAddr.getHostAddress();
            }
        }

        msgContext.setProperty(RequestResponseTransport.TRANSPORT_CONTROL,
                new HttpCoreRequestResponseTransport(msgContext));

        msgContext.setProperty(ServerHandler.SERVER_CONNECTION_DEBUG,
            conn.getContext().getAttribute(ServerHandler.SERVER_CONNECTION_DEBUG));

        msgContext.setProperty(NhttpConstants.NHTTP_INPUT_STREAM, is);
        msgContext.setProperty(NhttpConstants.NHTTP_OUTPUT_STREAM, os);

        return msgContext;
    }

    /**
     * Process the incoming request
     */
    @Override
    @SuppressWarnings({"unchecked"})
    public void run() {

        String method = request.getRequestLine().getMethod().toUpperCase();
        msgContext.setProperty(Constants.Configuration.HTTP_METHOD,
            request.getRequestLine().getMethod());

        if (NHttpConfiguration.getInstance().isHttpMethodDisabled(method)) {
            handleException("Unsupported method : " + method, null);
        }

        String uri = request.getRequestLine().getUri();
        String oriUri = uri;

        if (uri.indexOf(cfgCtx.getServicePath()) != -1) {
            // discard upto servicePath
            uri = uri.substring(uri.indexOf(cfgCtx.getServicePath()) +
                cfgCtx.getServicePath().length());
            // discard [proxy] service name if any
            int pos = uri.indexOf("/", 1);
            if (pos > 0) {
                uri = uri.substring(pos);
            } else {
                pos = uri.indexOf("?");
                if (pos != -1) {
                    uri = uri.substring(pos);
                } else {
                    uri = "";
                }
            }
        } else {
            // remove any absolute prefix if any
            int pos = uri.indexOf("://");
            if (pos != -1) {
                uri = uri.substring(pos + 3);
            }
            pos = uri.indexOf("/");
            if (pos != -1) {
                uri = uri.substring(pos + 1);
            }
        }
        msgContext.setProperty(NhttpConstants.REST_URL_POSTFIX, uri);
        String servicePrefix = oriUri.substring(0, oriUri.indexOf(uri));
        if (servicePrefix.indexOf("://") == -1) {
            // If the URL in the request line is not absolute, then we first try to get the Host
            // header to build the absolute URL. We only use the local network address/port of
            // the HTTP connection if no Host header is present. This is not only consistent, but
            // also avoids the overhead of the InetAddress#getHostName() method.
            String host;
            Header hostHeader = request.getFirstHeader(HTTP.TARGET_HOST);
            if (hostHeader != null) {
                host = hostHeader.getValue();
            } else {
                HttpInetConnection inetConn = (HttpInetConnection) conn;
                InetAddress localAddr = inetConn.getLocalAddress();
                if (localAddr != null) {
                    host = localAddr.getHostName() + ":" + inetConn.getLocalPort();
                } else {
                    host = null;
                }
            }
            if (host != null) {
                servicePrefix = (isHttps ? "https://" : "http://") + host + servicePrefix;
            }
        }
        msgContext.setProperty(NhttpConstants.SERVICE_PREFIX, servicePrefix);

        if ("GET".equals(method)) {           
            httpGetRequestProcessor.process(request, response,
                    msgContext, conn, os, isRestDispatching);
        } else if ("POST".equals(method)) {
            processEntityEnclosingMethod();
        } else if ("PUT".equals(method)) {
            processEntityEnclosingMethod();
        } else if ("HEAD".equals(method)) {
            processNonEntityEnclosingMethod();
        } else if ("OPTIONS".equals(method)) {
            processNonEntityEnclosingMethod();
        } else if ("DELETE".equals(method)) {
            processGetAndDelete("DELETE");
        } else if ("TRACE".equals(method)) {
            processNonEntityEnclosingMethod();
        } else {
            handleException("Unsupported method : " + method, null);
        }

        // here the RequestResponseTransport plays an important role when it comes to
        // dual channel invocation. This is becasue we need to ACK to the request once the request
        // is received to synapse. Otherwise we will not be able to support the single channel
        // invocation within the actual service and synapse for a dual channel request from the
        // client.
        if (isAckRequired()) {
            String respWritten = "";
            if (msgContext.getOperationContext() != null) {
                respWritten = (String) msgContext.getOperationContext().getProperty(
                        Constants.RESPONSE_WRITTEN);
            }
            boolean respWillFollow = !Constants.VALUE_TRUE.equals(respWritten)
                    && !"SKIP".equals(respWritten);
            boolean acked = (((RequestResponseTransport) msgContext.getProperty(
                    RequestResponseTransport.TRANSPORT_CONTROL)).getStatus()
                    == RequestResponseTransport.RequestResponseTransportStatus.ACKED);
            boolean forced = msgContext.isPropertyTrue(NhttpConstants.FORCE_SC_ACCEPTED);
            boolean nioAck = msgContext.isPropertyTrue("NIO-ACK-Requested", false);

            if (respWillFollow || acked || forced || nioAck) {

                if (!nioAck) {
                    if (log.isDebugEnabled()) {
                        log.debug("Sending 202 Accepted response for MessageID : " +
                                msgContext.getMessageID() +
                                " response written : " + respWritten +
                                " response will follow : " + respWillFollow +
                                " acked : " + acked + " forced ack : " + forced);
                    }
                    response.setStatusCode(HttpStatus.SC_ACCEPTED);
                } else {
                    if (log.isDebugEnabled()) {
                        log.debug("Sending ACK response with status "
                                + msgContext.getProperty(NhttpConstants.HTTP_SC)
                                + ", for MessageID : " + msgContext.getMessageID());
                    }
                    response.setStatusCode(Integer.parseInt(
                            msgContext.getProperty(NhttpConstants.HTTP_SC).toString()));
                    Map<String, String> responseHeaders = (Map<String, String>)
                            msgContext.getProperty(MessageContext.TRANSPORT_HEADERS);
                    if (responseHeaders != null) {
                        for (Map.Entry<String,String> entry : responseHeaders.entrySet()) {
                            response.addHeader(entry.getKey(), entry.getValue());
                        }
                    }
                }

                if (metrics != null) {
                    metrics.incrementMessagesSent();
                }

                try {
                    serverHandler.commitResponse(conn, response);

                } catch (HttpException e) {
                    if (metrics != null) {
                        metrics.incrementFaultsSending();
                    }
                    handleException("Unexpected HTTP protocol error : " + e.getMessage(), e);
                } catch (ConnectionClosedException e) {
                    if (metrics != null) {
                        metrics.incrementFaultsSending();
                    }
                    log.warn("Connection closed by client (Connection closed)");
                } catch (IllegalStateException e) {
                    if (metrics != null) {
                        metrics.incrementFaultsSending();
                    }
                    log.warn("Connection closed by client (Buffer closed)");
                } catch (IOException e) {
                    if (metrics != null) {
                        metrics.incrementFaultsSending();
                    }
                    handleException("IO Error sending response message", e);
                } catch (Exception e) {
                    if (metrics != null) {
                        metrics.incrementFaultsSending();
                    }
                    handleException("General Error sending response message", e);
                }

                if (is != null) {
                    try {
                        is.close();
                    } catch (IOException ignore) {}
                }

                // make sure that the output stream is flushed and closed properly
                try {
                    os.flush();
                    os.close();
                } catch (IOException ignore) {}
            }
        }
    }

    private boolean isAckRequired() {

        // This condition is a bit complex but cannot simplify any further.
        if (msgContext != null) {
            if (msgContext.getOperationContext() != null &&
                    (!msgContext.getOperationContext().getAxisOperation().isControlOperation() ||
                            msgContext.isPropertyTrue(NhttpConstants.FORCE_SC_ACCEPTED))) {

                return true;
            } else if (msgContext.isPropertyTrue("NIO-ACK-Requested", false)) {
                return true;
            }
        }

        return false;
    }

    /**
     *
     */
    private void processEntityEnclosingMethod() {

        try {
            Header contentType = request.getFirstHeader(HTTP.CONTENT_TYPE);
            String contentTypeStr = contentType != null ?
                    contentType.getValue() : inferContentType();

            String charSetEncoding = BuilderUtil.getCharSetEncoding(contentTypeStr);
            msgContext.setProperty(
                    Constants.Configuration.CHARACTER_SET_ENCODING, charSetEncoding);
            boolean eprFound = false;
            if (endpointsConfiguration != null) {
                URLEndpoint epr = endpointsConfiguration.getEndpoint(request.getRequestLine().getUri());
                if (epr != null) {
                    eprFound = true;
                    String type = TransportUtils.getContentType(contentTypeStr, msgContext);
                    msgContext.setProperty(Constants.Configuration.MESSAGE_TYPE, type);
                    epr.setParameters(msgContext);

                    Builder builder = epr.getBuilder(type);
                    if (HTTPTransportUtils.isRESTRequest(contentTypeStr)) {
                        RESTUtil.processPOSTRequest(msgContext, is, os,
                                request.getRequestLine().getUri(), contentType, builder, isRestDispatching);
                    } else {

                        Header soapAction = request.getFirstHeader(SOAPACTION);
                        HTTPTransportUtils.processHTTPPostRequest(
                                msgContext, is,
                                os,
                                contentTypeStr, builder,
                                (soapAction != null ? soapAction.getValue() : null),
                                request.getRequestLine().getUri());
                    }
                }
            }

            if (!eprFound) {
                if (HTTPTransportUtils.isRESTRequest(contentTypeStr)) {
                    RESTUtil.processPOSTRequest(msgContext, is, os,
                            request.getRequestLine().getUri(), contentType, isRestDispatching);
                } else {

                    Header soapAction = request.getFirstHeader(SOAPACTION);
                    HTTPTransportUtils.processHTTPPostRequest(
                            msgContext, is,
                            os,
                            contentTypeStr,
                            (soapAction != null ? soapAction.getValue() : null),
                            request.getRequestLine().getUri());
                }

            }
        } catch (AxisFault e) {
            handleException("Error processing POST request ", e);
        }
    }

    private String inferContentType() {
        Parameter param = cfgCtx.getAxisConfiguration().
                getParameter(NhttpConstants.REQUEST_CONTENT_TYPE);
        if (param != null) {
            return param.getValue().toString();
        }
        return null;
    }

    /**
     * Process HEAD, DELETE, TRACE, OPTIONS
     */
    private void processNonEntityEnclosingMethod() {

        try {
            RESTUtil.processURLRequest(
                msgContext, os, null,
                request.getRequestLine().getUri());

        } catch (AxisFault e) {
            handleException("Error processing " + request.getRequestLine().getMethod() +
                " request for : " + request.getRequestLine().getUri(), e);
        }
    }


    /**
     * Calls the RESTUtil to process GET and DELETE Request
     *
     * @param method HTTP method, either GET or DELETE
     */
    private void processGetAndDelete(String method) {
        try {
            Header contentType = request.getFirstHeader(HTTP.CONTENT_TYPE);
            String contentTypeStr = contentType != null ?
                    contentType.getValue() : inferContentType();

            boolean eprFound = false;
            if (endpointsConfiguration != null) {
                URLEndpoint epr = endpointsConfiguration.getEndpoint(request.getRequestLine().getUri());
                if (epr != null) {
                    eprFound = true;
                    String type = TransportUtils.getContentType(contentTypeStr, msgContext);
                    msgContext.setProperty(Constants.Configuration.MESSAGE_TYPE, type);

                    epr.setParameters(msgContext);

                    Builder builder = epr.getBuilder(type);
                    RESTUtil.processGetAndDeleteRequest(
                            msgContext, os, request.getRequestLine().getUri(),
                            request.getFirstHeader(HTTP.CONTENT_TYPE), builder,
                            method, isRestDispatching);
                }
            }

            if (!eprFound) {
                RESTUtil.processGetAndDeleteRequest(
                        msgContext, os, request.getRequestLine().getUri(),
                        request.getFirstHeader(HTTP.CONTENT_TYPE), method, isRestDispatching);
            }
            // do not let the output stream close (as by default below) since
            // we are serving this GET/DELETE request through the Synapse engine
        } catch (AxisFault axisFault) {
            handleException("Error processing " + method + " request for: " +
                    request.getRequestLine().getUri(), axisFault);
        }

    }

    private void handleException(String msg, Exception e) {
        
        if (e == null) {
            log.error(msg);
        } else {
            log.error(msg, e);
        }
        Exception newException = e;
        if (e == null) {
            newException = new Exception(msg);
        }

        try {
            MessageContext faultContext = MessageContextBuilder.createFaultMessageContext(
                    msgContext, newException);
            AxisEngine.sendFault(faultContext);

        } catch (Exception ex) {
            response.setStatusCode(HttpStatus.SC_INTERNAL_SERVER_ERROR);
            response.addHeader(CONTENT_TYPE, TEXT_XML);
            conn.getContext().setAttribute(NhttpConstants.FORCE_CONNECTION_CLOSE, true);
            serverHandler.commitResponseHideExceptions(conn, response);

            try {
                if (is != null) {
                    try {
                        is.close();
                    } catch (IOException ignore) {}
                }

                String body = "<html><body><h1>" + "Failed to process the request" +
                         "</h1><p>"+ msg + "</p>";
                if (e != null) {
                    body = body + "<p>"+ e.getMessage() + "</p></body></html>";
                }
                if (ex != null) {
                    body = body + "<p>"+ ex.getMessage() + "</p></body></html>";
                }
                os.write(body.getBytes());
                os.flush();
                os.close();
            } catch (IOException ignore) {}
        }
    }

    public HttpResponse getResponse() {
        return response;
    }

    public OutputStream getOutputStream() {
        return os;
    }

    public InputStream getIs() {
        return is;
    }

    public ServerHandler getServiceHandler() {
        return serverHandler;
    }

    public NHttpServerConnection getConn() {
        return conn;
    }

    public String getRemoteAddress() {
        return remoteAddress;
    }

}
