/*
 *  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 java.net.InetAddress;
import java.util.Comparator;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

import javax.xml.parsers.FactoryConfigurationError;

import org.apache.axiom.om.OMAbstractFactory;
import org.apache.axiom.soap.SOAP11Constants;
import org.apache.axiom.soap.SOAP12Constants;
import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axiom.soap.SOAPFactory;
import org.apache.axiom.util.UIDGenerator;
import org.apache.axis2.AxisFault;
import org.apache.axis2.Constants;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.builder.BuilderUtil;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.description.AxisService;
import org.apache.axis2.description.Parameter;
import org.apache.axis2.dispatchers.RequestURIBasedDispatcher;
import org.apache.axis2.engine.AxisEngine;
import org.apache.axis2.transport.RequestResponseTransport;
import org.apache.axis2.transport.TransportUtils;
import org.apache.axis2.transport.http.HTTPConstants;
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.HttpInetConnection;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.HttpVersion;
import org.apache.http.entity.BasicHttpEntity;
import org.apache.http.nio.NHttpServerConnection;
import org.apache.http.nio.reactor.ssl.SSLIOSession;
import org.apache.http.protocol.HTTP;
import org.apache.synapse.transport.nhttp.HttpCoreRequestResponseTransport;
import org.apache.synapse.transport.nhttp.NhttpConstants;
import org.apache.synapse.transport.nhttp.util.NhttpUtil;
import org.apache.synapse.transport.nhttp.util.RESTUtil;
import org.apache.synapse.transport.passthru.config.PassThroughConfiguration;
import org.apache.synapse.transport.passthru.config.SourceConfiguration;
import org.apache.synapse.transport.passthru.util.SourceResponseFactory;

/**
 * This is a worker thread for executing an incoming request in to the transport.
 */
public class ServerWorker implements Runnable {

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

    /** the incoming message to be processed */
    private org.apache.axis2.context.MessageContext msgContext = null;

    /** the http request */
    private SourceRequest request = null;

    /** The configuration of the receiver */
    private SourceConfiguration sourceConfiguration = null;

    private static final String SOAP_ACTION_HEADER = "SOAPAction";
        
    /** WSDL processor for Get requests */
    private HttpGetRequestProcessor httpGetRequestProcessor = null;
    
    private boolean isHttps = false;
  
    public ServerWorker(final SourceRequest request,
                        final SourceConfiguration sourceConfiguration) {
        this.request = request;
        this.sourceConfiguration = sourceConfiguration;
        this.isHttps = sourceConfiguration.isSsl();
        this.msgContext = createMessageContext(request);
        this.httpGetRequestProcessor = sourceConfiguration.getHttpGetRequestProcessor();

        // set these properties to be accessed by the engine
        msgContext.setProperty(
                PassThroughConstants.PASS_THROUGH_SOURCE_REQUEST, request);
        msgContext.setProperty(
                PassThroughConstants.PASS_THROUGH_SOURCE_CONFIGURATION, sourceConfiguration);
        msgContext.setProperty(PassThroughConstants.PASS_THROUGH_SOURCE_CONNECTION,
                request.getConnection());
    }

    @Override
    public void run() {
        try {
            if (log.isDebugEnabled()) {
                log.debug("Starting a new Server Worker instance");
            }
            ConfigurationContext cfgCtx = sourceConfiguration.getConfigurationContext();
            msgContext.setProperty(Constants.Configuration.HTTP_METHOD, request.getMethod());

            String method = request.getRequest() != null ?
                    request.getRequest().getRequestLine().getMethod().toUpperCase() : "";

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

            if (uri.contains(cfgCtx.getServicePath())) {
                // discard up to 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);
                    }
                }
            }

            String servicePrefix = oriUri.substring(0, oriUri.indexOf(uri));
            if (!servicePrefix.contains("://")) {
                HttpInetConnection conn = (HttpInetConnection) request.getConnection();
                InetAddress localAddress = conn.getLocalAddress();
                if (localAddress != null) {
                    servicePrefix = (sourceConfiguration.isSsl() ? "https://" : "http://") +
                            localAddress.getHostAddress() + ":" + conn.getLocalPort() + servicePrefix;
                }
            }

            msgContext.setProperty(PassThroughConstants.SERVICE_PREFIX, servicePrefix);

            msgContext.setTo(new EndpointReference(uri));
            msgContext.setProperty(PassThroughConstants.REST_URL_POSTFIX, uri);

            if (("GET".equals(method) || "DELETE".equals(method)) && httpGetRequestProcessor != null) {
                HttpResponse response = sourceConfiguration.getResponseFactory().newHttpResponse(
                        request.getVersion(), HttpStatus.SC_OK,
                        request.getConnection().getContext());

                // create a basic HttpEntity using the source channel of the response pipe
                BasicHttpEntity entity = new BasicHttpEntity();
                if (request.getVersion().greaterEquals(HttpVersion.HTTP_1_1)) {
                    entity.setChunked(true);
                }
                response.setEntity(entity);

                httpGetRequestProcessor.process(request.getRequest(), response, msgContext,
                        request.getConnection(), true);
            }

            //need special case to handle REST
            boolean restHandle = false;
            if (msgContext.getProperty(PassThroughConstants.REST_GET_DELETE_INVOKE) != null &&
                    (Boolean) msgContext.getProperty(PassThroughConstants.REST_GET_DELETE_INVOKE)) {
                msgContext.setProperty(HTTPConstants.HTTP_METHOD, method);
                msgContext.setServerSide(true);
                msgContext.setDoingREST(true);
                String contentTypeHeader = request.getHeaders().get(HTTP.CONTENT_TYPE);
                SOAPEnvelope soapEnvelope = this.handleRESTUrlPost(contentTypeHeader);
                processNonEntityEnclosingRESTHandler(soapEnvelope);
                restHandle = true;
            }

            // If WSDL generation is done then move out rather than hand over to
            // entity handle methods.
            SourceContext info = (SourceContext) request.getConnection().getContext().
                    getAttribute(SourceContext.CONNECTION_INFORMATION);
            Object getHandled = msgContext.getProperty(PassThroughConstants.GET_REQUEST_HANDLED);
            if ((info != null && info.getState().equals(ProtocolState.GET_REQUEST_COMPLETE)) ||
                    Boolean.TRUE.equals(getHandled)) {
                return;
            }

            //should be process normally
            if (!restHandle) {
                if (request.isEntityEnclosing()) {
                    processEntityEnclosingRequest();
                } else {
                    processNonEntityEnclosingRESTHandler(null);
                }
            }

            sendAck();
        } finally {
            cleanup();
        }
    }

	/**
	 * Method will setup the necessary parameters for the rest url post action
	 * 
	 * @param contentTypeHdr Content-type header value
	 * @return a SOAPEnvelope
	 * @throws FactoryConfigurationError
	 */
	private SOAPEnvelope handleRESTUrlPost(String contentTypeHdr) throws FactoryConfigurationError {
	    SOAPEnvelope soapEnvelope = null;
	    String contentType = contentTypeHdr!=null ?
                TransportUtils.getContentType(contentTypeHdr, msgContext) : null;

	    if (contentType == null || "".equals(contentType) ||
                HTTPConstants.MEDIA_TYPE_X_WWW_FORM.equals(contentType)) {

	        contentType = contentTypeHdr != null ?
                    contentTypeHdr : HTTPConstants.MEDIA_TYPE_X_WWW_FORM;
	        msgContext.setTo(new EndpointReference(request.getRequest().getRequestLine().getUri()));
	        msgContext.setProperty(Constants.Configuration.CONTENT_TYPE,contentType);
	        String charSetEncoding = BuilderUtil.getCharSetEncoding(contentType);
		    msgContext.setProperty(Constants.Configuration.CHARACTER_SET_ENCODING, charSetEncoding);

            try {
	            RESTUtil.dispatchAndVerify(msgContext);
	        } catch (AxisFault e1) {
	        	log.error("Error while building message for REST_URL request",e1);
	        }
	        
	        RequestURIBasedDispatcher requestDispatcher = new RequestURIBasedDispatcher();
			try {
				AxisService axisService = requestDispatcher.findService(msgContext);

				// the logic determines which service dispatcher to get invoke,
				// this will be determine
				// based on parameter defines at disableRestServiceDispatching,
				// and if super tenant invoke, with isTenantRequest
				// identifies whether the request to be dispatch to custom REST
				// Dispatcher Service.

				if (axisService == null) {
					String defaultSvcName = PassThroughConfiguration.getInstance().getStringProperty(
                            "nhttp.default.service", "__SynapseService");
					axisService =  msgContext.getConfigurationContext().getAxisConfiguration()
					                        .getService(defaultSvcName);
					msgContext.setAxisService(axisService);
				}
			} catch (AxisFault e) {
				handleException("Error processing " + request.getMethod() + " request for : " +
                        request.getUri(), e);
			}

	        try {
	        	 soapEnvelope = TransportUtils.createSOAPMessage(msgContext, null, contentType);
	           } catch (Exception e) {
	        	log.error("Error while building message for REST_URL request");
	        }
	        msgContext.setProperty(Constants.Configuration.MESSAGE_TYPE,
                    HTTPConstants.MEDIA_TYPE_APPLICATION_XML);
	    }
	    return soapEnvelope;
    }

    private void sendAck() {
        String respWritten = "";
        if (msgContext.getOperationContext() != null) {
            respWritten = (String) msgContext.getOperationContext().getProperty(
                    Constants.RESPONSE_WRITTEN);
        }
        
        if (msgContext.getProperty(PassThroughConstants.FORCE_SOAP_FAULT) != null) {
        	respWritten ="SKIP";
        }
        
        boolean respWillFollow = !Constants.VALUE_TRUE.equals(respWritten)
                && !"SKIP".equals(respWritten);
        boolean ack = (((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 || ack || forced || nioAck) {
            NHttpServerConnection conn = request.getConnection();
            SourceResponse sourceResponse;
            if (!nioAck) {
                msgContext.removeProperty(MessageContext.TRANSPORT_HEADERS);
                sourceResponse = SourceResponseFactory.create(msgContext,
                        request, sourceConfiguration);
                sourceResponse.setStatus(HttpStatus.SC_ACCEPTED);
            } else {
                if (log.isDebugEnabled()) {
                    log.debug("Sending ACK response with status "
                            + msgContext.getProperty(NhttpConstants.HTTP_SC)
                            + ", for MessageID : " + msgContext.getMessageID());
                }
                sourceResponse = SourceResponseFactory.create(msgContext,
                        request, sourceConfiguration);
                sourceResponse.setStatus(Integer.parseInt(
                        msgContext.getProperty(NhttpConstants.HTTP_SC).toString()));
            }

            SourceContext.setResponse(conn, sourceResponse);
            ProtocolState state = SourceContext.getState(conn);
            if (state != null && state.compareTo(ProtocolState.REQUEST_DONE) <= 0) {
                conn.requestOutput();
            } else {
                SourceContext.updateState(conn, ProtocolState.CLOSED);
                sourceConfiguration.getSourceConnections().shutDownConnection(conn);
            }
        }
    }

    private void processNonEntityEnclosingRESTHandler(SOAPEnvelope soapEnvelope) {
        String soapAction = request.getHeaders().get(SOAP_ACTION_HEADER);
        if ((soapAction != null) && soapAction.startsWith("\"") && soapAction.endsWith("\"")) {
            soapAction = soapAction.substring(1, soapAction.length() - 1);
        }

        msgContext.setSoapAction(soapAction);
        msgContext.setTo(new EndpointReference(request.getUri()));
        msgContext.setServerSide(true);
        msgContext.setDoingREST(true);
        if (!request.isEntityEnclosing()) {
        	msgContext.setProperty(PassThroughConstants.NO_ENTITY_BODY, Boolean.TRUE);
        }
        
        try {
        	if (soapEnvelope == null) {
        		 msgContext.setEnvelope(OMAbstractFactory.getSOAP11Factory().getDefaultEnvelope());
        	} else {
        		 msgContext.setEnvelope(soapEnvelope);
        	}

            AxisEngine.receive(msgContext);
        } catch (AxisFault axisFault) {
            handleException("Error processing " + request.getMethod() +
                " request for : " + request.getUri(), axisFault);
        }
    }

    private void processEntityEnclosingRequest() {
        try {
            String contentTypeHeader = request.getHeaders().get(HTTP.CONTENT_TYPE);
            contentTypeHeader = contentTypeHeader != null ? contentTypeHeader : inferContentType();

            String charSetEncoding = null;
            String contentType = null;

			if (contentTypeHeader != null) {
				charSetEncoding = BuilderUtil.getCharSetEncoding(contentTypeHeader);
				contentType = TransportUtils.getContentType(contentTypeHeader, msgContext);
			}
            // get the contentType of char encoding
            if (charSetEncoding == null) {
                charSetEncoding = MessageContext.DEFAULT_CHAR_SET_ENCODING;
            }
            String method = request.getRequest() != null ? request.getRequest().getRequestLine().getMethod().toUpperCase():"";
            

  

            msgContext.setTo(new EndpointReference(request.getUri()));
            msgContext.setProperty(HTTPConstants.HTTP_METHOD, method);
            msgContext.setProperty(Constants.Configuration.CHARACTER_SET_ENCODING, charSetEncoding);
            msgContext.setServerSide(true);
            
            msgContext.setProperty(Constants.Configuration.CONTENT_TYPE, contentTypeHeader);
            msgContext.setProperty(Constants.Configuration.MESSAGE_TYPE, contentType);

            if (contentTypeHeader ==null || HTTPTransportUtils.isRESTRequest(contentTypeHeader) || isRest(contentTypeHeader)) {
                msgContext.setProperty(PassThroughConstants.REST_REQUEST_CONTENT_TYPE, contentType);
                msgContext.setDoingREST(true);
                SOAPEnvelope soapEnvelope = this.handleRESTUrlPost(contentTypeHeader);
                msgContext.setProperty(PassThroughConstants.PASS_THROUGH_PIPE, request.getPipe());
                processNonEntityEnclosingRESTHandler(soapEnvelope);
    			return;
            } else {
                String soapAction = request.getHeaders().get(SOAP_ACTION_HEADER);

                int soapVersion = HTTPTransportUtils.
                        initializeMessageContext(msgContext, soapAction,
                                request.getUri(), contentTypeHeader);
                SOAPEnvelope envelope;

                if (soapVersion == 1) {
                    SOAPFactory fac = OMAbstractFactory.getSOAP11Factory();
                    envelope = fac.getDefaultEnvelope();
                } else {
                    SOAPFactory fac = OMAbstractFactory.getSOAP12Factory();
                    envelope = fac.getDefaultEnvelope();
                }

                msgContext.setEnvelope(envelope);
            }
            
           
            msgContext.setProperty(PassThroughConstants.PASS_THROUGH_PIPE, request.getPipe());
            AxisEngine.receive(msgContext);
        } catch (AxisFault axisFault) {
            handleException("Error processing " + request.getMethod() +
                " request for : " + request.getUri(), axisFault);
        } 
    }

    private boolean isRest(String contentType) {
        return contentType != null &&
                !contentType.contains(SOAP11Constants.SOAP_11_CONTENT_TYPE) &&
                !contentType.contains(SOAP12Constants.SOAP_12_CONTENT_TYPE);
    }

    /**
     * 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(SourceRequest request) {
    	Map excessHeaders = request.getExcessHeaders();
        ConfigurationContext cfgCtx = sourceConfiguration.getConfigurationContext();
        MessageContext msgContext =
                new MessageContext();
        msgContext.setMessageID(UIDGenerator.generateURNString());

        // Axis2 spawns a new thread to send a message if this is TRUE - and it has to
        // be the other way
        msgContext.setProperty(MessageContext.CLIENT_API_NON_BLOCKING,
                Boolean.FALSE);
        msgContext.setConfigurationContext(cfgCtx);
        NHttpServerConnection conn = request.getConnection();
        
        if (isHttps) {
            msgContext.setTransportOut(cfgCtx.getAxisConfiguration()
                .getTransportOut(Constants.TRANSPORT_HTTPS));
            msgContext.setTransportIn(cfgCtx.getAxisConfiguration()
                .getTransportIn(Constants.TRANSPORT_HTTPS));
            msgContext.setIncomingTransportName(Constants.TRANSPORT_HTTPS);
            SSLIOSession session = (SSLIOSession) conn.getContext().getAttribute(
                    SSLIOSession.SESSION_KEY);
            msgContext.setProperty("ssl.client.auth.cert.X509",
                    session.getAttribute("ssl.client.auth.cert.X509"));
        } 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);
        msgContext.setServerSide(true);
        msgContext.setProperty(
                Constants.Configuration.TRANSPORT_IN_URL, request.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);
            }
        });

        Set<Map.Entry<String, String>> entries = request.getHeaders().entrySet();
        for (Map.Entry<String, String> entry : entries) {
            headers.put(entry.getKey(), entry.getValue());
        }
        msgContext.setProperty(MessageContext.TRANSPORT_HEADERS, headers);
        msgContext.setProperty(NhttpConstants.EXCESS_TRANSPORT_HEADERS, excessHeaders);
        
        if (headers.get(HTTP.CONTENT_LEN) != null){
            msgContext.setProperty(PassThroughConstants.ORIGINAL_CONTENT_LENGTH,
                    headers.get(HTTP.CONTENT_LEN));
        }

        // Following section is required for throttling to work
        
        if (conn instanceof HttpInetConnection) {
            HttpInetConnection netConn = (HttpInetConnection) conn;
            InetAddress remoteAddress = netConn.getRemoteAddress();
            if (remoteAddress != null) {
                msgContext.setProperty(
                        MessageContext.REMOTE_ADDR, remoteAddress.getHostAddress());
                msgContext.setProperty(
                        NhttpConstants.REMOTE_HOST, NhttpUtil.getHostName(remoteAddress));
            }
        }

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

        return msgContext;
    }

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

        if (e == null) {
            e = new Exception(msg);
        }

        try {
            MessageContext faultContext =
                    MessageContextBuilder.createFaultMessageContext(
                    msgContext, e);
            msgContext.setProperty(PassThroughConstants.FORCE_SOAP_FAULT, Boolean.TRUE);
            AxisEngine.sendFault(faultContext);
            
        } catch (Exception ignored) {}
    }

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

    MessageContext getRequestContext() {
        return msgContext;
    }

    /**
     * Perform cleanup of ServerWorker
     */
    private void cleanup () {
        //clean threadLocal variables
        msgContext.destroyCurrentMessageContext();
    }
}
