<commit_before>/*
 * 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.ode.axis2;

import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axis2.AxisFault;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.client.OperationClient;
import org.apache.axis2.client.Options;
import org.apache.axis2.client.ServiceClient;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.description.AxisService;
import org.apache.axis2.description.OutInAxisOperation;
import org.apache.axis2.description.OutOnlyAxisOperation;
import org.apache.axis2.engine.AxisConfiguration;
import org.apache.axis2.transport.jms.JMSConstants;
import org.apache.axis2.wsdl.WSDLConstants;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.neethi.Policy;
import org.apache.neethi.PolicyEngine;
import org.apache.ode.axis2.util.SoapMessageConverter;
import org.apache.ode.bpel.epr.EndpointFactory;
import org.apache.ode.bpel.epr.MutableEndpoint;
import org.apache.ode.bpel.epr.WSAEndpoint;
import org.apache.ode.bpel.iapi.BpelServer;
import org.apache.ode.bpel.iapi.Message;
import org.apache.ode.bpel.iapi.MessageExchange;
import org.apache.ode.bpel.iapi.MessageExchange.FailureType;
import org.apache.ode.bpel.iapi.PartnerRoleMessageExchange;
import org.apache.ode.bpel.iapi.ProcessConf;
import org.apache.ode.bpel.iapi.Scheduler;
import org.apache.ode.il.OMUtils;
import org.apache.ode.utils.*;
import org.apache.ode.utils.fs.FileUtils;
import org.apache.ode.utils.uuid.UUID;
import org.apache.ode.utils.wsdl.Messages;
import org.apache.rampart.RampartMessageData;
import org.w3c.dom.Document;
import org.w3c.dom.Element;

import javax.wsdl.Definition;
import javax.wsdl.Fault;
import javax.wsdl.Operation;
import javax.xml.namespace.QName;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;

/**
 * Acts as a service not provided by ODE. Used mainly for invocation as a way to maintain the WSDL description of used
 * services.
 *
 * @author <NAME> <mriou at apache dot org>
 */
public class SoapExternalService implements ExternalService {

    private static final Log __log = LogFactory.getLog(SoapExternalService.class);

    private static final org.apache.ode.utils.wsdl.Messages msgs = Messages.getMessages(Messages.class);


    private static ThreadLocal<ServiceClient> _cachedClients = new ThreadLocal<ServiceClient>();
    private WatchDog<Map, OptionsObserver> _axisOptionsWatchDog;
    private WatchDog<Long, ServiceFileObserver> _axisServiceWatchDog;
    private ConfigurationContext _configContext;


    private ExecutorService _executorService;
    private Definition _definition;
    private QName _serviceName;
    private String _portName;
    protected WSAEndpoint endpointReference;
    private AxisConfiguration _axisConfig;
    private SoapMessageConverter _converter;
    private Scheduler _sched;
    private BpelServer _server;
    private ProcessConf _pconf;

    public SoapExternalService(ProcessConf pconf, QName serviceName, String portName, ExecutorService executorService,
                               AxisConfiguration axisConfig, Scheduler sched, BpelServer server) throws AxisFault {
        _definition = pconf.getDefinitionForService(serviceName);
        _serviceName = serviceName;
        _portName = portName;
        _executorService = executorService;
        _axisConfig = axisConfig;
        _sched = sched;
        _converter = new SoapMessageConverter(_definition, serviceName, portName);
        _server = server;
        _pconf = pconf;

        File fileToWatch = new File(_pconf.getBaseURI().resolve(_serviceName.getLocalPart() + ".axis2"));
        _axisServiceWatchDog = WatchDog.watchFile(fileToWatch, new ServiceFileObserver(fileToWatch));
        _axisOptionsWatchDog = new WatchDog<Map, OptionsObserver>(new EndpointPropertiesMutable(), new OptionsObserver());
        _configContext = new ConfigurationContext(_axisConfig);

        // initial endpoint reference
        Element eprElmt = ODEService.genEPRfromWSDL(_definition, serviceName, portName);
        if (eprElmt == null)
            throw new IllegalArgumentException(msgs.msgPortDefinitionNotFound(serviceName, portName));
        endpointReference = EndpointFactory.convertToWSA(ODEService.createServiceRef(eprElmt));
    }

    public void invoke(final PartnerRoleMessageExchange odeMex) {
        boolean isTwoWay = odeMex.getMessageExchangePattern() == org.apache.ode.bpel.iapi.MessageExchange.MessageExchangePattern.REQUEST_RESPONSE;
        try {

            ServiceClient client = getServiceClient();

            // Override options are passed to the axis MessageContext so we can
            // retrieve them in our session out changeHandler.
            MessageContext mctx = new MessageContext();
            /* make the given options the parent so it becomes the defaults of the MessageContexgt. That allows the user to override
            *  specific options on a given message context and not affect the overall options.
            */
            mctx.getOptions().setParent(client.getOptions());
            writeHeader(mctx, odeMex);

            _converter.createSoapRequest(mctx, odeMex.getRequest(), odeMex.getOperation());

            SOAPEnvelope soapEnv = mctx.getEnvelope();
            EndpointReference axisEPR = new EndpointReference(((MutableEndpoint) odeMex.getEndpointReference())
                    .getUrl());
            if (__log.isDebugEnabled()) {
                __log.debug("Axis2 sending message to " + axisEPR.getAddress() + " using MEX " + odeMex);
                __log.debug("Message: " + soapEnv);
            }

            final OperationClient operationClient = client.createClient(isTwoWay ? ServiceClient.ANON_OUT_IN_OP
                    : ServiceClient.ANON_OUT_ONLY_OP);
            operationClient.addMessageContext(mctx);
            // this Options can be alter without impacting the ServiceClient options (which is a requirement)
            Options operationOptions = operationClient.getOptions();

            // provide HTTP credentials if any
            AuthenticationHelper.setHttpAuthentication(odeMex, operationOptions);

            operationOptions.setAction(mctx.getSoapAction());
            operationOptions.setTo(axisEPR);

            if (isTwoWay) {
                final String mexId = odeMex.getMessageExchangeId();
                final Operation operation = odeMex.getOperation();

                // Defer the invoke until the transaction commits.
                _sched.registerSynchronizer(new Scheduler.Synchronizer() {
                    public void afterCompletion(boolean success) {
                        // If the TX is rolled back, then we don't send the request.
                        if (!success) return;

                        // The invocation must happen in a separate thread, holding on the afterCompletion
                        // blocks other operations that could have been listed there as well.
                        _executorService.submit(new Callable<Object>() {
                            public Object call() throws Exception {
                                try {
                                    operationClient.execute(true);
                                    MessageContext response = operationClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);
                                    MessageContext flt = operationClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_FAULT_VALUE);
                                    if (response != null && __log.isDebugEnabled())
                                        __log.debug("Service response:\n" + response.getEnvelope().toString());

                                    if (flt != null) {
                                        reply(mexId, operation, flt, true);
                                    } else {
                                        reply(mexId, operation, response, response.isFault());
                                    }
                                } catch (Throwable t) {
                                    String errmsg = "Error sending message (mex=" + odeMex + "): " + t.getMessage();
                                    __log.error(errmsg, t);
                                    replyWithFailure(mexId, MessageExchange.FailureType.COMMUNICATION_ERROR, errmsg, null);
                                }
                                return null;
                            }
                        });
                    }

                    public void beforeCompletion() {
                    }
                });
                odeMex.replyAsync();

            } else { /** one-way case * */
                _executorService.submit(new Callable<Object>() {
                    public Object call() throws Exception {
                        operationClient.execute(false);
                        return null;
                    }
                });
                odeMex.replyOneWayOk();
            }
        } catch (AxisFault axisFault) {
            String errmsg = "Error sending message to Axis2 for ODE mex " + odeMex;
            __log.error(errmsg, axisFault);
            odeMex.replyWithFailure(MessageExchange.FailureType.COMMUNICATION_ERROR, errmsg, null);
        }
    }

    private ServiceClient getServiceClient() throws AxisFault {
        try {
            // call manually the check procedure
            // we dont want a dedicated thread for that
            _axisServiceWatchDog.check();
            _axisOptionsWatchDog.check();
        } catch (RuntimeException e) {
            throw AxisFault.makeFault(e.getCause() != null ? e.getCause() : e);
        }

        // apply the options to the service client
        ServiceClient serviceClient = _cachedClients.get();
        if (serviceClient == null) {
            serviceClient = new ServiceClient(_configContext, null);
            _cachedClients.set(serviceClient);
        }
        AxisService anonymousService = _axisServiceWatchDog.getObserver().anonymousService;
        serviceClient.setAxisService(anonymousService);
        serviceClient.setOptions(_axisOptionsWatchDog.getObserver().options);

        applySecuritySettings(serviceClient);

        return serviceClient;
    }
    private void applySecuritySettings(ServiceClient serviceClient) throws AxisFault {
        Options options = serviceClient.getOptions();
        if (options.getProperty(Properties.PROP_SECURITY_POLICY) != null) {
            String policy = (String) options.getProperty(Properties.PROP_SECURITY_POLICY);
            // if the policy path is relative, the full uri is resolved against the process conf directory
            URI policyUri;
            if (FileUtils.isRelative(policy)) {
                policyUri = _pconf.getBaseURI().resolve(policy);
            } else {
                policyUri = new File(policy).toURI();
            }
            try {
                InputStream policyStream = policyUri.toURL().openStream();
                try {
                    Policy policyDoc = PolicyEngine.getPolicy(policyStream);
                    options.setProperty(RampartMessageData.KEY_RAMPART_POLICY, policyDoc);

                    // make sure the proper modules are engaged
                    if (!serviceClient.getAxisService().getAxisConfiguration().isEngaged("rampart")
                            && !serviceClient.getAxisService().isEngaged("rampart")) {
                        serviceClient.engageModule("rampart");
                    }
                } finally {
                    policyStream.close();
                }
            } catch (IOException e) {
                throw new IllegalArgumentException("Exception while parsing policy: " + policyUri, e);
            }
        }
    }

    /**
     * Extracts the action to be used for the given operation.  It first checks to see
     * if a value is specified using WS-Addressing in the portType, it then falls back onto
     * getting it from the SOAP Binding.
     *
     * @param operation the name of the operation to get the Action for
     * @return The action value for the specified operation
     */
    private String getAction(String operation) {
        String action = _converter.getWSAInputAction(operation);
        if (action == null || "".equals(action)) {
            action = _converter.getSoapAction(operation);
        }
        return action;
    }

    /**
     * Extracts endpoint information from ODE message exchange to stuff them into Axis MessageContext.
     */
    private void writeHeader(MessageContext ctxt, PartnerRoleMessageExchange odeMex) {
        Options options = ctxt.getOptions();
        WSAEndpoint targetWSAEPR = EndpointFactory.convertToWSA((MutableEndpoint) odeMex.getEndpointReference());
        WSAEndpoint myRoleWSAEPR = EndpointFactory.convertToWSA((MutableEndpoint) odeMex.getMyRoleEndpointReference());
        WSAEndpoint targetEPR = new WSAEndpoint(targetWSAEPR);
        
        EndpointReference replyEPR = null; 

        String partnerSessionId = odeMex.getProperty(MessageExchange.PROPERTY_SEP_PARTNERROLE_SESSIONID);
        String myRoleSessionId = odeMex.getProperty(MessageExchange.PROPERTY_SEP_MYROLE_SESSIONID);

        if (partnerSessionId != null) {
            if (__log.isDebugEnabled()) {
                __log.debug("Partner session identifier found for WSA endpoint: " + partnerSessionId);
            }
            targetEPR.setSessionId(partnerSessionId);
        }
        options.setProperty("targetSessionEndpoint", targetEPR);

        if (myRoleWSAEPR != null) {
            WSAEndpoint myRoleEPR = new WSAEndpoint(myRoleWSAEPR);
            if (myRoleSessionId != null) {
                if (__log.isDebugEnabled()) {
                    __log.debug("MyRole session identifier found for myrole (callback) WSA endpoint: "
                            + myRoleSessionId);
                }
                myRoleEPR.setSessionId(myRoleSessionId);
            }
            options.setProperty("callbackSessionEndpoint", myRoleEPR);

            // Map My Session ID to JMS Correlation ID
            Document callbackEprXml = odeMex.getMyRoleEndpointReference().toXML();
            Element serviceElement = callbackEprXml.getDocumentElement();
            
        	if (myRoleSessionId != null) {
	            options.setProperty(JMSConstants.JMS_COORELATION_ID, myRoleSessionId);
        	} else {
                if (myRoleWSAEPR.getSessionId() != null) {
                    options.setProperty(JMSConstants.JMS_COORELATION_ID, myRoleSessionId);
                }
        	}

            Element address = DOMUtils.findChildByName(serviceElement, 
            		new QName(Namespaces.WS_ADDRESSING_NS, "Address"), true);
            if (__log.isDebugEnabled()) {
                __log.debug("The system-defined wsa address is : "
                        + address);
            }
            if (address != null) {
	            String url = address.getTextContent();
	            String jmsDestination = (String) options.getProperty(JMSConstants.REPLY_PARAM);
                if (__log.isDebugEnabled()) {
                    __log.debug("The user-defined JMS replyTo destination is: "
                            + jmsDestination);
                    __log.debug("The user-defined JMS wait timeout is: "
                            + options.getProperty(JMSConstants.JMS_WAIT_REPLY));
                }
	            if (jmsDestination == null || "".equals(jmsDestination.trim())) {
	            	// If the REPLY_PARAM property is not user-defined, then use the default value from myRole EPR	            	
		            int startIndex = url.indexOf("jms:/");
		            if (startIndex != -1) {
		            	startIndex += "jms:/".length();
		            	if (url.charAt(startIndex + 1) == '/') {
		            		// startIndex++; // treat "/" as valid start character for queue name
		            	}
		            	if (url.startsWith("dynamic")) {
		            		startIndex += "dynamicQueues".length();
		            	}
		            	int jmsEndIndex = url.indexOf("?", startIndex);
		            	if (jmsEndIndex == -1) {
		            		jmsEndIndex = url.length();
		            	}
		        		jmsDestination = url.substring(startIndex, jmsEndIndex);
		                options.setProperty(JMSConstants.REPLY_PARAM, jmsDestination);
		            } else {
			            startIndex = url.indexOf("http://");
			            if (startIndex != -1) {
			            	startIndex = url.indexOf("/processes/");
			            	if (startIndex != -1) {
			            		startIndex += "/processes/".length();
			            		jmsDestination = url.substring(startIndex);
				                options.setProperty(JMSConstants.REPLY_PARAM, jmsDestination);
			            	}
			            }
		            }
		            replyEPR = new EndpointReference(url);
	            } else {
	            	replyEPR = new EndpointReference("jms:/" + jmsDestination);
	            }
            }
        } else {
            __log.debug("My-Role EPR not specified, SEP will not be used.");
        }

        String action = getAction(odeMex.getOperationName());
        ctxt.setSoapAction(action);

        if (replyEPR == null) {
        	if (MessageExchange.MessageExchangePattern.REQUEST_RESPONSE == odeMex.getMessageExchangePattern()) {
	        	replyEPR = new EndpointReference(Namespaces.WS_ADDRESSING_ANON_URI);
        	}
        }
        if (replyEPR != null) {<commit_msg>Don't change the reply-to header in the case of HTTP URLs.<commit_after>/*
 * 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.ode.axis2;

import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axis2.AxisFault;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.client.OperationClient;
import org.apache.axis2.client.Options;
import org.apache.axis2.client.ServiceClient;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.description.AxisService;
import org.apache.axis2.description.OutInAxisOperation;
import org.apache.axis2.description.OutOnlyAxisOperation;
import org.apache.axis2.engine.AxisConfiguration;
import org.apache.axis2.transport.jms.JMSConstants;
import org.apache.axis2.wsdl.WSDLConstants;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.neethi.Policy;
import org.apache.neethi.PolicyEngine;
import org.apache.ode.axis2.util.SoapMessageConverter;
import org.apache.ode.bpel.epr.EndpointFactory;
import org.apache.ode.bpel.epr.MutableEndpoint;
import org.apache.ode.bpel.epr.WSAEndpoint;
import org.apache.ode.bpel.iapi.BpelServer;
import org.apache.ode.bpel.iapi.Message;
import org.apache.ode.bpel.iapi.MessageExchange;
import org.apache.ode.bpel.iapi.MessageExchange.FailureType;
import org.apache.ode.bpel.iapi.PartnerRoleMessageExchange;
import org.apache.ode.bpel.iapi.ProcessConf;
import org.apache.ode.bpel.iapi.Scheduler;
import org.apache.ode.il.OMUtils;
import org.apache.ode.utils.*;
import org.apache.ode.utils.fs.FileUtils;
import org.apache.ode.utils.uuid.UUID;
import org.apache.ode.utils.wsdl.Messages;
import org.apache.rampart.RampartMessageData;
import org.w3c.dom.Document;
import org.w3c.dom.Element;

import javax.wsdl.Definition;
import javax.wsdl.Fault;
import javax.wsdl.Operation;
import javax.xml.namespace.QName;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;

/**
 * Acts as a service not provided by ODE. Used mainly for invocation as a way to maintain the WSDL description of used
 * services.
 *
 * @author <NAME> <mriou at apache dot org>
 */
public class SoapExternalService implements ExternalService {

    private static final Log __log = LogFactory.getLog(SoapExternalService.class);

    private static final org.apache.ode.utils.wsdl.Messages msgs = Messages.getMessages(Messages.class);


    private static ThreadLocal<ServiceClient> _cachedClients = new ThreadLocal<ServiceClient>();
    private WatchDog<Map, OptionsObserver> _axisOptionsWatchDog;
    private WatchDog<Long, ServiceFileObserver> _axisServiceWatchDog;
    private ConfigurationContext _configContext;


    private ExecutorService _executorService;
    private Definition _definition;
    private QName _serviceName;
    private String _portName;
    protected WSAEndpoint endpointReference;
    private AxisConfiguration _axisConfig;
    private SoapMessageConverter _converter;
    private Scheduler _sched;
    private BpelServer _server;
    private ProcessConf _pconf;

    public SoapExternalService(ProcessConf pconf, QName serviceName, String portName, ExecutorService executorService,
                               AxisConfiguration axisConfig, Scheduler sched, BpelServer server) throws AxisFault {
        _definition = pconf.getDefinitionForService(serviceName);
        _serviceName = serviceName;
        _portName = portName;
        _executorService = executorService;
        _axisConfig = axisConfig;
        _sched = sched;
        _converter = new SoapMessageConverter(_definition, serviceName, portName);
        _server = server;
        _pconf = pconf;

        File fileToWatch = new File(_pconf.getBaseURI().resolve(_serviceName.getLocalPart() + ".axis2"));
        _axisServiceWatchDog = WatchDog.watchFile(fileToWatch, new ServiceFileObserver(fileToWatch));
        _axisOptionsWatchDog = new WatchDog<Map, OptionsObserver>(new EndpointPropertiesMutable(), new OptionsObserver());
        _configContext = new ConfigurationContext(_axisConfig);

        // initial endpoint reference
        Element eprElmt = ODEService.genEPRfromWSDL(_definition, serviceName, portName);
        if (eprElmt == null)
            throw new IllegalArgumentException(msgs.msgPortDefinitionNotFound(serviceName, portName));
        endpointReference = EndpointFactory.convertToWSA(ODEService.createServiceRef(eprElmt));
    }

    public void invoke(final PartnerRoleMessageExchange odeMex) {
        boolean isTwoWay = odeMex.getMessageExchangePattern() == org.apache.ode.bpel.iapi.MessageExchange.MessageExchangePattern.REQUEST_RESPONSE;
        try {

            ServiceClient client = getServiceClient();

            // Override options are passed to the axis MessageContext so we can
            // retrieve them in our session out changeHandler.
            MessageContext mctx = new MessageContext();
            /* make the given options the parent so it becomes the defaults of the MessageContexgt. That allows the user to override
            *  specific options on a given message context and not affect the overall options.
            */
            mctx.getOptions().setParent(client.getOptions());
            writeHeader(mctx, odeMex);

            _converter.createSoapRequest(mctx, odeMex.getRequest(), odeMex.getOperation());

            SOAPEnvelope soapEnv = mctx.getEnvelope();
            EndpointReference axisEPR = new EndpointReference(((MutableEndpoint) odeMex.getEndpointReference())
                    .getUrl());
            if (__log.isDebugEnabled()) {
                __log.debug("Axis2 sending message to " + axisEPR.getAddress() + " using MEX " + odeMex);
                __log.debug("Message: " + soapEnv);
            }

            final OperationClient operationClient = client.createClient(isTwoWay ? ServiceClient.ANON_OUT_IN_OP
                    : ServiceClient.ANON_OUT_ONLY_OP);
            operationClient.addMessageContext(mctx);
            // this Options can be alter without impacting the ServiceClient options (which is a requirement)
            Options operationOptions = operationClient.getOptions();

            // provide HTTP credentials if any
            AuthenticationHelper.setHttpAuthentication(odeMex, operationOptions);

            operationOptions.setAction(mctx.getSoapAction());
            operationOptions.setTo(axisEPR);

            if (isTwoWay) {
                final String mexId = odeMex.getMessageExchangeId();
                final Operation operation = odeMex.getOperation();

                // Defer the invoke until the transaction commits.
                _sched.registerSynchronizer(new Scheduler.Synchronizer() {
                    public void afterCompletion(boolean success) {
                        // If the TX is rolled back, then we don't send the request.
                        if (!success) return;

                        // The invocation must happen in a separate thread, holding on the afterCompletion
                        // blocks other operations that could have been listed there as well.
                        _executorService.submit(new Callable<Object>() {
                            public Object call() throws Exception {
                                try {
                                    operationClient.execute(true);
                                    MessageContext response = operationClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);
                                    MessageContext flt = operationClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_FAULT_VALUE);
                                    if (response != null && __log.isDebugEnabled())
                                        __log.debug("Service response:\n" + response.getEnvelope().toString());

                                    if (flt != null) {
                                        reply(mexId, operation, flt, true);
                                    } else {
                                        reply(mexId, operation, response, response.isFault());
                                    }
                                } catch (Throwable t) {
                                    String errmsg = "Error sending message (mex=" + odeMex + "): " + t.getMessage();
                                    __log.error(errmsg, t);
                                    replyWithFailure(mexId, MessageExchange.FailureType.COMMUNICATION_ERROR, errmsg, null);
                                }
                                return null;
                            }
                        });
                    }

                    public void beforeCompletion() {
                    }
                });
                odeMex.replyAsync();

            } else { /** one-way case * */
                _executorService.submit(new Callable<Object>() {
                    public Object call() throws Exception {
                        operationClient.execute(false);
                        return null;
                    }
                });
                odeMex.replyOneWayOk();
            }
        } catch (AxisFault axisFault) {
            String errmsg = "Error sending message to Axis2 for ODE mex " + odeMex;
            __log.error(errmsg, axisFault);
            odeMex.replyWithFailure(MessageExchange.FailureType.COMMUNICATION_ERROR, errmsg, null);
        }
    }

    private ServiceClient getServiceClient() throws AxisFault {
        try {
            // call manually the check procedure
            // we dont want a dedicated thread for that
            _axisServiceWatchDog.check();
            _axisOptionsWatchDog.check();
        } catch (RuntimeException e) {
            throw AxisFault.makeFault(e.getCause() != null ? e.getCause() : e);
        }

        // apply the options to the service client
        ServiceClient serviceClient = _cachedClients.get();
        if (serviceClient == null) {
            serviceClient = new ServiceClient(_configContext, null);
            _cachedClients.set(serviceClient);
        }
        AxisService anonymousService = _axisServiceWatchDog.getObserver().anonymousService;
        serviceClient.setAxisService(anonymousService);
        serviceClient.setOptions(_axisOptionsWatchDog.getObserver().options);

        applySecuritySettings(serviceClient);

        return serviceClient;
    }
    private void applySecuritySettings(ServiceClient serviceClient) throws AxisFault {
        Options options = serviceClient.getOptions();
        if (options.getProperty(Properties.PROP_SECURITY_POLICY) != null) {
            String policy = (String) options.getProperty(Properties.PROP_SECURITY_POLICY);
            // if the policy path is relative, the full uri is resolved against the process conf directory
            URI policyUri;
            if (FileUtils.isRelative(policy)) {
                policyUri = _pconf.getBaseURI().resolve(policy);
            } else {
                policyUri = new File(policy).toURI();
            }
            try {
                InputStream policyStream = policyUri.toURL().openStream();
                try {
                    Policy policyDoc = PolicyEngine.getPolicy(policyStream);
                    options.setProperty(RampartMessageData.KEY_RAMPART_POLICY, policyDoc);

                    // make sure the proper modules are engaged
                    if (!serviceClient.getAxisService().getAxisConfiguration().isEngaged("rampart")
                            && !serviceClient.getAxisService().isEngaged("rampart")) {
                        serviceClient.engageModule("rampart");
                    }
                } finally {
                    policyStream.close();
                }
            } catch (IOException e) {
                throw new IllegalArgumentException("Exception while parsing policy: " + policyUri, e);
            }
        }
    }

    /**
     * Extracts the action to be used for the given operation.  It first checks to see
     * if a value is specified using WS-Addressing in the portType, it then falls back onto
     * getting it from the SOAP Binding.
     *
     * @param operation the name of the operation to get the Action for
     * @return The action value for the specified operation
     */
    private String getAction(String operation) {
        String action = _converter.getWSAInputAction(operation);
        if (action == null || "".equals(action)) {
            action = _converter.getSoapAction(operation);
        }
        return action;
    }

    /**
     * Extracts endpoint information from ODE message exchange to stuff them into Axis MessageContext.
     */
    private void writeHeader(MessageContext ctxt, PartnerRoleMessageExchange odeMex) {
        Options options = ctxt.getOptions();
        WSAEndpoint targetWSAEPR = EndpointFactory.convertToWSA((MutableEndpoint) odeMex.getEndpointReference());
        WSAEndpoint myRoleWSAEPR = EndpointFactory.convertToWSA((MutableEndpoint) odeMex.getMyRoleEndpointReference());
        WSAEndpoint targetEPR = new WSAEndpoint(targetWSAEPR);
        
        EndpointReference replyEPR = null; 

        String partnerSessionId = odeMex.getProperty(MessageExchange.PROPERTY_SEP_PARTNERROLE_SESSIONID);
        String myRoleSessionId = odeMex.getProperty(MessageExchange.PROPERTY_SEP_MYROLE_SESSIONID);

        if (partnerSessionId != null) {
            if (__log.isDebugEnabled()) {
                __log.debug("Partner session identifier found for WSA endpoint: " + partnerSessionId);
            }
            targetEPR.setSessionId(partnerSessionId);
        }
        options.setProperty("targetSessionEndpoint", targetEPR);

        if (myRoleWSAEPR != null) {
            WSAEndpoint myRoleEPR = new WSAEndpoint(myRoleWSAEPR);
            if (myRoleSessionId != null) {
                if (__log.isDebugEnabled()) {
                    __log.debug("MyRole session identifier found for myrole (callback) WSA endpoint: "
                            + myRoleSessionId);
                }
                myRoleEPR.setSessionId(myRoleSessionId);
            }
            options.setProperty("callbackSessionEndpoint", myRoleEPR);

            // Map My Session ID to JMS Correlation ID
            Document callbackEprXml = odeMex.getMyRoleEndpointReference().toXML();
            Element serviceElement = callbackEprXml.getDocumentElement();
            
        	if (myRoleSessionId != null) {
	            options.setProperty(JMSConstants.JMS_COORELATION_ID, myRoleSessionId);
        	} else {
                if (myRoleWSAEPR.getSessionId() != null) {
                    options.setProperty(JMSConstants.JMS_COORELATION_ID, myRoleSessionId);
                }
        	}

            Element address = DOMUtils.findChildByName(serviceElement, 
            		new QName(Namespaces.WS_ADDRESSING_NS, "Address"), true);
            if (__log.isDebugEnabled()) {
                __log.debug("The system-defined wsa address is : "
                        + address);
            }
            if (address != null) {
	            String url = address.getTextContent();
	            String jmsDestination = (String) options.getProperty(JMSConstants.REPLY_PARAM);
                if (__log.isDebugEnabled()) {
                    __log.debug("The user-defined JMS replyTo destination is: "
                            + jmsDestination);
                    __log.debug("The user-defined JMS wait timeout is: "
                            + options.getProperty(JMSConstants.JMS_WAIT_REPLY));
                }
	            if (jmsDestination == null || "".equals(jmsDestination.trim())) {
	            	// If the REPLY_PARAM property is not user-defined, then use the default value from myRole EPR	            	
		            int startIndex = url.indexOf("jms:/");
		            if (startIndex != -1) {
		            	startIndex += "jms:/".length();
		            	if (url.charAt(startIndex + 1) == '/') {
		            		// startIndex++; // treat "/" as valid start character for queue name
		            	}
		            	if (url.startsWith("dynamic")) {
		            		startIndex += "dynamicQueues".length();
		            	}
		            	int jmsEndIndex = url.indexOf("?", startIndex);
		            	if (jmsEndIndex == -1) {
		            		jmsEndIndex = url.length();
		            	}
		        		jmsDestination = url.substring(startIndex, jmsEndIndex);
		                options.setProperty(JMSConstants.REPLY_PARAM, jmsDestination);
			            replyEPR = new EndpointReference(url);
		            } else {
			            startIndex = url.indexOf("http://");
			            if (startIndex != -1) {
			            	startIndex = url.indexOf("/processes/");
			            	if (startIndex != -1) {
			            		startIndex += "/processes/".length();
			            		jmsDestination = url.substring(startIndex);
				                options.setProperty(JMSConstants.REPLY_PARAM, jmsDestination);
			            	}
			            }
		            }
	            } else {
	            	replyEPR = new EndpointReference("jms:/" + jmsDestination);
	            }
            }
        } else {
            __log.debug("My-Role EPR not specified, SEP will not be used.");
        }

        String action = getAction(odeMex.getOperationName());
        ctxt.setSoapAction(action);

        if (replyEPR == null) {
        	if (MessageExchange.MessageExchangePattern.REQUEST_RESPONSE == odeMex.getMessageExchangePattern()) {
	        	replyEPR = new EndpointReference(Namespaces.WS_ADDRESSING_ANON_URI);
        	}
        }
        if (replyEPR != null) {
{'fraction_non_alphanumeric': 0.08488886491152352, 'fraction_numerical': 0.0015645230902028486, 'mean_word_length': 2.197602208038641, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1115405', 'n_tokens_mistral': 9749, 'n_tokens_neox': 9249, 'n_words': 2246}