soapexternalservice.java

bpel执行引擎用来执行bpel业务流程

/*
 * 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 java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.Map;
import java.util.HashMap;
import java.io.File;
import java.io.InputStream;

import javax.wsdl.Definition;
import javax.wsdl.Fault;
import javax.wsdl.Operation;
import javax.xml.namespace.QName;

import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axis2.AxisFault;
import org.apache.axis2.deployment.ServiceBuilder;
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.engine.AxisConfiguration;
import org.apache.axis2.wsdl.WSDLConstants;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
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.epr.WSDL11Endpoint;
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.PartnerRoleMessageExchange;
import org.apache.ode.bpel.iapi.Scheduler;
import org.apache.ode.bpel.iapi.ProcessConf;
import org.apache.ode.bpel.iapi.MessageExchange.FailureType;
import org.apache.ode.il.OMUtils;
import org.apache.ode.utils.DOMUtils;
import org.apache.ode.utils.Namespaces;
import org.apache.ode.utils.WatchDog;
import org.apache.ode.utils.CollectionUtils;
import org.apache.ode.utils.fs.FileWatchDog;
import org.apache.ode.utils.wsdl.Messages;
import org.apache.ode.utils.uuid.UUID;
import org.w3c.dom.Document;
import org.w3c.dom.Element;

/**
 * Acts as a service not provided by ODE. Used mainly for invocation as a way to maintain the WSDL description of used
 * services.
 *
 * @author Matthieu Riou <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 final int EXPIRE_SERVICE_CLIENT = 30000;

    private static ThreadLocal<CachedOptions> _cachedOptions = new ThreadLocal<CachedOptions>();
    private static ThreadLocal<CachedServiceClient> _cachedClients = new ThreadLocal<CachedServiceClient>();

    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;

        // 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 {
            // Override options are passed to the axis MessageContext so we can
            // retrieve them in our session out handler.
            MessageContext mctx = new MessageContext();
            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);
            }


            ServiceClient client = getCachedServiceClient().client;
            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();
            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 CachedServiceClient getCachedServiceClient() throws AxisFault {
        CachedServiceClient cachedServiceClient = _cachedClients.get();
        if (cachedServiceClient == null) {
            cachedServiceClient = new CachedServiceClient(new File(_pconf.getBaseURI().resolve(_serviceName.getLocalPart() + ".axis2")), EXPIRE_SERVICE_CLIENT);
            _cachedClients.set(cachedServiceClient);
        }
        try {
            // call manually the check procedure
            // we dont want a dedicated thread for that
            cachedServiceClient.check();
        } catch (RuntimeException e) {
            throw AxisFault.makeFault(e.getCause() != null ? e.getCause() : e);
        }

        SoapExternalService.CachedOptions cachedOptions = _cachedOptions.get();
        if (cachedOptions == null) {
            cachedOptions = new CachedOptions();
            _cachedOptions.set(cachedOptions);
        }
        cachedOptions.check();

        // apply the options to the service client
        cachedServiceClient.client.setOptions(cachedOptions.options);
        return cachedServiceClient;
    }

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