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

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.ode.bpel.compiler.api.CompilationException;
import org.apache.ode.bpel.dd.DeployDocument;
import org.apache.ode.bpel.dd.TDeployment;
import org.apache.ode.bpel.iapi.*;
import org.apache.ode.il.config.OdeConfigProperties;
import org.apache.ode.store.DeploymentUnitDir.CBPInfo;
import org.apache.ode.utils.DOMUtils;
import org.apache.ode.utils.GUID;
import org.apache.ode.utils.msg.MessageBundle;
import org.h2.jdbcx.JdbcDataSource;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;

import javax.sql.DataSource;
import javax.xml.namespace.QName;
import java.io.File;
import java.io.IOException;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * <p>
 * JDBC-based implementation of a process store. Also provides an "in-memory" store by way of H2 database.
 * </p>
 *
 * <p>
 * The philsophy here is to keep things simple. Process store operations are relatively infrequent. Performance of the public
 * methods is not a concern. However, note that the {@link org.apache.ode.bpel.iapi.ProcessConf} objects returned by the class are
 * going to be used from within the engine runtime, and hence their performance needs to be very good. Similarly, these objects
 * should be immutable so as not to confuse the engine.
 *
 * Note the way that the database is used in this class, it is more akin to a recovery log, this is intentional: we want to start
 * up, load stuff from the database and then pretty much forget about it when it comes to reads.
 *
 * @author Maciej Szefler <mszefler at gmail dot com>
 * @author mriou <mriou at apache dot org>
 */
public class ProcessStoreImpl implements ProcessStore {

    private static final Logger __log = LoggerFactory.getLogger(ProcessStoreImpl.class);

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

    private final CopyOnWriteArrayList<ProcessStoreListener> _listeners = new CopyOnWriteArrayList<ProcessStoreListener>();

    protected Map<QName, ProcessConfImpl> _processes = new HashMap<QName, ProcessConfImpl>();

    protected Map<String, DeploymentUnitDir> _deploymentUnits = new HashMap<String, DeploymentUnitDir>();

    /** Guards access to the _processes and _deploymentUnits */
    private final ReadWriteLock _rw = new ReentrantReadWriteLock();

    private ConfStoreConnectionFactory _cf;

    private EndpointReferenceContext eprContext;

    private boolean generateProcessEventsAll;

    protected File _deployDir;

    protected File _configDir;


    /**
     * Executor used to process DB transactions. Allows us to isolate the TX context, and to ensure that only one TX gets executed a
     * time. We don't really care to parallelize these operations because: i) HSQL does not isolate transactions and we don't want
     * to get confused ii) we're already serializing all the operations with a read/write lock. iii) we don't care about
     * performance, these are infrequent operations.
     */
    private ExecutorService _executor = Executors.newSingleThreadExecutor(new SimpleThreadFactory());

    /**
     * In-memory DataSource, or <code>null</code> if we are using a real DS. We need this to shutdown the DB.
     */
    private DataSource _inMemDs;

    private static final ThreadLocal<Long> _currentVersion = new ThreadLocal<Long>();

    public ProcessStoreImpl() {
        this(null, null, "", new OdeConfigProperties(new Properties(), ""), true);
    }

    public ProcessStoreImpl(EndpointReferenceContext eprContext, DataSource ds, String persistenceType, OdeConfigProperties props, boolean createDatamodel) {
        this.eprContext = eprContext;
        this.generateProcessEventsAll = props.getProperty("generateProcessEvents", "all").equals("all");
        if (ds != null) {
            // ugly hack
            if (persistenceType.toLowerCase().indexOf("hib") != -1) {
                _cf = new org.apache.ode.store.hib.DbConfStoreConnectionFactory(ds, props.getProperties(), createDatamodel, props.getTxFactoryClass());
            } else {
                _cf = new org.apache.ode.store.jpa.DbConfStoreConnectionFactory(ds, props.getProperties(), createDatamodel, props.getTxFactoryClass());
            }
         } else {
            // If the datasource is not provided, then we create a H2-based
            // in-memory database. Makes testing a bit simpler.
            DataSource h2 = createInternalDS(new GUID().toString());
            if ("hibernate".equalsIgnoreCase(persistenceType)) {
                _cf = new org.apache.ode.store.hib.DbConfStoreConnectionFactory(h2, props.getProperties(), createDatamodel, props.getTxFactoryClass());
            } else {
                _cf = new org.apache.ode.store.jpa.DbConfStoreConnectionFactory(h2, props.getProperties(), createDatamodel, props.getTxFactoryClass());
            }
            _inMemDs = h2;
        }


    }


    /**
     * Constructor that hardwires OpenJPA on a new in-memory database. Suitable for tests.
     */
    public ProcessStoreImpl(EndpointReferenceContext eprContext, DataSource inMemDs) {
        this.eprContext = eprContext;
        DataSource h2 = createInternalDS(new GUID().toString());
        //when in memory we always create the model as we are starting from scratch
        _cf = new org.apache.ode.store.jpa.DbConfStoreConnectionFactory(h2, true, OdeConfigProperties.DEFAULT_TX_FACTORY_CLASS_NAME);
        _inMemDs = h2;
    }

    public void shutdown() {
        if (_inMemDs != null) {
            shutdownInternalDB(_inMemDs);
            _inMemDs = null;
        }
        if (_executor != null) {
            _executor.shutdownNow();
            _executor = null;
        }
    }

    @Override
    protected void finalize() throws Throwable {
        // force a shutdown so that HSQL cleans up its mess.
        try {
            shutdown();
        } catch (Throwable t) {
            ; // we tried, no worries.
        }
        super.finalize();
    }

    /**
     * Deploys a process.
     */
    public Collection<QName> deploy(final File deploymentUnitDirectory, boolean autoincrementVersion) {
        return deploy(deploymentUnitDirectory, true, null, autoincrementVersion);
    }

    public Collection<QName> deploy(final File deploymentUnitDirectory) {
        return deploy(deploymentUnitDirectory, true, null, OdeGlobalConfig.autoincrementVersion());
    }

    /**
     * Deploys a process.
     */
    public Collection<QName> deploy(final File deploymentUnitDirectory, boolean activate, String duName, boolean autoincrementVersion) {
        __log.info(__msgs.msgDeployStarting(deploymentUnitDirectory));

        final Date deployDate = new Date();

        // Create the DU and compile/scan it before acquiring lock.
        final DeploymentUnitDir du = new DeploymentUnitDir(deploymentUnitDirectory);
        if( duName != null ) {
            // Override the package name if given from the parameter
            du.setName(duName);
        }

        long version;
        if (autoincrementVersion || du.getStaticVersion() == -1) {
            // Process and DU use a monotonically increased single version number by default.
            try {
                version = getCurrentVersion();
            } finally {
                //we need to reset the current version thread local value.
                _currentVersion.set(null);
            }
        } else {
            version = du.getStaticVersion();
        }
        du.setVersion(version);

        try {
            du.compile();
        } catch (CompilationException ce) {
            String errmsg = __msgs.msgDeployFailCompileErrors(ce);
            __log.error(errmsg, ce);
            throw new ContextException(errmsg, ce);
        }

        du.scan();
        final DeployDocument dd = du.getDeploymentDescriptor();
        final ArrayList<ProcessConfImpl> processes = new ArrayList<ProcessConfImpl>();
        Collection<QName> deployed;

        _rw.writeLock().lock();

        try {
            if (_deploymentUnits.containsKey(du.getName())) {
                String errmsg = __msgs.msgDeployFailDuplicateDU(du.getName());
                __log.error(errmsg);
                throw new ContextException(errmsg);
            }

            retirePreviousPackageVersions(du);

            for (TDeployment.Process processDD : dd.getDeploy().getProcessArray()) {
                QName pid = toPid(processDD.getName(), version);

                if (_processes.containsKey(pid)) {
                    String errmsg = __msgs.msgDeployFailDuplicatePID(processDD.getName(), du.getName());
                    __log.error(errmsg);
                    throw new ContextException(errmsg);
                }

                QName type = processDD.getType() != null ? processDD.getType() : processDD.getName();

                CBPInfo cbpInfo = du.getCBPInfo(type);
                if (cbpInfo == null) {
                    String errmsg = __msgs.msgDeployFailedProcessNotFound(processDD.getName(), du.getName());
                    __log.error(errmsg);
                    throw new ContextException(errmsg);
                }

                ProcessConfImpl pconf = new ProcessConfImpl(pid, processDD.getName(), version, du, processDD, deployDate,
                        calcInitialProperties(du.getProperties(), processDD), calcInitialState(processDD), eprContext, _configDir, generateProcessEventsAll);
                processes.add(pconf);
            }

            _deploymentUnits.put(du.getName(), du);

            for (ProcessConfImpl process : processes) {
                __log.info(__msgs.msgProcessDeployed(du.getDeployDir(), process.getProcessId()));
                _processes.put(process.getProcessId(), process);

            }

        } finally {
            _rw.writeLock().unlock();
        }

        // Do the deployment in the DB. We need this so that we remember deployments across system shutdowns.
        // We don't fail if there is a DB error, simply print some errors.
        deployed = exec(new Callable<Collection<QName>>() {
            public Collection<QName> call(ConfStoreConnection conn) {
                // Check that this deployment unit is not deployed.
                DeploymentUnitDAO dudao = conn.getDeploymentUnit(du.getName());
                if (dudao != null) {
                    String errmsg = "Database out of synch for DU " + du.getName();
                    __log.warn(errmsg);
                    dudao.delete();
                }

                dudao = conn.createDeploymentUnit(du.getName());
                try {
                    dudao.setDeploymentUnitDir(deploymentUnitDirectory.getCanonicalPath());
                } catch (IOException e1) {
                    String errmsg = "Error getting canonical path for " + du.getName()
                            + "; deployment unit will not be available after restart!";
                    __log.error(errmsg);

                }

                ArrayList<QName> deployed = new ArrayList<QName>();
                // Going trough each process declared in the dd
                for (ProcessConfImpl pc : processes) {
                    try {
                        ProcessConfDAO newDao = dudao.createProcess(pc.getProcessId(), pc.getType(), pc.getVersion());
                        newDao.setState(pc.getState());
                        for (Map.Entry<QName, Node> prop : pc.getProcessProperties().entrySet()) {
                            newDao.setProperty(prop.getKey(), DOMUtils.domToString(prop.getValue()));
                        }
                        deployed.add(pc.getProcessId());
                    } catch (Throwable e) {
                        String errmsg = "Error persisting deployment record for " + pc.getProcessId()
                                + "; process will not be available after restart!";
                        __log.error(errmsg, e);
                    }
                }
                return deployed;
            }

        });


        _rw.readLock().lock();
        boolean readLockHeld = true;
        try {
            for (ProcessConfImpl process : processes) {
                fireEvent(new ProcessStoreEvent(ProcessStoreEvent.Type.DEPLOYED, process.getProcessId(), process.getDeploymentUnit()
                        .getName()));
                fireStateChange(process.getProcessId(), process.getState(), process.getDeploymentUnit().getName());
            }
        } catch (Exception e) {
            //need to unlock as undeploy operation will need a writeLock
            _rw.readLock().unlock();
            readLockHeld = false;

            // A problem at that point means that engine deployment failed, we don't want the store to keep the du
            __log.warn("Deployment failed within the engine, store undeploying process.", e);
            undeploy(deploymentUnitDirectory);
            if (e instanceof ContextException) throw (ContextException) e;
            else throw new ContextException("Deployment failed within the engine. " + e.getMessage(), e);
        } finally {
            if(readLockHeld)
                _rw.readLock().unlock();
        }

        return deployed;
    }

    /**
     * Retire all the other versions of the same DU:
     * first take the DU name and insert version regexp,
     * than try to match the this string against names of already deployed DUs.
     * For instance if we are deploying DU "AbsenceRequest-2/AbsenceRequest.ode" and
     * there's already version 2 than regexp
     * "AbsenceRequest([-\\.](\d)+)?/AbsenceRequest.ode" will be matched against
     * "AbsenceRequest-2/AbsenceRequest.ode" and setRetirePackage() will be called accordingly.
     */
    private void retirePreviousPackageVersions(DeploymentUnitDir du) {

        Pattern duNamePattern = getPreviousPackageVersionPattern(du.getName());

        for (String deployedDUname : _deploymentUnits.keySet()) {
            Matcher matcher = duNamePattern.matcher(deployedDUname);
            if (matcher.matches()) {
                setRetiredPackage(deployedDUname, true);
            }
        }
    }

    public Collection<QName> undeploy(final File dir) {
        return undeploy(dir.getName());
    }

    public Collection<QName> undeploy(final String duName) {
        try {
            exec(new Callable<Collection<QName>>() {
                public Collection<QName> call(ConfStoreConnection conn) {
                    DeploymentUnitDAO dudao = conn.getDeploymentUnit(duName);
                    if (dudao != null)
                        dudao.delete();
                    return null;
                }
            });
        } catch (Exception ex) {
            __log.error("Error synchronizing with data store; " + duName + " may be reappear after restart!");
        }

        return undeployProcesses(duName);
    }

    public Collection<String> getPackages() {
        _rw.readLock().lock();
        try {
            return new ArrayList<String>(_deploymentUnits.keySet());
        } finally {
            _rw.readLock().unlock();
        }
    }

    public List<QName> listProcesses(String packageName) {
        _rw.readLock().lock();
        try {
            DeploymentUnitDir du = _deploymentUnits.get(packageName);
            if (du == null)
                return null;
            return toPids(du.getProcessNames(), du.getVersion());
        } finally {
            _rw.readLock().unlock();
        }
    }

    public void setState(final QName pid, final ProcessState state) {
        __log.debug("Changing process state for " + pid + " to " + state);

        final ProcessConfImpl pconf;

        _rw.readLock().lock();
        try {
            pconf = _processes.get(pid);
            if (pconf == null) {
                String msg = __msgs.msgProcessNotFound(pid);
                __log.info(msg);
                throw new ContextException(msg);
            }
        } finally {
            _rw.readLock().unlock();
        }

        final DeploymentUnitDir dudir = pconf.getDeploymentUnit();

        // Update in the database.
        ProcessState old = exec(new Callable<ProcessState>() {
            public ProcessState call(ConfStoreConnection conn) {
                DeploymentUnitDAO dudao = conn.getDeploymentUnit(dudir.getName());
                if (dudao == null) {
                    String errmsg = __msgs.msgProcessNotFound(pid);
                    __log.error(errmsg);
                    throw new ContextException(errmsg);
                }

                ProcessConfDAO dao = dudao.getProcess(pid);
                if (dao == null) {
                    String errmsg = __msgs.msgProcessNotFound(pid);
                    __log.error(errmsg);
                    throw new ContextException(errmsg);
                }

                Set processKeys = _processes.keySet();
                Iterator processConfQNameItr = processKeys.iterator();

                while(processConfQNameItr.hasNext()){
                    ProcessConf cachedProcessConf = _processes.get(processConfQNameItr.next());
                    if(dao.getType().equals(cachedProcessConf.getType())){
                        if (ProcessState.ACTIVE == cachedProcessConf.getState()
                                && ProcessState.RETIRED == dao.getState()
                                && ProcessState.ACTIVE == state) {
                            String errorMsg = "Can't activate the process with PID: " + dao.getPID() + " with version " + dao.getVersion() +
                                    ", as another version of the process with PID : " + cachedProcessConf.getProcessId() + " with version " +
                                    cachedProcessConf.getVersion() + " is already active.";
                            __log.error(errorMsg);
                            throw new ContextException(errorMsg);
                        }
                    }
                }

                ProcessState old = dao.getState();
                dao.setState(state);
                pconf.setState(state);
                return old;
            }
        });

        pconf.setState(state);
        if (old != null && old != state)
            fireStateChange(pid, state, pconf.getDeploymentUnit().getName());
    }

    public void setRetiredPackage(String packageName, boolean retired) {
        DeploymentUnitDir duDir = _deploymentUnits.get(packageName);
        if (duDir == null) throw new ContextException("Could not find package " + packageName);
        for (QName processName : duDir.getProcessNames()) {
            setState(toPid(processName, duDir.getVersion()), retired ? ProcessState.RETIRED : ProcessState.ACTIVE);
        }
    }

    public ProcessConf getProcessConfiguration(final QName processId) {
        _rw.readLock().lock();
        try {
            return _processes.get(processId);
        } finally {
            _rw.readLock().unlock();
        }
    }

    public void setProperty(final QName pid, final QName propName, final Node value) {
        setProperty(pid, propName, DOMUtils.domToStringLevel2(value));
    }

    public void setProperty(final QName pid, final QName propName, final String value) {
        if (__log.isDebugEnabled())
            __log.debug("Setting property " + propName + " on process " + pid);

        ProcessConfImpl pconf = _processes.get(pid);
        if (pconf == null) {
            String msg = __msgs.msgProcessNotFound(pid);
            __log.info(msg);
            throw new ContextException(msg);
        }

        final DeploymentUnitDir dudir = pconf.getDeploymentUnit();
        exec(new ProcessStoreImpl.Callable<Object>() {
            public Object call(ConfStoreConnection conn) {
                DeploymentUnitDAO dudao = conn.getDeploymentUnit(dudir.getName());
                if (dudao == null)
                    return null;
                ProcessConfDAO proc = dudao.getProcess(pid);
                if (proc == null)
                    return null;
                proc.setProperty(propName, value);
                return null;
            }
        });

        fireEvent(new ProcessStoreEvent(ProcessStoreEvent.Type.PROPERTY_CHANGED, pid, dudir.getName()));
    }

    /**
     * Load all the deployment units out of the store. Called on start-up.
     *
     */
    public void loadAll() {
        final ArrayList<ProcessConfImpl> loaded = new ArrayList<ProcessConfImpl>();
        exec(new Callable<Object>() {
            public Object call(ConfStoreConnection conn) {
                Collection<DeploymentUnitDAO> dus = conn.getDeploymentUnits();
                for (DeploymentUnitDAO du : dus)
                    try {
                        loaded.addAll(load(du));
                    } catch (Exception ex) {
                        __log.error("Error loading DU from store: " + du.getName(), ex);
                    }
                return null;
            }
        });

        // Dispatch DISABLED, RETIRED and ACTIVE events in that order
        Collections.sort(loaded, new Comparator<ProcessConf>() {
            public int compare(ProcessConf o1, ProcessConf o2) {
                return stateValue(o1.getState()) - stateValue(o2.getState());
            }
            int stateValue(ProcessState state) {
                if (ProcessState.DISABLED.equals(state)) return 0;
                if (ProcessState.RETIRED.equals(state)) return 1;
                if (ProcessState.ACTIVE.equals(state)) return 2;
                throw new IllegalStateException("Unexpected process state: "+state);
            }
        });
        for (ProcessConfImpl p : loaded) {
            try {
                fireStateChange(p.getProcessId(), p.getState(), p.getDeploymentUnit().getName());
            } catch (Exception except) {
                __log.error("Error while activating process: pid=" + p.getProcessId() + " package="+p.getDeploymentUnit().getName(), except);
            }
        }

    }

    public List<QName> getProcesses() {
        _rw.readLock().lock();
        try {
            return new ArrayList<QName>(_processes.keySet());
        } finally {
            _rw.readLock().unlock();
        }
    }

    public long getCurrentVersion() {
        if (_currentVersion.get() != null){
            return _currentVersion.get();
        }

        long version = exec(new Callable<Long>() {
            public Long call(ConfStoreConnection conn) {
                return conn.getNextVersion();
            }
        });

        _currentVersion.set(version);
        return _currentVersion.get();
    }

    protected void fireEvent(ProcessStoreEvent pse) {
        __log.debug("firing event: " + pse);
        for (ProcessStoreListener psl : _listeners)
            psl.onProcessStoreEvent(pse);
    }

    protected void fireStateChange(QName processId, ProcessState state, String duname) {
        switch (state) {
            case ACTIVE:
                fireEvent(new ProcessStoreEvent(ProcessStoreEvent.Type.ACTVIATED, processId, duname));
                break;
            case DISABLED:
                fireEvent(new ProcessStoreEvent(ProcessStoreEvent.Type.DISABLED, processId, duname));
                break;
            case RETIRED:
                fireEvent(new ProcessStoreEvent(ProcessStoreEvent.Type.RETIRED, processId, duname));
                break;
        }

    }

    public void registerListener(ProcessStoreListener psl) {
        __log.debug("Registering listener " + psl);
        _listeners.add(psl);
    }

    public void unregisterListener(ProcessStoreListener psl) {
        __log.debug("Unregistering listener " + psl);
        _listeners.remove(psl);
    }

    /**
     * Execute database transactions in an isolated context.
     *
     * @param <T>
     *            return type
     * @param callable
     *            transaction
     * @return
     */
    synchronized <T> T exec(Callable<T> callable) {
        // We want to submit db jobs to an executor to isolate
        // them from the current thread,
        Future<T> future = _executor.submit(callable);
        try {
            return future.get();
        } catch (Exception e) {
            throw new ContextException("DbError", e);
        }
    }

    private ConfStoreConnection getConnection() {
        return _cf.getConnection();
    }

    /**
     * Create a property mapping based on the initial values in the deployment descriptor.
     *
     * @param dd
     * @return
     */
    public static Map<QName, Node> calcInitialProperties(Properties properties, TDeployment.Process dd) {
        HashMap<QName, Node> ret = new HashMap<QName, Node>();

        for (Object key1 : properties.keySet()) {
            String key = (String) key1;
            Document doc = DOMUtils.newDocument();
            doc.appendChild(doc.createElementNS(null, "temporary-simple-type-wrapper"));
            doc.getDocumentElement().appendChild(doc.createTextNode(properties.getProperty(key)));

            ret.put(new QName(key), doc.getDocumentElement());
        }

        for (TDeployment.Process.Property property : dd.getPropertyArray()) {
            Element elmtContent = DOMUtils.getElementContent(property.getDomNode());
            if (elmtContent != null) {
                // We'll need DOM Level 3
                Document doc = DOMUtils.newDocument();
                doc.appendChild(doc.importNode(elmtContent, true));
                ret.put(property.getName(), doc.getDocumentElement());
            } else
                ret.put(property.getName(), property.getDomNode().getFirstChild());

        }
        return ret;
    }

    /**
     * Figure out the initial process state from the state in the deployment descriptor.
     *
     * @param dd
     *            deployment descriptor
     * @return
     */
    private static ProcessState calcInitialState(TDeployment.Process dd) {
        ProcessState state = ProcessState.ACTIVE;

        if (dd.isSetActive() && dd.getActive() == false)
            state = ProcessState.DISABLED;
        if (dd.isSetRetired() && dd.getRetired() == true)
            state = ProcessState.RETIRED;

        return state;
    }

    /**
     * Load a deployment unit record stored in the db into memory.
     *
     * @param dudao
     */
    protected List<ProcessConfImpl> load(DeploymentUnitDAO dudao) {
        __log.debug("Loading deployment unit record from db: " + dudao.getName());

        File dudir = findDeployDir(dudao);

        if (dudir == null || !dudir.exists())
            throw new ContextException("Deployed directory " + (dudir == null ? "(unknown)" : dudir) + " no longer there!");
        DeploymentUnitDir dud = new DeploymentUnitDir(dudir);
        // set the name with the one from database
        dud.setName(dudao.getName());
        dud.scan();

        ArrayList<ProcessConfImpl> loaded = new ArrayList<ProcessConfImpl>();

        _rw.writeLock().lock();
        try {
            _deploymentUnits.put(dud.getName(), dud);

            long version = 0;
            for (ProcessConfDAO p : dudao.getProcesses()) {
                TDeployment.Process pinfo = dud.getProcessDeployInfo(p.getType());
                if (pinfo == null) {
                    __log.warn("Cannot load " + p.getPID() + "; cannot find descriptor.");
                    continue;
                }

                Map<QName, Node> props = calcInitialProperties(dud.getProperties(), pinfo);
                // TODO: update the props based on the values in the DB.

                ProcessConfImpl pconf = new ProcessConfImpl(p.getPID(), p.getType(), p.getVersion(), dud, pinfo, dudao
                        .getDeployDate(), props, p.getState(), eprContext, _configDir, generateProcessEventsAll);
                version = p.getVersion();

                _processes.put(pconf.getProcessId(), pconf);
                loaded.add(pconf);
            }

            // All processes and the DU have the same version
            dud.setVersion(version);
        } finally {
            _rw.writeLock().unlock();
        }

        return loaded;
    }

    protected File findDeployDir(DeploymentUnitDAO dudao) {
        File f = new File(dudao.getDeploymentUnitDir());
        if (f.exists())
            return f;
        f = new File(_deployDir, dudao.getName());
        if (f.exists()) {
            try {
                dudao.setDeploymentUnitDir(f.getCanonicalPath());
            } catch (IOException e) {
                __log.warn("Could not update deployment unit directory for " + dudao.getName(), e);
            }
            return f;
        }

        return null;
    }

    /**
     * Make sure that the deployment unit is loaded.
     *
     * @param duName
     *            deployment unit name
     */
    protected boolean load(final String duName) {
        _rw.writeLock().lock();
        try {
            if (_deploymentUnits.containsKey(duName))
                return true;
        } finally {
            _rw.writeLock().unlock();
        }

        try {
            return exec(new Callable<Boolean>() {
                public Boolean call(ConfStoreConnection conn) {
                    DeploymentUnitDAO dudao = conn.getDeploymentUnit(duName);
                    if (dudao == null)
                        return false;
                    load(dudao);
                    return true;
                }
            });
        } catch (Exception ex) {
            __log.error("Error loading deployment unit: " + duName);
            return false;
        }

    }

    /**
     * Wrapper for database transactions.
     *
     * @author Maciej Szefler
     *
     * @param <V>
     *            return type
     */
    abstract class Callable<V> implements java.util.concurrent.Callable<V> {
        public V call() {
            boolean success = false;
            // in JTA, transaction is bigger than the session
            _cf.beginTransaction();
            ConfStoreConnection conn = getConnection();
            try {
                V r = call(conn);
                _cf.commitTransaction();
                success = true;
                return r;
            } finally {
                if (!success)
                    try {
                        _cf.rollbackTransaction();
                    } catch (Exception ex) {
                        __log.error("DbError", ex);
                    }
            }
         // session is closed automatically when committed or rolled back under JTA
        }

        abstract V call(ConfStoreConnection conn);
    }

    public void setDeployDir(File depDir) {
        if (depDir != null) {
            if( !depDir.exists() ) {
                depDir.mkdirs();
                __log.warn("Deploy directory: " + depDir.getAbsolutePath() + " does not exist; created it.");
            } else if(!depDir.isDirectory()) {
                throw new IllegalArgumentException("Deploy directory is not a directory:  " + depDir);
            }
        }

        _deployDir = depDir;
    }

    public File getDeployDir() {
        return _deployDir;
    }

    public File getConfigDir() {
        return _configDir;
    }

    public void setConfigDir(File configDir) {
        if (configDir != null && !configDir.isDirectory())
            throw new IllegalArgumentException("Config directory is not a directory or does not exist: " + configDir);
        this._configDir = configDir;
    }

    public static DataSource createInternalDS(String guid) {
        JdbcDataSource h2 = new JdbcDataSource();
        h2.setURL("jdbc:h2:mem:" + new GUID().toString()+";DB_CLOSE_DELAY=-1");
        h2.setUser("sa");
        return h2;
    }

    public static void shutdownInternalDB(DataSource ds) {
        try {
            ds.getConnection().createStatement().execute("SHUTDOWN;");
        } catch (SQLException e) {
            __log.error("Error shutting down.", e);
        }
    }

    private List<QName> toPids(Collection<QName> processTypes, long version) {
        ArrayList<QName> result = new ArrayList<QName>();
        for (QName pqName : processTypes) {
            result.add(toPid(pqName, version));
        }
        return result;
    }

    protected QName toPid(QName processType, long version) {
        return new QName(processType.getNamespaceURI(), processType.getLocalPart() + "-" + version);
    }

    private class SimpleThreadFactory implements ThreadFactory {
        int threadNumber = 0;
        public Thread newThread(Runnable r) {
            threadNumber += 1;
            Thread t = new Thread(r, "ProcessStoreImpl-"+threadNumber);
            t.setDaemon(true);
            return t;
        }
    }

    public void refreshSchedules(String packageName) {
        List<QName> pids = listProcesses(packageName);
        if (pids != null) {
            for( QName pid : pids ) {
                fireEvent(new ProcessStoreEvent(ProcessStoreEvent.Type.SCHEDULE_SETTINGS_CHANGED, pid, packageName));
            }
        }
    }

    protected Pattern getPreviousPackageVersionPattern(String duName) {
        //retire all the other versions of the same DU
        String[] nameParts = duName.split("/");
        /* Replace the version number (if any) with regexp to match any version number */
        nameParts[0] = nameParts[0].replaceAll("([-\\Q.\\E](\\d)+)?\\z", "");
        nameParts[0] += "([-\\Q.\\E](\\d)+)?";
        StringBuilder duNameRegExp = new StringBuilder(duName.length() * 2);
        for (int i = 0, n = nameParts.length; i < n; i++) {
            if (i > 0) duNameRegExp.append("/");
            duNameRegExp.append(nameParts[i]);
        }
        Pattern duNamePattern = Pattern.compile(duNameRegExp.toString());
        return duNamePattern;
    }

    protected  Collection<QName> undeployProcesses(final String duName) {
        Collection<QName> undeployed = Collections.emptyList();
        DeploymentUnitDir du;
        _rw.writeLock().lock();
        try {
            du = _deploymentUnits.remove(duName);
            if (du != null) {
                undeployed = toPids(du.getProcessNames(), du.getVersion());
            }

            for (QName pn : undeployed) {
                fireEvent(new ProcessStoreEvent(ProcessStoreEvent.Type.UNDEPLOYED, pn, du.getName()));
                __log.info(__msgs.msgProcessUndeployed(pn));
            }

            _processes.keySet().removeAll(undeployed);
        } finally {
            _rw.writeLock().unlock();
        }
        return undeployed;
    }
}