odeserver.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 org.apache.axis2.AxisFault;
import org.apache.axis2.description.AxisOperation;
import org.apache.axis2.description.AxisService;
import org.apache.axis2.engine.AxisConfiguration;
import org.apache.commons.collections.map.MultiKeyMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.ode.axis2.deploy.DeploymentPoller;
import org.apache.ode.axis2.hooks.ODEAxisService;
import org.apache.ode.axis2.hooks.ODEMessageReceiver;
import org.apache.ode.axis2.service.DeploymentWebService;
import org.apache.ode.axis2.service.ManagementService;
import org.apache.ode.axis2.httpbinding.HttpExternalService;
import org.apache.ode.bpel.connector.BpelServerConnector;
import org.apache.ode.bpel.dao.BpelDAOConnectionFactory;
import org.apache.ode.bpel.engine.BpelServerImpl;
import org.apache.ode.bpel.engine.CountLRUDehydrationPolicy;
import org.apache.ode.bpel.extvar.jdbc.JdbcExternalVariableModule;
import org.apache.ode.bpel.iapi.BpelEventListener;
import org.apache.ode.bpel.iapi.ContextException;
import org.apache.ode.bpel.iapi.ProcessConf;
import org.apache.ode.bpel.iapi.ProcessStoreEvent;
import org.apache.ode.bpel.iapi.ProcessStoreListener;
import org.apache.ode.bpel.iapi.Scheduler;
import org.apache.ode.bpel.iapi.EndpointReferenceContext;
import org.apache.ode.bpel.intercept.MessageExchangeInterceptor;
import org.apache.ode.bpel.memdao.BpelDAOConnectionFactoryImpl;
import org.apache.ode.bpel.pmapi.InstanceManagement;
import org.apache.ode.bpel.pmapi.ProcessManagement;
import org.apache.ode.il.dbutil.Database;
import org.apache.ode.scheduler.simple.JdbcDelegate;
import org.apache.ode.scheduler.simple.SimpleScheduler;
import org.apache.ode.store.ProcessStoreImpl;
import org.apache.ode.utils.GUID;
import org.apache.ode.utils.wsdl.WsdlUtils;
import org.apache.ode.utils.fs.TempFileManager;

import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.sql.DataSource;
import javax.transaction.*;
import javax.transaction.xa.XAResource;
import javax.wsdl.Definition;
import javax.xml.namespace.QName;
import java.io.File;
import java.io.FileNotFoundException;
import java.util.StringTokenizer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

/**
 * Server class called by our Axis hooks to handle all ODE lifecycle management.
 *
 * @author Matthieu Riou <mriou at apache dot org>
 */
public class ODEServer {

    protected final Log __log = LogFactory.getLog(getClass());
    protected final Log __logTx = LogFactory.getLog("org.apache.ode.tx");

    private static final Messages __msgs = Messages.getMessages(Messages.class);

    protected File _appRoot;

    protected File _workRoot;

    protected BpelServerImpl _server;

    protected ProcessStoreImpl _store;

    protected ODEConfigProperties _odeConfig;

    protected AxisConfiguration _axisConfig;

    protected TransactionManager _txMgr;

    protected BpelDAOConnectionFactory _daoCF;

    protected ExecutorService _executorService;

    protected Scheduler _scheduler;

    protected Database _db;

    private DeploymentPoller _poller;

    private MultiKeyMap _services = new MultiKeyMap();

    private MultiKeyMap _externalServices = new MultiKeyMap();

    private BpelServerConnector _connector;

    private ManagementService _mgtService;


    public void init(ServletConfig config, AxisConfiguration axisConf) throws ServletException {
        init(config.getServletContext().getRealPath("/WEB-INF"), axisConf);
    }

    public void init(String contextPath, AxisConfiguration axisConf) throws ServletException {
        boolean success = false;
        try {
            _axisConfig = axisConf;
            String rootDir = System.getProperty("org.apache.ode.rootDir");
            if (rootDir != null) _appRoot = new File(rootDir);
            else _appRoot = new File(contextPath);
            TempFileManager.setWorkingDirectory(_appRoot);

            __log.debug("Loading properties");
            String confDir = System.getProperty("org.apache.ode.configDir");
            if (confDir == null) _odeConfig = new ODEConfigProperties(new File(_appRoot, "conf"));
            else _odeConfig = new ODEConfigProperties(new File(confDir));

            try {
                _odeConfig.load();
            } catch (FileNotFoundException fnf) {
                String errmsg = __msgs.msgOdeInstallErrorCfgNotFound(_odeConfig.getFile());
                __log.warn(errmsg);
            } catch (Exception ex) {
                String errmsg = __msgs.msgOdeInstallErrorCfgReadError(_odeConfig.getFile());
                __log.error(errmsg, ex);
                throw new ServletException(errmsg, ex);
            }

            String wdir = _odeConfig.getWorkingDir();
            if (wdir == null) _workRoot = _appRoot;
            else _workRoot = new File(wdir.trim());

            __log.debug("Initializing transaction manager");
            initTxMgr();
            __log.debug("Creating data source.");
            initDataSource();
            __log.debug("Starting DAO.");
            initDAO();
            EndpointReferenceContextImpl eprContext = new EndpointReferenceContextImpl(this);            
            __log.debug("Initializing BPEL process store.");
            initProcessStore(eprContext);
            __log.debug("Initializing BPEL server.");
            initBpelServer(eprContext);

            // Register BPEL event listeners configured in axis2.properties file.
            registerEventListeners();
            registerMexInterceptors();

            registerExternalVariableModules();

            try {
                _server.start();
            } catch (Exception ex) {
                String errmsg = __msgs.msgOdeBpelServerStartFailure();
                __log.error(errmsg, ex);
                throw new ServletException(errmsg, ex);
            }

            File deploymentDir = new File(_workRoot, "processes");
            _poller = new DeploymentPoller(deploymentDir, this);

            _mgtService = new ManagementService();
            _mgtService.enableService(_axisConfig, _server, _store, _appRoot.getAbsolutePath());

            new DeploymentWebService().enableService(_axisConfig, _server, _store, _poller, _appRoot.getAbsolutePath(), _workRoot
                    .getAbsolutePath());

            _store.loadAll();

            __log.debug("Initializing JCA adapter.");
            initConnector();

            _poller.start();
            __log.info(__msgs.msgPollingStarted(deploymentDir.getAbsolutePath()));
            __log.info(__msgs.msgOdeStarted());
            success = true;
        } finally {
            if (!success)
                try {
                    // shutDown();
                } catch (Exception ex) {
                    // Problem rolling back start(). Not so important
                    __log.debug("Error rolling back incomplete shutdown.", ex);
                }
        }

    }

    private void initDataSource() throws ServletException {
        _db = new Database(_odeConfig);
        _db.setTransactionManager(_txMgr);
        _db.setWorkRoot(_workRoot);

        try {
            _db.start();
        } catch (Exception ex) {
            String errmsg = __msgs.msgOdeDbConfigError();
            __log.error(errmsg, ex);
            throw new ServletException(errmsg, ex);
        }

    }

    /**
     * Shutdown the service engine. This performs cleanup before the BPE is terminated. Once this method has been called, init()
     * must be called before the transformation engine can be started again with a call to start().
     *
     * @throws AxisFault if the engine is unable to shut down.
     */
    public void shutDown() throws AxisFault {

        ClassLoader old = Thread.currentThread().getContextClassLoader();
        Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
        try {
            if (_poller != null)
                try {
                    __log.debug("shutting down poller");
                    _poller.stop();
                    _poller = null;
                } catch (Throwable t) {
                    __log.debug("Error stopping poller.", t);
                }

            if (_server != null)
                try {
                    __log.debug("shutting down ODE server.");
                    _server.shutdown();
                    _server = null;
                } catch (Throwable ex) {
                    __log.debug("Error stopping services.", ex);
                }

            if (_scheduler != null)
                try {
                    __log.debug("shutting down quartz scheduler.");
                    _scheduler.shutdown();
                    _scheduler = null;
                } catch (Exception ex) {
                    __log.debug("Scheduler couldn't be shutdown.", ex);
                }

            if (_store != null)
                try {
                    _store.shutdown();
                    _store = null;
                } catch (Throwable t) {
                    __log.debug("Store could not be shutdown.", t);
                }

            if (_daoCF != null)
                try {
                    _daoCF.shutdown();
                } catch (Throwable ex) {
                    __log.debug("DOA shutdown failed.", ex);
                } finally {
                    _daoCF = null;
                }

            if (_db != null)
                try {
                    _db.shutdown();

                } catch (Throwable ex) {
                    __log.debug("DB shutdown failed.", ex);
                } finally {
                    _db = null;
                }

            if (_txMgr != null) {
                __log.debug("shutting down transaction manager.");
                _txMgr = null;
            }

            if (_connector != null) {
                try {
                    __log.debug("shutdown BpelConnector");
                    _connector.shutdown();
                } catch (Throwable t) {
                    __log.error("Unable to cleanup temp files.", t);
                }
            }
            
            try {
                __log.debug("cleaning up temporary files.");
                TempFileManager.cleanup();
            } catch (Throwable t) {
                __log.error("Unable to cleanup temp files.", t);
            }


            __log.info(__msgs.msgOdeShutdownCompleted());
        } finally {
            Thread.currentThread().setContextClassLoader(old);
        }
    }

    public ODEService createService(ProcessConf pconf, QName serviceName, String portName) throws AxisFault {
        destroyService(serviceName, portName);
        AxisService axisService = ODEAxisService.createService(_axisConfig, pconf, serviceName, portName);
        ODEService odeService = new ODEService(axisService, pconf, serviceName, portName, _server, _txMgr);

        _services.put(serviceName, portName, odeService);

        // Setting our new service on the receiver, the same receiver handles
        // all
        // operations so the first one should fit them all
        AxisOperation firstOp = (AxisOperation) axisService.getOperations().next();
        ((ODEMessageReceiver) firstOp.getMessageReceiver()).setService(odeService);

        // We're public!
        _axisConfig.addService(axisService);
        __log.debug("Created Axis2 service " + serviceName);
        return odeService;
    }

    public ExternalService createExternalService(ProcessConf pconf, QName serviceName, String portName) throws ContextException {
        ExternalService extService = (ExternalService) _externalServices.get(serviceName);
        if (extService != null)
            return extService;

        Definition def = pconf.getDefinitionForService(serviceName);
        try {
            if (WsdlUtils.useHTTPBinding(def, serviceName, portName)) {
                if(__log.isDebugEnabled())__log.debug("Creating HTTP-bound external service " + serviceName);
                extService = new HttpExternalService(pconf, serviceName, portName, _executorService, _scheduler, _server);
            } else if (WsdlUtils.useSOAPBinding(def, serviceName, portName)) {
                if(__log.isDebugEnabled())__log.debug("Creating SOAP-bound external service " + serviceName);