odelifecycle.java

来自「bpel执行引擎用来执行bpel业务流程」· Java 代码 · 共 462 行 · 第 1/2 页

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

import java.io.File;
import java.io.FileNotFoundException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;

import javax.jbi.JBIException;
import javax.jbi.component.ComponentContext;
import javax.jbi.component.ComponentLifeCycle;
import javax.jbi.component.ServiceUnitManager;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.transaction.TransactionManager;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.ode.bpel.connector.BpelServerConnector;
import org.apache.ode.bpel.dao.BpelDAOConnectionFactoryJDBC;
import org.apache.ode.bpel.engine.BpelServerImpl;
import org.apache.ode.bpel.engine.ProcessAndInstanceManagementMBean;
import org.apache.ode.bpel.engine.ProcessAndInstanceManagementImpl;

import org.apache.ode.bpel.iapi.BpelEventListener;
import org.apache.ode.bpel.intercept.MessageExchangeInterceptor;
import org.apache.ode.il.dbutil.Database;
import org.apache.ode.il.dbutil.DatabaseConfigException;
import org.apache.ode.jbi.msgmap.Mapper;
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.fs.TempFileManager;

/**
 * This class implements ComponentLifeCycle. The JBI framework will start this engine class automatically when JBI framework starts
 * up.
 */
public class OdeLifeCycle implements ComponentLifeCycle {
    private static final Messages __msgs = Messages.getMessages(Messages.class);

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

    private OdeSUManager _suManager = null;

    private boolean _initSuccess = false;

    private OdeContext _ode;

    private Receiver _receiver;

    private boolean _started;

    private BpelServerConnector _connector;

    private Database _db;

    private ObjectName _mbeanName;

    ServiceUnitManager getSUManager() {
        return _suManager;
    }

    OdeContext getOdeContext() {
        return _ode;
    }

    public ObjectName getExtensionMBeanName() {
        return _mbeanName;
    }

    public void init(ComponentContext context) throws JBIException {
        try {
            _ode = OdeContext.getInstance();
            _ode.setContext(context);

            // Use system property to determine if DeliveryChannel.sendSync or DeliveryChannel.send is used.
            if (Boolean.getBoolean("org.apache.ode.jbi.sendSynch"))
                _ode._consumer = new OdeConsumerSync(_ode);
            else
                _ode._consumer = new OdeConsumerAsync(_ode);

            if (_ode.getContext().getWorkspaceRoot() != null)
                TempFileManager.setWorkingDirectory(new File(_ode.getContext().getWorkspaceRoot()));

            __log.debug("Loading properties.");
            initProperties();

            __log.debug("Initializing message mappers.");
            initMappers();

            __log.debug("Creating data source.");
            initDataSource();

            __log.debug("Starting Dao.");
            initDao();

            __log.info("Hibernate started.");

            __log.debug("Starting BPEL server.");
            initBpelServer();

            // Register BPEL event listeners configured in ode-jbi.properties.
            registerEventListeners();

            registerMexInterceptors();

            registerMBean();

            __log.debug("Starting JCA connector.");
            initConnector();

            __log.debug("Register ProcessManagement APIs");
            _ode.activatePMAPIs();

            _suManager = new OdeSUManager(_ode);
            _initSuccess = true;
            __log.info(__msgs.msgOdeInitialized());
        } finally {
            if (!_initSuccess) {
                // TODO ..then what? at least shutdown the scheduler
            }
        }
    }

    @SuppressWarnings("unchecked")
    private void initMappers() throws JBIException {
        Class<Mapper> mapperClass;
        try {
            mapperClass = (Class<Mapper>) Class.forName(_ode._config.getMessageMapper());
        } catch (ClassNotFoundException e) {
            String errmsg = __msgs.msgOdeInitMapperClassNotFound(_ode._config.getMessageMapper());
            __log.error(errmsg);
            throw new JBIException(errmsg, e);
        } catch (Throwable t) {
            String errmsg = __msgs.msgOdeInitMapperClassLoadFailed(_ode._config.getMessageMapper());
            __log.error(errmsg);
            throw new JBIException(errmsg, t);
        }
        try {
            _ode.registerMapper((Mapper) mapperClass.newInstance());
        } catch (Throwable t) {
            String errmsg = __msgs.msgOdeInitMapperInstantiationFailed(_ode._config.getMessageMapper());
            __log.error(errmsg);
            throw new JBIException(errmsg, t);
        }
    }

    private void initDataSource() throws JBIException {
        _db = new Database(_ode._config);
        _db.setTransactionManager(_ode.getTransactionManager());
        _db.setWorkRoot(new File(_ode.getContext().getInstallRoot()));

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

        _ode._dataSource = _db.getDataSource();
    }

    /**
     * Load the "ode-jbi.properties" file from the install directory.
     *
     * @throws JBIException
     */
    private void initProperties() throws JBIException {
        OdeConfigProperties config = new OdeConfigProperties(new File(_ode.getContext().getInstallRoot(),
                OdeConfigProperties.CONFIG_FILE_NAME));

        try {
            config.load();
        } catch (FileNotFoundException fnf) {
            __log.warn(__msgs.msgOdeInstallErrorCfgNotFound(config.getFile()));
        } catch (Exception ex) {
           String errmsg = __msgs.msgOdeInstallErrorCfgReadError(config.getFile());
           throw new JBIException(errmsg,ex);
        }
        _ode._config = config;
    }

    private void initBpelServer() throws JBIException {
        if (__log.isDebugEnabled()) {
            __log.debug("ODE initializing");
        }

        _ode._server = new BpelServerImpl();
        // We don't want the server to automatically load deployed processes,
        // we'll do that explcitly
        _ode._eprContext = new EndpointReferenceContextImpl(_ode);
        _ode._mexContext = new MessageExchangeContextImpl(_ode);
        if (_ode._config.getThreadPoolMaxSize() == 0)
            _ode._executorService = Executors.newCachedThreadPool();
        else
            _ode._executorService = Executors.newFixedThreadPool(_ode._config.getThreadPoolMaxSize());
        _ode._scheduler = new SimpleScheduler(new GUID().toString(),new JdbcDelegate(_ode._dataSource));
        _ode._scheduler.setJobProcessor(_ode._server);
        _ode._scheduler.setExecutorService(_ode._executorService);
        _ode._scheduler.setTransactionManager((TransactionManager) _ode.getContext().getTransactionManager());

        _ode._store = new ProcessStoreImpl(_ode._eprContext , _ode._dataSource, _ode._config.getDAOConnectionFactory(), _ode._config, false);
        _ode._store.loadAll();

        _ode._server.setInMemDaoConnectionFactory(new org.apache.ode.bpel.memdao.BpelDAOConnectionFactoryImpl(
                _ode._scheduler, _ode._config.getInMemMexTtl()));
        _ode._server.setDaoConnectionFactory(_ode._daocf);
        _ode._server.setEndpointReferenceContext(_ode._eprContext);
        _ode._server.setMessageExchangeContext(_ode._mexContext);