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

import java.io.File;
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;

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

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.ode.agents.memory.SizingAgent;
import org.apache.ode.bpel.common.FaultException;
import org.apache.ode.bpel.common.ProcessState;
import org.apache.ode.bpel.dao.BpelDAOConnection;
import org.apache.ode.bpel.dao.MessageExchangeDAO;
import org.apache.ode.bpel.dao.ProcessDAO;
import org.apache.ode.bpel.dao.ProcessInstanceDAO;
import org.apache.ode.bpel.engine.extvar.ExternalVariableConf;
import org.apache.ode.bpel.engine.extvar.ExternalVariableManager;
import org.apache.ode.bpel.evt.ProcessInstanceEvent;
import org.apache.ode.bpel.explang.ConfigurationException;
import org.apache.ode.bpel.explang.EvaluationException;
import org.apache.ode.bpel.iapi.BpelEngineException;
import org.apache.ode.bpel.iapi.Endpoint;
import org.apache.ode.bpel.iapi.EndpointReference;
import org.apache.ode.bpel.iapi.Message;
import org.apache.ode.bpel.iapi.MessageExchange;
import org.apache.ode.bpel.iapi.MessageExchange.Status;
import org.apache.ode.bpel.iapi.PartnerRoleChannel;
import org.apache.ode.bpel.iapi.PartnerRoleMessageExchange;
import org.apache.ode.bpel.iapi.ProcessConf;
import org.apache.ode.bpel.iapi.ProcessConf.CLEANUP_CATEGORY;
import org.apache.ode.bpel.iapi.Scheduler.JobDetails;
import org.apache.ode.bpel.iapi.Scheduler.JobType;
import org.apache.ode.bpel.intercept.InstanceCountThrottler;
import org.apache.ode.bpel.intercept.InterceptorInvoker;
import org.apache.ode.bpel.intercept.MessageExchangeInterceptor;
import org.apache.ode.bpel.obj.OElementVarType;
import org.apache.ode.bpel.obj.OExpressionLanguage;
import org.apache.ode.bpel.obj.OMessageVarType;
import org.apache.ode.bpel.obj.OPartnerLink;
import org.apache.ode.bpel.obj.OProcess;
import org.apache.ode.bpel.obj.serde.DeSerializer;
import org.apache.ode.bpel.runtime.BpelRuntimeContext;
import org.apache.ode.bpel.runtime.ExpressionLanguageRuntimeRegistry;
import org.apache.ode.bpel.runtime.InvalidProcessException;
import org.apache.ode.bpel.runtime.PROCESS;
import org.apache.ode.bpel.runtime.PropertyAliasEvaluationContext;
import org.apache.ode.bpel.runtime.channels.FaultData;
import org.apache.ode.jacob.soup.ReplacementMap;
import org.apache.ode.utils.ObjectPrinter;
import org.apache.ode.utils.Properties;
import org.apache.ode.utils.msg.MessageBundle;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.w3c.dom.Text;

/**
 * Entry point into the runtime of a BPEL process.
 *
 * @author mszefler
 * @author Matthieu Riou <mriou at apache dot org>
 */
public class BpelProcess {
    static final Logger __log = LoggerFactory.getLogger(BpelProcess.class);

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

    private volatile Map<OPartnerLink, PartnerLinkPartnerRoleImpl> _partnerRoles;

    private volatile Map<OPartnerLink, PartnerLinkMyRoleImpl> _myRoles;

    /** Mapping from a myrole to a {"Service Name" (QNAME) / port}. It's actually more a tuple than a map as
     * it's important to note that the same process with the same endpoint can have 2 different myroles. */
    private volatile Map<PartnerLinkMyRoleImpl, Endpoint> _endpointToMyRoleMap;

    /** Mapping from a potentially shared endpoint to its EPR */
    private SharedEndpoints _sharedEps;

    // Backup hashmaps to keep initial endpoints handy after dehydration
    private Map<Endpoint, EndpointReference> _myEprs = new HashMap<Endpoint, EndpointReference>();
    private Map<Endpoint, EndpointReference> _partnerEprs = new HashMap<Endpoint, EndpointReference>();
    private Map<Endpoint, PartnerRoleChannel> _partnerChannels = new HashMap<Endpoint, PartnerRoleChannel>();

    final QName _pid;
    private volatile OProcess _oprocess;
    // Has the process already been hydrated before?
    private boolean _hydratedOnce = false;
    /** Last time the process was used. */
    private volatile long _lastUsed;

    BpelEngineImpl _engine;
    ClassLoader _classLoader = getClass().getClassLoader();

    DebuggerSupport _debugger;
    ExpressionLanguageRuntimeRegistry _expLangRuntimeRegistry;
    private ReplacementMap _replacementMap;
    final ProcessConf _pconf;

    /** {@link MessageExchangeInterceptor}s registered for this process. */
    private final List<MessageExchangeInterceptor> _mexInterceptors = new ArrayList<MessageExchangeInterceptor>();

    /** Latch-like thing to control hydration/dehydration. */
    private HydrationLatch _hydrationLatch;

    /** Deploy-time configuraton for external variables. */
    private ExternalVariableConf _extVarConf;

    private ExternalVariableManager _evm;

    public static final QName PROP_PATH = new QName("PATH");
    public static final QName PROP_SVG = new QName("SVG");
    public static final QName PROP_LAZY_HYDRATE = new QName("process.hydration.lazy");
    public static final QName PROP_MAX_INSTANCES = new QName("process.instance.throttled.maximum.count");

    // The ratio of in-memory vs serialized size of compiled bpel object.
    private static final int PROCESS_MEMORY_TO_SERIALIZED_SIZE_RATIO = 5;

    public BpelProcess(ProcessConf conf) {
        _pid = conf.getProcessId();
        _pconf = conf;
        _hydrationLatch = new HydrationLatch();
    }


    /**
     * Retrives the base URI to use for local resource resolution.
     *
     * @return URI - instance representing the absolute file path to the physical location of the process definition folder.
     */
    public URI getBaseResourceURI() {
        return this._pconf.getBaseURI();
    }

    /**
     * Intiialize the external variable configuration/engine manager. This is called from hydration logic, so it
     * is possible to change the external variable configuration at runtime.
     *
     */
    void initExternalVariables() {
        List<Element> conf = _pconf.getExtensionElement(ExternalVariableConf.EXTVARCONF_ELEMENT);
        _extVarConf = new ExternalVariableConf(conf);
        _evm = new ExternalVariableManager(_pid, _extVarConf, _engine._contexts.externalVariableEngines, _oprocess);
    }


    public String toString() {
        return "BpelProcess[" + _pid + "]";
    }

    public ExternalVariableManager getEVM() {
        return _evm;
    }

    public void recoverActivity(ProcessInstanceDAO instanceDAO, String channel, long activityId, String action, FaultData fault) {
        if (__log.isDebugEnabled())
            __log.debug("Recovering activity in process " + instanceDAO.getInstanceId() + " with action " + action);
        markused();
        BpelRuntimeContextImpl processInstance = createRuntimeContext(instanceDAO, null, null);
        processInstance.recoverActivity(channel, activityId, action, fault);
    }

    protected DebuggerSupport createDebuggerSupport() {
        return new DebuggerSupport(this);
    }

    protected DebuggerSupport getDebuggerSupport() {
        return _debugger;
    }

    static String generateMessageExchangeIdentifier(String partnerlinkName, String operationName) {
        StringBuffer sb = new StringBuffer(partnerlinkName);
        sb.append('.');
        sb.append(operationName);
        return sb.toString();
    }

    public interface InvokeHandler {
        boolean invoke(PartnerLinkMyRoleImpl target, PartnerLinkMyRoleImpl.RoutingInfo routingInfo, boolean createInstance);
    }

    public boolean invokeProcess(MyRoleMessageExchangeImpl mex, InvokeHandler invokeHandler, boolean enqueue) {
        boolean routed = false;

        try {
            _hydrationLatch.latch(1);
            List<PartnerLinkMyRoleImpl> targets = getMyRolesForService(mex.getServiceName());
            if (targets.isEmpty()) {
//                String errmsg = __msgs.msgMyRoleRoutingFailure(mex.getMessageExchangeId());
//                __log.error(errmsg);
//                mex.setFailure(MessageExchange.FailureType.UNKNOWN_ENDPOINT, errmsg, null);
                return false;
            }

            mex.getDAO().setProcess(getProcessDAO());

            if (!processInterceptors(mex, InterceptorInvoker.__onProcessInvoked)) {
                if (__log.isDebugEnabled()) {
                    __log.debug("Aborting processing of mex " + mex + " due to interceptors.");
                }
                return false;
            }

            markused();

            // Ideally, if Java supported closure, the routing code would return null or the appropriate
            // closure to handle the route.
            List<PartnerLinkMyRoleImpl.RoutingInfo> routings = null;
            for (PartnerLinkMyRoleImpl target : targets) {
                routings = target.findRoute(mex);
                boolean createInstance = target.isCreateInstance(mex);

                if (mex.getStatus() != MessageExchange.Status.FAILURE && routings!=null) {
                    for (PartnerLinkMyRoleImpl.RoutingInfo routing : routings) {
                        routed = routed || invokeHandler.invoke(target, routing, createInstance);
                    }
                }
                if (routed) {
                    break;
                }
            }

            // Nothing found, saving for later
            if (!routed && enqueue) {
                // TODO this is kind of hackish when no match and more than one myrole is selected.
                // we save the routing on the last myrole
                // actually the message queue should be attached to the instance instead of the correlator
                targets.get(targets.size()-1).noRoutingMatch(mex, routings);
            } else {
                // Now we have to update our message exchange status. If the <reply> was not hit during the
                // invocation, then we will be in the "REQUEST" phase which means that either this was a one-way
                // or a two-way that needs to delivery the reply asynchronously.
                if (mex.getStatus() == MessageExchange.Status.REQUEST) {
                    mex.setStatus(MessageExchange.Status.ASYNC);
                }

                markused();
            }
        } finally {
            _hydrationLatch.release(1);
        }

        // For a one way, once the engine is done, the mex can be safely released.
        // Sean: not really, if route is not found, we cannot delete the mex yet
        // Rafal: We don't do cleanup here
//        if (mex.getPattern().equals(MessageExchange.MessageExchangePattern.REQUEST_ONLY) && routed && getCleanupCategories(false).contains(CLEANUP_CATEGORY.MESSAGES)) {
//            mex.release();
//        }
        return routed;
    }

    private boolean isActive() {
        return _pconf.getState() == org.apache.ode.bpel.iapi.ProcessState.ACTIVE;
    }

    /**
     * Entry point for message exchanges aimed at the my role.
     *
     * @param mex
     */
    boolean invokeProcess(final MyRoleMessageExchangeImpl mex, boolean enqueue) {
        return invokeProcess(mex, new InvokeHandler() {
            public boolean invoke(PartnerLinkMyRoleImpl target, PartnerLinkMyRoleImpl.RoutingInfo routing, boolean createInstance) {
                  if (routing.messageRoute == null && createInstance && isActive()) {
                      // No route but we can create a new instance
                      target.invokeNewInstance(mex, routing);
                      return true;
                  } else if (routing.messageRoute != null) {
                      // Found a route, hitting it
                      _engine.acquireInstanceLock(routing.messageRoute.getTargetInstance().getInstanceId());
                      target.invokeInstance(mex, routing);
                      return true;
                  }
                  return false;
            }
        }, enqueue);
    }

    /** Several myroles can use the same service in a given process */
    private List<PartnerLinkMyRoleImpl> getMyRolesForService(QName serviceName) {
        List<PartnerLinkMyRoleImpl> myRoles = new ArrayList<PartnerLinkMyRoleImpl>(5);
        for (Map.Entry<PartnerLinkMyRoleImpl,Endpoint> e : getEndpointToMyRoleMap().entrySet()) {
            if (e.getValue().serviceName.equals(serviceName))
                myRoles.add(e.getKey());
        }
        return myRoles;
    }

    void initMyRoleMex(MyRoleMessageExchangeImpl mex) {
        markused();

        PartnerLinkMyRoleImpl target = null;
        for (Map.Entry<PartnerLinkMyRoleImpl,Endpoint> e : getEndpointToMyRoleMap().entrySet()) {
            if (e.getValue().serviceName.equals(mex.getServiceName())) {
                // First one is fine as we're only interested in the portType and operation here and
                // even if a process has 2 myrole partner links
                target = e.getKey();
                break;
            }
        }
        if (target != null) {
            mex.setPortOp(target._plinkDef.getMyRolePortType(), target._plinkDef.getMyRoleOperation(mex.getOperationName()));
        } else {
            __log.warn("Couldn't find endpoint from service " + mex.getServiceName() + " when initializing a myRole mex.");
        }
    }

    /**
     * Extract the value of a BPEL property from a BPEL messsage variable.
     *
     * @param msgData
     *            message variable data
     * @param alias
     *            alias to apply
     * @param target
     *            description of the data (for error logging only)
     * @return value of the property
     * @throws FaultException
     */
    String extractProperty(Element msgData, Map<String, Node> headerParts, OProcess.OPropertyAlias alias, String target) throws FaultException {
        markused();
        PropertyAliasEvaluationContext ectx = new PropertyAliasEvaluationContext(msgData, headerParts, alias);
        Node lValue = ectx.getRootNode();

        if (alias.getLocation() != null) {
            try {
                lValue = _expLangRuntimeRegistry.evaluateNode(alias.getLocation(), ectx);
            } catch (EvaluationException ec) {
                throw new FaultException(getOProcess().getConstants().getQnSelectionFailure(), alias.getDescription());
            }
        }

        if (lValue == null) {
            String errmsg = __msgs.msgPropertyAliasReturnedNullSet(alias.getDescription(), target);
            if (__log.isErrorEnabled()) {
                __log.error(errmsg);
            }
            throw new FaultException(getOProcess().getConstants().getQnSelectionFailure(), errmsg);
        }

        if (lValue.getNodeType() == Node.ELEMENT_NODE) {
            // This is a bit hokey, we concatenate all the children's values; we
            // really should be checking to make sure that we are only dealing
            // with text and attribute nodes.
            StringBuffer val = new StringBuffer();
            NodeList nl = lValue.getChildNodes();
            for (int i = 0; i < nl.getLength(); ++i) {
                Node n = nl.item(i);
                val.append(n.getNodeValue());
            }
            return val.toString();
        } else if (lValue.getNodeType() == Node.TEXT_NODE) {
            return ((Text) lValue).getWholeText();
        } else
            return null;
    }

    /**
     * Get the element name for a given WSDL part. If the part is an <em>element</em> part, the name of that element is returned.
     * If the part is an XML schema typed part, then the name of the part is returned in the null namespace.
     *
     * @param part
     *            WSDL {@link javax.wsdl.Part}
     * @return name of element containing said part
     */
    static QName getElementNameForPart(OMessageVarType.Part part) {
        return (part.getType() instanceof OElementVarType) ? ((OElementVarType) part.getType()).getElementType() : new QName(null, part.getName());
    }

    /**
     * Process the message-exchange interceptors.
     *
     * @param mex
     *            message exchange
     * @return <code>true</code> if execution should continue, <code>false</code> otherwise
     */
    public boolean processInterceptors(MyRoleMessageExchangeImpl mex, InterceptorInvoker invoker) {
        InterceptorContextImpl ictx = new InterceptorContextImpl(_engine._contexts.dao.getConnection(), getProcessDAO(), _pconf, _engine, this);

        for (MessageExchangeInterceptor i : _mexInterceptors)
            if (!mex.processInterceptor(i, mex, ictx, invoker))
                return false;
        for (MessageExchangeInterceptor i : getEngine().getGlobalInterceptors())
            if (!mex.processInterceptor(i, mex, ictx, invoker))
                return false;

        return true;
    }

    /**
     * @see org.apache.ode.bpel.engine.BpelProcess#handleJobDetails(java.util.Map<java.lang.String,java.lang.Object>)
     */
    public boolean handleJobDetails(JobDetails jobData) {
        boolean routed = true;
        try {
            _hydrationLatch.latch(1);
            markused();
            if (__log.isDebugEnabled()) {
                __log.debug(ObjectPrinter.stringifyMethodEnter("handleJobDetails", new Object[] { "jobData", jobData }));
            }

            JobDetails we = jobData;

            // Process level events
            if (we.getType() == JobType.INVOKE_INTERNAL || we.getType() == JobType.MEX_MATCHER) {
                if (__log.isDebugEnabled()) {
                    __log.debug(we.getType() + " event for mexid " + we.getMexId());
                }
                MyRoleMessageExchangeImpl mex = (MyRoleMessageExchangeImpl) _engine.getMessageExchange(we.getMexId());
            	if (we.getType() == JobType.MEX_MATCHER && !mex.getDAO().lockPremieMessages()) {
            		//Skip if already processed
            		return true;
            	}

                routed = invokeProcess(mex, (Boolean) jobData.detailsExt.get("enqueue"));
                if (we.getType() == JobType.MEX_MATCHER && routed) {
                	mex.getDAO().releasePremieMessages();
                }
            } else {
                // Instance level events
                ProcessInstanceDAO procInstance = getProcessDAO().getInstance(we.getInstanceId());
                if (procInstance == null) {
                    if (__log.isDebugEnabled()) {
                        __log.debug("handleJobDetails: no ProcessInstance found with iid " + we.getInstanceId() + "; ignoring.");
                    }
                    return true;
                }

                BpelRuntimeContextImpl processInstance = createRuntimeContext(procInstance, null, null);
                switch (we.getType()) {
                    case TIMER:
                        if (__log.isDebugEnabled()) {
                            __log.debug("handleJobDetails: TimerWork event for process instance " + processInstance);
                        }
                        processInstance.timerEvent(we.getChannel());
                        break;
                    case RESUME:
                        if (__log.isDebugEnabled()) {
                            __log.debug("handleJobDetails: ResumeWork event for iid " + we.getInstanceId());
                        }
                        processInstance.execute();
                        break;
                    case INVOKE_RESPONSE:
                        if (__log.isDebugEnabled()) {
                            __log.debug("InvokeResponse event for iid " + we.getInstanceId());
                        }
                        processInstance.invocationResponse(we.getMexId(), we.getChannel());
                        processInstance.execute();
                        break;
                    case MATCHER:
                        if (__log.isDebugEnabled()) {
                            __log.debug("Matcher event for iid " + we.getInstanceId());
                        }
                        if( procInstance.getState() == ProcessState.STATE_COMPLETED_OK
                                || procInstance.getState() == ProcessState.STATE_COMPLETED_WITH_FAULT ) {
                            __log.debug("A matcher event was aborted. The process is already completed.");
                            return true;
                        }
                        processInstance.matcherEvent(we.getCorrelatorId(), we.getCorrelationKeySet());
                }
            }
        } finally {
            _hydrationLatch.release(1);
        }
        return routed;
    }

    private void setRoles(OProcess oprocess) {
        _partnerRoles = new HashMap<OPartnerLink, PartnerLinkPartnerRoleImpl>();
        _myRoles = new HashMap<OPartnerLink, PartnerLinkMyRoleImpl>();
        _endpointToMyRoleMap = new HashMap<PartnerLinkMyRoleImpl, Endpoint>();

        // Create myRole endpoint name mapping (from deployment descriptor)
        HashMap<OPartnerLink, Endpoint> myRoleEndpoints = new HashMap<OPartnerLink, Endpoint>();
        for (Map.Entry<String, Endpoint> provide : _pconf.getProvideEndpoints().entrySet()) {
            OPartnerLink plink = oprocess.getPartnerLink(provide.getKey());
            if (plink == null) {
                String errmsg = "Error in deployment descriptor for process " + _pid + "; reference to unknown partner link "
                        + provide.getKey();
                __log.error(errmsg);
                throw new BpelEngineException(errmsg);
            }
            myRoleEndpoints.put(plink, provide.getValue());
        }

        // Create partnerRole initial value mapping
        for (Map.Entry<String, Endpoint> invoke : _pconf.getInvokeEndpoints().entrySet()) {
            OPartnerLink plink = oprocess.getPartnerLink(invoke.getKey());
            if (plink == null) {
                String errmsg = "Error in deployment descriptor for process " + _pid + "; reference to unknown partner link "
                        + invoke.getKey();
                __log.error(errmsg);
                throw new BpelEngineException(errmsg);
            }
            if (__log.isDebugEnabled()) {
                __log.debug("Processing <invoke> element for process " + _pid + ": partnerlink " + invoke.getKey() + " --> "
                    + invoke.getValue());
            }
        }

        for (OPartnerLink pl : oprocess.getAllPartnerLinks()) {
            if (pl.hasMyRole()) {
                Endpoint endpoint = myRoleEndpoints.get(pl);
                if (endpoint == null)
                    throw new IllegalArgumentException("No service name for myRole plink " + pl.getName());
                PartnerLinkMyRoleImpl myRole = new PartnerLinkMyRoleImpl(this, pl, endpoint);
                _myRoles.put(pl, myRole);
                _endpointToMyRoleMap.put(myRole, endpoint);
            }

            if (pl.hasPartnerRole()) {
                Endpoint endpoint = _pconf.getInvokeEndpoints().get(pl.getName());
                if (endpoint == null && pl.isInitializePartnerRole())
                    throw new IllegalArgumentException(pl.getName() + " must be bound to an endpoint in deploy.xml");
                PartnerLinkPartnerRoleImpl partnerRole = new PartnerLinkPartnerRoleImpl(this, pl, endpoint);
                _partnerRoles.put(pl, partnerRole);
            }
        }
    }

    public ProcessDAO getProcessDAO() {
        return _pconf.isTransient() ? _engine._contexts.inMemDao.getConnection().getProcess(_pid) : getEngine()._contexts.dao
                .getConnection().getProcess(_pid);
    }

    static String genCorrelatorId(OPartnerLink plink, String opName) {
        return plink.getName() + "." + opName;
    }

    /**
     * De-serialize the compiled process representation from a stream.
     *
     * @param is
     *            input stream
     * @return process information from configuration database
     */
    private OProcess deserializeCompiledProcess(File file) throws Exception {
        OProcess compiledProcess;
        DeSerializer deserializer = new DeSerializer(file);
        compiledProcess = deserializer.deserialize();
        return compiledProcess;
    }

    /**
     * Get all the services that are implemented by this process.
     *
     * @return list of qualified names corresponding to the myroles.
     */
    public Set<Endpoint> getServiceNames() {
        Set<Endpoint> endpoints = new HashSet<Endpoint>();
        for (Endpoint provide : _pconf.getProvideEndpoints().values()) {
            endpoints.add(provide);
        }
        return endpoints;
    }

    void activate(BpelEngineImpl engine) {
        _engine = engine;
        _sharedEps = _engine.getSharedEndpoints();
        _debugger = createDebuggerSupport();

        if (getInstanceMaximumCount() < Integer.MAX_VALUE)
            registerMessageExchangeInterceptor(new InstanceCountThrottler());

        if (__log.isDebugEnabled()) {
            __log.debug("Activating " + _pid);
        }
        // Activate all the my-role endpoints.
        for (Map.Entry<String, Endpoint> entry : _pconf.getProvideEndpoints().entrySet()) {
            Endpoint endpoint = entry.getValue();
            EndpointReference initialEPR = null;
            if (isShareable(endpoint)) {
                // Check if the EPR already exists for the given endpoint
                initialEPR = _sharedEps.getEndpointReference(endpoint);
                if (initialEPR == null) {
                    // Create an EPR by physically activating the endpoint
                    initialEPR = _engine._contexts.bindingContext.activateMyRoleEndpoint(_pid, endpoint);
                    _sharedEps.addEndpoint(endpoint, initialEPR);
                    if (__log.isDebugEnabled()) {
                        __log.debug("Activated " + _pid + " myrole " + entry.getKey() + ": EPR is " + initialEPR);
                    }
                }
                // Increment the reference count on the endpoint
                _sharedEps.incrementReferenceCount(endpoint);
            } else {
                // Create an EPR by physically activating the endpoint
                initialEPR = _engine._contexts.bindingContext.activateMyRoleEndpoint(_pid, endpoint);
                if (__log.isDebugEnabled()) {
                    __log.debug("Activated " + _pid + " myrole " + entry.getKey() + ": EPR is " + initialEPR);
                }
            }
            _myEprs.put(endpoint, initialEPR);
        }
        if (__log.isDebugEnabled()) {
            __log.debug("Activated " + _pid);
        }

        markused();
    }

    void deactivate() {
        // Deactivate all the my-role endpoints.
        for (Endpoint endpoint : _myEprs.keySet()) {
            // Deactivate the EPR only if there are no more references
            // to this endpoint from any (active) BPEL process.
            if (isShareable(endpoint)) {
                if(__log.isDebugEnabled()) __log.debug("deactivating shared endpoint " + endpoint+ " for pid "+ _pid);
                if (!_sharedEps.decrementReferenceCount(endpoint)) {
                    _engine._contexts.bindingContext.deactivateMyRoleEndpoint(endpoint);
                    _sharedEps.removeEndpoint(endpoint);
                }
            } else {
                if(__log.isDebugEnabled()) __log.debug("deactivating non-shared endpoint " + endpoint + " for pid "+ _pid);
                _engine._contexts.bindingContext.deactivateMyRoleEndpoint(endpoint);
            }
        }
        // TODO Deactivate all the partner-role channels
    }

    private boolean isShareable(Endpoint endpoint) {
        return _pconf.isSharedService(endpoint.serviceName);
    }

    protected EndpointReference getInitialPartnerRoleEPR(OPartnerLink link) {
        try {
            _hydrationLatch.latch(1);
            PartnerLinkPartnerRoleImpl prole = _partnerRoles.get(link);
            if (prole == null)
                throw new IllegalStateException("Unknown partner link " + link);
            return prole.getInitialEPR();
        } finally {
            _hydrationLatch.release(1);
        }
    }

    protected Endpoint getInitialPartnerRoleEndpoint(OPartnerLink link) {
        try {
            _hydrationLatch.latch(1);
            PartnerLinkPartnerRoleImpl prole = _partnerRoles.get(link);
            if (prole == null)
                throw new IllegalStateException("Unknown partner link " + link);
            return prole._initialPartner;
        } finally {
            _hydrationLatch.release(1);
        }
    }

    protected EndpointReference getInitialMyRoleEPR(OPartnerLink link) {
        try {
            _hydrationLatch.latch(1);
            PartnerLinkMyRoleImpl myRole = _myRoles.get(link);
            if (myRole == null)
                throw new IllegalStateException("Unknown partner link " + link);
            return myRole.getInitialEPR();
        } finally {
            _hydrationLatch.release(1);
        }
    }

    public QName getPID() {
        return _pid;
    }

    protected PartnerRoleChannel getPartnerRoleChannel(OPartnerLink partnerLink) {
        try {
            _hydrationLatch.latch(1);
            PartnerLinkPartnerRoleImpl prole = _partnerRoles.get(partnerLink);
            if (prole == null)
                throw new IllegalStateException("Unknown partner link " + partnerLink);
            return prole._channel;
        } finally {
            _hydrationLatch.release(1);
        }
    }

    public void saveEvent(ProcessInstanceEvent event, ProcessInstanceDAO instanceDao) {
        saveEvent(event, instanceDao, null);
    }

    public void saveEvent(ProcessInstanceEvent event, ProcessInstanceDAO instanceDao, List<String> scopeNames) {
        markused();
        if (_pconf.isEventEnabled(scopeNames, event.getType())) {
            // notify the listeners
            _engine.fireEvent(event);

            if (instanceDao != null && !event.isEventPersistingCancelled())
                instanceDao.insertBpelEvent(event);
            else
                __log.debug("Couldn't find instance to save event, no event generated!");
        }
    }

    /**
     * Ask the process to dehydrate.
     */
    void dehydrate() {
        try {
            _hydrationLatch.latch(0);
            // We don't actually need to do anything, the latch will run the doDehydrate method
            // when necessary..
        } finally {
            _hydrationLatch.release(0);
        }

    }

    void hydrate() {
        try {
            _hydrationLatch.latch(1);
            // We don't actually need to do anything, the latch will run the doHydrate method
            // when necessary..
        } finally {
            _hydrationLatch.release(1);
        }
    }

    public OProcess getOProcess() {
        try {
            _hydrationLatch.latch(1);
            return _oprocess;
        } finally {
            _hydrationLatch.release(1);
        }
    }

    private Map<PartnerLinkMyRoleImpl,Endpoint> getEndpointToMyRoleMap() {
        try {
            _hydrationLatch.latch(1);
            return _endpointToMyRoleMap;
        } finally {
            _hydrationLatch.release(1);
        }
    }

    public ReplacementMap getReplacementMap(QName processName) {
        try {
            _hydrationLatch.latch(1);

            if (processName.equals(_pid)) return _replacementMap;
            else
                try {
                    // We're asked for an older version of this process, fetching it
                    OProcess oprocess = _engine.getOProcess(processName);
                    if (oprocess == null) {
                        String errmsg = "The process " + _pid + " is not available anymore.";
                        __log.error(errmsg);
                        throw new InvalidProcessException(errmsg, InvalidProcessException.RETIRED_CAUSE_CODE);
                    }
                    // Older versions may ventually need more expression languages
                    registerExprLang(oprocess);

                    return new ReplacementMapImpl(oprocess);
                } catch (Exception e) {
                    String errmsg = "The process " + _pid + " is not available anymore.";
                    __log.error(errmsg, e);
                    throw new InvalidProcessException(errmsg, InvalidProcessException.RETIRED_CAUSE_CODE);
                }
        } finally {
            _hydrationLatch.release(1);
        }
    }

    protected BpelEngineImpl getEngine() {
        return _engine;
    }

    public boolean isInMemory() {
        return _pconf.isTransient();
    }

    public long getLastUsed() {
        return _lastUsed;
    }

    QName getProcessType() {
        return _pconf.getType();
    }

    /**
     * Get a hint as to whether this process is hydrated. Note this is only a hint, since things could change.
     */
    public boolean hintIsHydrated() {
        return _oprocess != null;
    }

    /** Keep track of the time the process was last used. */
    private final void markused() {
        _lastUsed = System.currentTimeMillis();
    }

    /** Create a version-appropriate runtime context. */
    protected BpelRuntimeContextImpl createRuntimeContext(ProcessInstanceDAO dao, PROCESS template,
                                                MyRoleMessageExchangeImpl instantiatingMessageExchange) {
        return new BpelRuntimeContextImpl(this, dao, template, instantiatingMessageExchange);

    }

    private class HydrationLatch extends NStateLatch {
        HydrationLatch() {
            super(new Runnable[2]);
            _transitions[0] = new Runnable() {
                public void run() {
                    doDehydrate();
                }
            };
            _transitions[1] = new Runnable() {
                public void run() {
                    doHydrate();
                }
            };
        }

        private void doDehydrate() {
            if (_oprocess != null) {
                _oprocess.dehydrate();
                _oprocess = null;
            }
            if (_myRoles != null) {
                _myRoles.clear();
            }
            if (_endpointToMyRoleMap != null) {
                _endpointToMyRoleMap.clear();
            }
            if (_partnerRoles != null) {
                _partnerRoles.clear();
            }
            // Don't clear stuff you can't re-populate
//            if (_myEprs != null) {
//                _myEprs.clear();
//            }
//            if (_partnerChannels != null) {
//                _partnerChannels.clear();
//            }
//            if (_partnerEprs != null) {
//                _partnerEprs.clear();
//            }
            _replacementMap = null;
            _expLangRuntimeRegistry = null;
        }

        private void doHydrate() {
            markused();
            if (__log.isDebugEnabled()) {
                __log.debug("Rehydrating process " + _pconf.getProcessId());
            }
            try {
            	File file = _pconf.getCBPFile();
            	_oprocess = deserializeCompiledProcess(file);
            } catch (Exception e) {
                String errmsg = "The process " + _pid + " is no longer available.";
                __log.error(errmsg, e);
                throw new BpelEngineException(errmsg, e);
            }

            if (_partnerRoles == null) {
                _partnerRoles = new HashMap<OPartnerLink, PartnerLinkPartnerRoleImpl>();
            }
            if (_myRoles == null) {
                _myRoles = new HashMap<OPartnerLink, PartnerLinkMyRoleImpl>();
            }
            if (_endpointToMyRoleMap == null) {
                _endpointToMyRoleMap = new HashMap<PartnerLinkMyRoleImpl, Endpoint>();
            }
            if (_myEprs == null) {
                _myEprs = new HashMap<Endpoint, EndpointReference>();
            }
            if (_partnerChannels == null) {
                _partnerChannels = new HashMap<Endpoint, PartnerRoleChannel>();
            }
            if (_partnerEprs == null) {
                _partnerEprs = new HashMap<Endpoint, EndpointReference>();
            }

            _replacementMap = new ReplacementMapImpl(_oprocess);

            // Create an expression language registry for this process
            _expLangRuntimeRegistry = new ExpressionLanguageRuntimeRegistry();
            registerExprLang(_oprocess);

            setRoles(_oprocess);
            initExternalVariables();

            if (!_hydratedOnce) {
                for (PartnerLinkPartnerRoleImpl prole : _partnerRoles.values()) {
                    // Null for initializePartnerRole = false
                    if (prole._initialPartner != null) {
                        PartnerRoleChannel channel = _engine._contexts.bindingContext.createPartnerRoleChannel(_pid,
                                prole._plinkDef.getPartnerRolePortType(), prole._initialPartner);
                        prole._channel = channel;
                        _partnerChannels.put(prole._initialPartner, prole._channel);
                        EndpointReference epr = channel.getInitialEndpointReference();
                        if (epr != null) {
                            prole._initialEPR = epr;
                            _partnerEprs.put(prole._initialPartner, epr);
                        }
                        if (__log.isDebugEnabled()) {
                            __log.debug("Activated " + _pid + " partnerrole " + prole.getPartnerLinkName() + ": EPR is "
                                    + prole._initialEPR);
                        }
                              
                    }
                }
                _engine.setProcessSize(_pid, true);
                _hydratedOnce = true;
            }

            for (PartnerLinkMyRoleImpl myrole : _myRoles.values()) {
                myrole._initialEPR = _myEprs.get(myrole._endpoint);
            }

            for (PartnerLinkPartnerRoleImpl prole : _partnerRoles.values()) {
                prole._channel = _partnerChannels.get(prole._initialPartner);
                if (_partnerEprs.get(prole._initialPartner) != null) {
                    prole._initialEPR = _partnerEprs.get(prole._initialPartner);
                }
            }

            /*
             * If necessary, create an object in the data store to represent the process. We'll re-use an existing object if it already
             * exists and matches the GUID.
             */
            if (isInMemory()) {
                createProcessDAO(_engine._contexts.inMemDao.getConnection(), _pid, _pconf.getVersion(), _oprocess);
            } else if (_engine._contexts.scheduler.isTransacted()) {
                // If we have a transaction, we do this in the current transaction
                if(__log.isDebugEnabled()) __log.debug("Creating new process DAO for " + _pid + " (guid=" + _oprocess.getGuid() + ")...");
                createProcessDAO(_engine._contexts.dao.getConnection(), _pid, _pconf.getVersion(), _oprocess);
                if(__log.isDebugEnabled()) __log.debug("Created new process DAO for " + _pid + " (guid=" + _oprocess.getGuid() + ").");
            } else {
                try {
                    _engine._contexts.scheduler.execTransaction(new Callable<Object>() {
                        public Object call() throws Exception {
                            bounceProcessDAOInDB(_engine._contexts.dao.getConnection(), _pid, _pconf.getVersion(), _oprocess);
                            return null;
                        }
                    });
                } catch( RuntimeException re ) {
                    throw re;
                } catch(Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private void bounceProcessDAOInMemory(BpelDAOConnection conn, final QName pid, final long version, final OProcess oprocess) {
        if(__log.isInfoEnabled()) __log.info("Creating new process DAO[mem] for " + pid + " (guid=" + oprocess.getGuid() + ").");
        createProcessDAO(conn, pid, version, oprocess);
    }

    private void bounceProcessDAOInDB(final BpelDAOConnection conn, final QName pid, final long version, final OProcess oprocess) {
        if(__log.isDebugEnabled()) __log.debug("Creating new process DAO for " + pid + " (guid=" + oprocess.getGuid() + ")...");
        createProcessDAO(conn, pid, version, oprocess);
        if(__log.isDebugEnabled()) __log.debug("Created new process DAO for " + pid + " (guid=" + oprocess.getGuid() + ").");
    }

    public int getInstanceInUseCount() {
        return hintIsHydrated() ? _hydrationLatch.getDepth(1) : 0;
    }

    private void createProcessDAO(BpelDAOConnection conn, final QName pid, final long version, final OProcess oprocess) {
        try {
            boolean create = true;
            ProcessDAO old = conn.getProcess(pid);
            if (old != null) {
                if (__log.isDebugEnabled()) {
                    __log.debug("Found ProcessDAO for " + pid + " with GUID " + old.getGuid());
                }
                if (oprocess.getGuid() == null) {
                    // No guid, old version assume its good
                    create = false;
                } else {
                    if (old.getGuid().equals(oprocess.getGuid())) {
                        // Guids match, no need to create
                        create = false;
                    }
                }
            }

            if (create) {
                if(__log.isDebugEnabled()) __log.debug("Creating process DAO for " + pid + " (guid=" + oprocess.getGuid() + ")");

                ProcessDAO newDao = conn.createProcess(pid, oprocess.getQName(), oprocess.getGuid(), (int) version);
                for (String correlator : oprocess.getCorrelators()) {
                    newDao.addCorrelator(correlator);
                }
                if(__log.isDebugEnabled()) __log.debug("Created new process DAO for " + pid + " (guid=" + oprocess.getGuid() + ")");
            }
        } catch (BpelEngineException ex) {
            throw ex;
        } catch (Exception dce) {
            __log.error("DbError", dce);
            throw new BpelEngineException("DbError", dce);
        }
    }

    private void registerExprLang(OProcess oprocess) {
        for (OExpressionLanguage elang : oprocess.getExpressionLanguages()) {
            try {
                _expLangRuntimeRegistry.registerRuntime(elang);
            } catch (ConfigurationException e) {
                String msg = __msgs.msgExpLangRegistrationError(elang.getExpressionLanguageUri(), elang.getProperties());
                __log.error(msg, e);
                throw new BpelEngineException(msg, e);
            }
        }
    }

    public boolean isCleanupCategoryEnabled(boolean instanceSucceeded, CLEANUP_CATEGORY category) {
        return _pconf.isCleanupCategoryEnabled(instanceSucceeded, category);
    }

    public Set<CLEANUP_CATEGORY> getCleanupCategories(boolean instanceSucceeded) {
        return _pconf.getCleanupCategories(instanceSucceeded);
    }

    public Node getProcessProperty(QName propertyName) {
        Node value = getEngine()._contexts.customProcessProperties.getProperty(propertyName);
        if (value != null) {
            return value;
        }
        Map<QName, Node> properties = _pconf.getProcessProperties();
        if (properties != null) {
            return properties.get(propertyName);
        }
        return null;
    }

    public ProcessConf getConf() {
        return _pconf;
    }

    public boolean hasActiveInstances() {
        try {
            _hydrationLatch.latch(1);
            if (isInMemory() || _engine._contexts.scheduler.isTransacted()) {
                return hasActiveInstances(getProcessDAO());
            } else {
                // If we do not have a transaction we need to create one.
                try {
                    return (Boolean) _engine._contexts.scheduler.execTransaction(new Callable<Object>() {
                        public Object call() throws Exception {
                            return hasActiveInstances(getProcessDAO());
                        }
                    });
                } catch (Exception e) {
                    String errmsg = "DbError";
                    __log.error(errmsg, e);
                    return false;
                }
            }
        } finally {
            _hydrationLatch.release(1);
        }
    }

    private boolean hasActiveInstances(ProcessDAO processDAO) {
        // Select count of instances instead of all active instances
        // Collection<ProcessInstanceDAO> activeInstances = processDAO.getActiveInstances();
        // return (activeInstances != null && activeInstances.size() > 0);
        return processDAO.getNumInstances() > 0;
    }

    public void registerMessageExchangeInterceptor(MessageExchangeInterceptor interceptor) {
        _mexInterceptors.add(interceptor);
    }

    public void unregisterMessageExchangeInterceptor(MessageExchangeInterceptor interceptor) {
        _mexInterceptors.remove(interceptor);
    }

    public long sizeOf() {
        // try to get actual size from sizing agent, if enabled
        long footprint = SizingAgent.deepSizeOf(this);
        // if unsuccessful, estimate size (this is a inaccurate guess)
        if (footprint == 0) {
            footprint = getEstimatedHydratedSize();
        }
        // add the sizes of all the services this process provides
        for (EndpointReference myEpr : _myEprs.values()) {
            footprint += _engine._contexts.bindingContext.calculateSizeofService(myEpr);
        }
        // return the total footprint
        return footprint;
    }

    public String getProcessProperty(QName property, String defaultValue) {
        Text text = (Text) getProcessProperty(property);
        if (text == null) {
            return defaultValue;
        }
        String value = text.getWholeText();
        return (value == null) ?  defaultValue : value;
    }

    public boolean isHydrationLazy() {
        return Boolean.valueOf(getProcessProperty(PROP_LAZY_HYDRATE, "true"));
    }

    public boolean isHydrationLazySet() {
        return getProcessProperty(PROP_LAZY_HYDRATE) != null;
    }

    public int getInstanceMaximumCount() {
        return Integer.valueOf(getProcessProperty(PROP_MAX_INSTANCES, Integer.toString(_engine.getInstanceThrottledMaximumCount())));
    }

    public long getEstimatedHydratedSize() {
        return _pconf.getCBPFileSize() *
                    PROCESS_MEMORY_TO_SERIALIZED_SIZE_RATIO;
    }

    public long getTimeout(OPartnerLink partnerLink, boolean p2p) {
        // OPartnerLink, PartnerLinkPartnerRoleImpl
        final PartnerLinkPartnerRoleImpl linkPartnerRole = _partnerRoles.get(partnerLink);
        long timeout = Properties.DEFAULT_MEX_TIMEOUT;
        if (linkPartnerRole._initialEPR != null) {
            String property = null;
            String value = null;
            Map<String, String> props = _pconf.getEndpointProperties(linkPartnerRole._initialEPR);
            if (p2p) {
                property = Properties.PROP_P2P_MEX_TIMEOUT;
                value = props.get(property);
            }
            if (value == null) {
                property = Properties.PROP_MEX_TIMEOUT;
                value = props.get(property);
            }
            if (value != null) {
                try {
                    timeout = Long.parseLong(value);
                } catch (NumberFormatException e) {
                    if (__log.isWarnEnabled())
                        __log.warn("Mal-formatted Property: [" + property + "=" + value + "] Default value (" + timeout + ") will be used");
                }
            }
        }

        return timeout;
    }

    public long getVersion() {
        return Long.parseLong(_pid.getLocalPart().substring(_pid.getLocalPart().lastIndexOf('-') + 1));
    }
    
    public void doAsyncReply(MyRoleMessageExchangeImpl m, BpelRuntimeContext context) {
        MessageExchangeDAO mex = m.getDAO();
        PartnerRoleMessageExchange pmex = null;

        if (mex.getPipedMessageExchangeId() != null) {
            pmex = (PartnerRoleMessageExchange) getEngine().getMessageExchange(mex.getPipedMessageExchangeId());
        }

        if (pmex != null) {
            if (BpelProcess.__log.isDebugEnabled()) {
                __log.debug("Replying to a p2p mex, myrole " + m + " - partnerole " + pmex);
            }

            if (pmex.getStatus() == Status.ASYNC || pmex.getStatus() == Status.REQUEST) {
                try {
                    switch (m.getStatus()) {
                        case FAILURE:
                            // We can't seem to get the failure out of the myrole mex?
                            pmex.replyWithFailure(MessageExchange.FailureType.OTHER, "operation failed", null);
                            break;
                        case FAULT:
                            Fault fault = pmex.getOperation().getFault(m.getFault().getLocalPart());
                            if (fault == null) {
                                __log.error("process " + this + " instance " + (context != null ? context.getPid() : null) + " thrown unmapped fault in p2p communication " + m.getFault() + " " + m.getFaultExplanation() + " - converted to failure");
                                pmex.replyWithFailure(MessageExchange.FailureType.OTHER, "process thrown unmapped fault in p2p communication " + m.getFault() + " " + m.getFaultExplanation() + " - converted to failure", m.getFaultResponse().getMessage());
                            } else {
                                Message faultRes = pmex.createMessage(pmex.getOperation().getFault(m.getFault().getLocalPart())
                                        .getMessage().getQName());
                                faultRes.setMessage(m.getResponse().getMessage());
                                pmex.replyWithFault(m.getFault(), faultRes);
                            }
                            break;
                        case RESPONSE:
                            Message response = pmex.createMessage(pmex.getOperation().getOutput().getMessage().getQName());
                            response.setMessage(m.getResponse().getMessage());
                            pmex.reply(response);
                            break;
                        default:
                            __log.warn("Unexpected state: " + m.getStatus());
                            break;
                    }
                } finally {
                    mex.release(this.isCleanupCategoryEnabled(m.getStatus() == MessageExchange.Status.RESPONSE, CLEANUP_CATEGORY.MESSAGES));
                }
            } else {
                __log.warn("Can't send response to a p2p mex: " + mex + " partner mex: " + pmex);
            }
        } else {
            if (context != null) context.checkInvokeExternalPermission();
            this._engine._contexts.mexContext.onAsyncReply(m);
            //mex.release(_bpelProcess.isCleanupCategoryEnabled(m.getStatus() == MessageExchange.Status.RESPONSE, CLEANUP_CATEGORY.MESSAGES));
        }
    }
    
}