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

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.transaction.Transaction;

import org.apache.ofbiz.base.config.GenericConfigException;
import org.apache.ofbiz.base.util.Debug;
import org.apache.ofbiz.base.util.GeneralRuntimeException;
import org.apache.ofbiz.base.util.UtilMisc;
import org.apache.ofbiz.base.util.UtilProperties;
import org.apache.ofbiz.base.util.UtilTimer;
import org.apache.ofbiz.base.util.UtilValidate;
import org.apache.ofbiz.entity.Delegator;
import org.apache.ofbiz.entity.DelegatorFactory;
import org.apache.ofbiz.entity.GenericDelegator;
import org.apache.ofbiz.entity.GenericEntityException;
import org.apache.ofbiz.entity.GenericValue;
import org.apache.ofbiz.entity.transaction.DebugXaResource;
import org.apache.ofbiz.entity.transaction.GenericTransactionException;
import org.apache.ofbiz.entity.transaction.TransactionUtil;
import org.apache.ofbiz.security.Security;
import org.apache.ofbiz.security.SecurityConfigurationException;
import org.apache.ofbiz.security.SecurityFactory;
import org.apache.ofbiz.service.config.ServiceConfigUtil;
import org.apache.ofbiz.service.config.model.StartupService;
import org.apache.ofbiz.service.eca.ServiceEcaRule;
import org.apache.ofbiz.service.eca.ServiceEcaUtil;
import org.apache.ofbiz.service.engine.GenericEngine;
import org.apache.ofbiz.service.engine.GenericEngineFactory;
import org.apache.ofbiz.service.group.ServiceGroupReader;
import org.apache.ofbiz.service.jms.JmsListenerFactory;
import org.apache.ofbiz.service.job.JobManager;
import org.apache.ofbiz.service.job.JobManagerException;
import org.apache.ofbiz.service.semaphore.ServiceSemaphore;

import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap;

/**
 * The global service dispatcher. This is the "engine" part of the
 * Service Engine.
 */
public class ServiceDispatcher {

    public static final String module = ServiceDispatcher.class.getName();
    public static final int lruLogSize = 200;
    public static final int LOCK_RETRIES = 3;

    protected static final Map<RunningService, ServiceDispatcher> runLog = new ConcurrentLinkedHashMap.Builder<RunningService, ServiceDispatcher>().maximumWeightedCapacity(lruLogSize).build();
    protected static ConcurrentHashMap<String, ServiceDispatcher> dispatchers = new ConcurrentHashMap<String, ServiceDispatcher>();
    // FIXME: These fields are not thread-safe. They are modified by EntityDataLoadContainer.
    // We need a better design - like have this class query EntityDataLoadContainer if data is being loaded.
    protected static boolean enableJM = true;
    protected static boolean enableJMS = true;
    protected static boolean enableSvcs = true;

    protected Delegator delegator = null;
    protected GenericEngineFactory factory = null;
    protected Security security = null;
    protected Map<String, DispatchContext> localContext = new HashMap<String, DispatchContext>();
    protected Map<String, List<GenericServiceCallback>> callbacks = new HashMap<String, List<GenericServiceCallback>>();
    protected JobManager jm = null;
    protected JmsListenerFactory jlf = null;

    protected ServiceDispatcher(Delegator delegator, boolean enableJM, boolean enableJMS) {
        factory = new GenericEngineFactory(this);
        ServiceGroupReader.readConfig();
        ServiceEcaUtil.readConfig();

        this.delegator = delegator;

        if (delegator != null) {
            try {
                this.security = SecurityFactory.getInstance(delegator);
            } catch (SecurityConfigurationException e) {
                Debug.logError(e, "[ServiceDispatcher.init] : No instance of security implementation found.", module);
            }
        }

        // clean up the service semaphores of same instance
        if (delegator != null) {
            try {
                int rn = delegator.removeByAnd("ServiceSemaphore", "lockedByInstanceId", JobManager.instanceId);
                if (rn > 0) {
                    Debug.logInfo("[ServiceDispatcher.init] : Clean up " + rn + " service semaphors." , module);
                }
            } catch (GenericEntityException e) {
                Debug.logError(e, module);
            }
        }
        
        // job manager needs to always be running, but the poller thread does not
        try {
            Delegator origDelegator = this.delegator;
            if (!this.delegator.getOriginalDelegatorName().equals(this.delegator.getDelegatorName())) {
                origDelegator = DelegatorFactory.getDelegator(this.delegator.getOriginalDelegatorName());
            }
            this.jm = JobManager.getInstance(origDelegator, enableJM);
        } catch (GeneralRuntimeException e) {
            Debug.logWarning(e.getMessage(), module);
        }

        // make sure we haven't disabled these features from running
        if (enableJMS) {
            this.jlf = JmsListenerFactory.getInstance(delegator);
        }
    }

    protected ServiceDispatcher(Delegator delegator) {
        this(delegator, enableJM, enableJMS);
    }

    /**
     * Returns a pre-registered instance of the ServiceDispatcher associated with this delegator.
     * @param name the name of the DispatchContext
     * @param delegator the local delegator
     * @return A reference to the LocalDispatcher associated with the DispatchContext
     */
    public static LocalDispatcher getLocalDispatcher(String name, Delegator delegator) {
        // get the ServiceDispatcher associated to the delegator (if not found in the cache, it will be created and added to the cache)
        ServiceDispatcher sd = getInstance(delegator);
        // if a DispatchContext has already been already registered as "name" then return the LocalDispatcher associated with it
        if (sd.containsContext(name)) {
            return sd.getLocalDispatcher(name);
        }
        // otherwise return null
        return null;
    }

    /**
     * Returns an instance of the ServiceDispatcher associated with this delegator.
     * @param delegator the local delegator
     * @return A reference to this global ServiceDispatcher
     */
    public static ServiceDispatcher getInstance(Delegator delegator) {
        ServiceDispatcher sd;
        String dispatcherKey = delegator != null ? delegator.getDelegatorName() : "null";
        sd = dispatchers.get(dispatcherKey);
        if (sd == null) {
            if (Debug.verboseOn())
                Debug.logVerbose("[ServiceDispatcher.getInstance] : No instance found (" + dispatcherKey + ").", module);
            sd = new ServiceDispatcher(delegator);
            ServiceDispatcher cachedDispatcher = dispatchers.putIfAbsent(dispatcherKey, sd);
            if (cachedDispatcher == null) {
                // if the cachedDispatcher is null, then it means that
                // the new dispatcher created by this thread was successfully added to the cache
                // only in this case, the thread runs runStartupServices
                sd.runStartupServices();
                cachedDispatcher = sd;
            }
            sd = cachedDispatcher;
        }
        return sd;
    }

    /**
     * Registers the loader with this ServiceDispatcher
     * @param context the context of the local dispatcher
     */
    public void register(DispatchContext context) {
        if (Debug.infoOn()) Debug.logInfo("Registering dispatcher: " + context.getName(), module);
        this.localContext.put(context.getName(), context);
    }
    /**
     * De-Registers the loader with this ServiceDispatcher
     * @param local the LocalDispatcher to de-register
     */
    public void deregister(LocalDispatcher local) {
        if (Debug.infoOn()) Debug.logInfo("De-Registering dispatcher: " + local.getName(), module);
        localContext.remove(local.getName());
        if (localContext.size() == 0) {
            try {
                 this.shutdown();
             } catch (GenericServiceException e) {
                 Debug.logError(e, "Trouble shutting down ServiceDispatcher!", module);
             }
        }
    }

    public synchronized void registerCallback(String serviceName, GenericServiceCallback cb) {
        List<GenericServiceCallback> callBackList = callbacks.get(serviceName);
        if (callBackList == null) {
            callBackList = new LinkedList<GenericServiceCallback>();
        }
        callBackList.add(cb);
        callbacks.put(serviceName, callBackList);
    }

    public List<GenericServiceCallback> getCallbacks(String serviceName) {
        return callbacks.get(serviceName);
    }

    /**
     * Run the service synchronously and return the result.
     * @param localName Name of the context to use.
     * @param service Service model object.
     * @param context Map of name, value pairs composing the context.
     * @return Map of name, value pairs composing the result.
     * @throws ServiceAuthException
     * @throws ServiceValidationException
     * @throws GenericServiceException
     */
    public Map<String, Object> runSync(String localName, ModelService service, Map<String, ? extends Object> context) throws ServiceAuthException, ServiceValidationException, GenericServiceException {
        return runSync(localName, service, context, true);
    }

    /**
     * Run the service synchronously and IGNORE the result.
     * @param localName Name of the context to use.
     * @param service Service model object.
     * @param context Map of name, value pairs composing the context.
     * @throws ServiceAuthException
     * @throws ServiceValidationException
     * @throws GenericServiceException
     */
    public void runSyncIgnore(String localName, ModelService service, Map<String, ? extends Object> context) throws ServiceAuthException, ServiceValidationException, GenericServiceException {
        runSync(localName, service, context, false);
    }

    /**
     * Run the service synchronously and return the result.
     * @param localName Name of the context to use.
     * @param modelService Service model object.
     * @param params Map of name, value pairs composing the parameters.
     * @param validateOut Validate OUT parameters
     * @return Map of name, value pairs composing the result.
     * @throws ServiceAuthException
     * @throws ServiceValidationException
     * @throws GenericServiceException
     */
    public Map<String, Object> runSync(String localName, ModelService modelService, Map<String, ? extends Object> params, boolean validateOut) throws ServiceAuthException, ServiceValidationException, GenericServiceException {
        long serviceStartTime = System.currentTimeMillis();
        Map<String, Object> result = new HashMap<String, Object>();
        ServiceSemaphore lock = null;
        Map<String, List<ServiceEcaRule>> eventMap = null;
        Map<String, Object> ecaContext = null;
        RunningService rs = null;
        DispatchContext ctx = localContext.get(localName);
        GenericEngine engine = null;
        Transaction parentTransaction = null;
        boolean isFailure = false;
        boolean isError = false;
        boolean beganTrans = false;
        try {
            // check for semaphore and acquire a lock
            if ("wait".equals(modelService.semaphore) || "fail".equals(modelService.semaphore)) {
                lock = new ServiceSemaphore(delegator, modelService);
                lock.acquire();
            }

            if (Debug.verboseOn() || modelService.debug) {
                Debug.logVerbose("[ServiceDispatcher.runSync] : invoking service " + modelService.name + " [" + modelService.location +
                    "/" + modelService.invoke + "] (" + modelService.engineName + ")", module);
            }

            Map<String, Object> context = new HashMap<String, Object>();
            if (params != null) {
                context.putAll(params);
            }
            // check the locale
            Locale locale = this.checkLocale(context);

            // set up the running service log
            rs = this.logService(localName, modelService, GenericEngine.SYNC_MODE);

            // get eventMap once for all calls for speed, don't do event calls if it is null
            eventMap = ServiceEcaUtil.getServiceEventMap(modelService.name);
            engine = this.getGenericEngine(modelService.engineName);


            // set IN attributes with default-value as applicable
            modelService.updateDefaultValues(context, ModelService.IN_PARAM);
            //Debug.logInfo("=========================== " + modelService.name + " 1 tx status =" + TransactionUtil.getStatusString() + ", modelService.requireNewTransaction=" + modelService.requireNewTransaction + ", modelService.useTransaction=" + modelService.useTransaction + ", TransactionUtil.isTransactionInPlace()=" + TransactionUtil.isTransactionInPlace(), module);
            if (modelService.useTransaction) {
                if (TransactionUtil.isTransactionInPlace()) {
                    // if a new transaction is needed, do it here; if not do nothing, just use current tx
                    if (modelService.requireNewTransaction) {
                        parentTransaction = TransactionUtil.suspend();
                        if (TransactionUtil.isTransactionInPlace()) {
                            throw new GenericTransactionException("In service " + modelService.name + " transaction is still in place after suspend, status is " + TransactionUtil.getStatusString());
                        }
                        // now start a new transaction
                        beganTrans = TransactionUtil.begin(modelService.transactionTimeout);
                    }
                } else {
                    beganTrans = TransactionUtil.begin(modelService.transactionTimeout);
                }
                // enlist for XAResource debugging
                if (beganTrans && TransactionUtil.debugResources()) {
                    DebugXaResource dxa = new DebugXaResource(modelService.name);
                    try {
                        dxa.enlist();
                    } catch (Exception e) {
                        Debug.logError(e, module);
                    }
                }
            }

            try {
                int lockRetriesRemaining = LOCK_RETRIES;
                boolean needsLockRetry = false;

                do {
                    // Ensure this is reset to false on each pass
                    needsLockRetry = false;

                    lockRetriesRemaining--;

                    // NOTE: general pattern here is to do everything up to the main service call, and retry it all if
                    //needed because those will be part of the same transaction and have been rolled back
                    // TODO: if there is an ECA called async or in a new transaction it won't get rolled back
                    //but will be called again, which means the service may complete multiple times! that would be for
                    //pre-invoke and earlier events only of course


                    // setup global transaction ECA listeners to execute later
                    if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "global-rollback", ctx, context, result, isError, isFailure);
                    if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "global-commit", ctx, context, result, isError, isFailure);

                    // pre-auth ECA
                    if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "auth", ctx, context, result, isError, isFailure);

                    // check for pre-auth failure/errors
                    isFailure = ServiceUtil.isFailure(result);
                    isError = ServiceUtil.isError(result);

                    //Debug.logInfo("After [" + modelService.name + "] pre-auth ECA, before auth; isFailure=" + isFailure + ", isError=" + isError, module);

                    context = checkAuth(localName, context, modelService);
                    GenericValue userLogin = (GenericValue) context.get("userLogin");

                    if (modelService.auth && userLogin == null) {
                        throw new ServiceAuthException("User authorization is required for this service: " + modelService.name + modelService.debugInfo());
                    }

                    // now that we have authed, if there is a userLogin, set the EE userIdentifier
                    if (userLogin != null && userLogin.getString("userLoginId") != null) {
                        GenericDelegator.pushUserIdentifier(userLogin.getString("userLoginId"));
                    }

                    // pre-validate ECA
                    if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "in-validate", ctx, context, result, isError, isFailure);

                    // check for pre-validate failure/errors
                    isFailure = ServiceUtil.isFailure(result);
                    isError = ServiceUtil.isError(result);

                    //Debug.logInfo("After [" + modelService.name + "] pre-in-validate ECA, before in-validate; isFailure=" + isFailure + ", isError=" + isError, module);

                    // validate the context
                    if (modelService.validate && !isError && !isFailure) {
                        try {
                            modelService.validate(context, ModelService.IN_PARAM, locale);
                        } catch (ServiceValidationException e) {
                            Debug.logError(e, "Incoming context (in runSync : " + modelService.name + ") does not match expected requirements", module);
                            throw e;
                        }
                    }

                    // pre-invoke ECA
                    if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "invoke", ctx, context, result, isError, isFailure);

                    // check for pre-invoke failure/errors
                    isFailure = ServiceUtil.isFailure(result);
                    isError = ServiceUtil.isError(result);

                    //Debug.logInfo("After [" + modelService.name + "] pre-invoke ECA, before invoke; isFailure=" + isFailure + ", isError=" + isError, module);

                    // ===== invoke the service =====
                    if (!isError && !isFailure) {
                        Map<String, Object> invokeResult = null;
                        invokeResult = engine.runSync(localName, modelService, context);
                        engine.sendCallbacks(modelService, context, invokeResult, GenericEngine.SYNC_MODE);
                        if (invokeResult != null) {
                            result.putAll(invokeResult);
                        } else {
                            Debug.logWarning("Service (in runSync : " + modelService.name + ") returns null result", module);
                        }
                    }

                    // re-check the errors/failures
                    isFailure = ServiceUtil.isFailure(result);
                    isError = ServiceUtil.isError(result);

                    //Debug.logInfo("After [" + modelService.name + "] invoke; isFailure=" + isFailure + ", isError=" + isError, module);

                    if (beganTrans) {
                        // crazy stuff here: see if there was a deadlock or other such error and if so retry... which we can ONLY do if we own the transaction!

                        String errMsg = ServiceUtil.getErrorMessage(result);

                        // look for the string DEADLOCK in an upper-cased error message; tested on: Derby, MySQL
                        // - Derby 10.2.2.0 deadlock string: "A lock could not be obtained due to a deadlock"
                        // - MySQL ? deadlock string: "Deadlock found when trying to get lock; try restarting transaction"
                        // - Postgres ? deadlock string: TODO
                        // - Other ? deadlock string: TODO
                        // TODO need testing in other databases because they all return different error messages for this!

                        // NOTE DEJ20070908 are there other things we need to check? I don't think so because these will
                        //be Entity Engine errors that will be caught and come back in an error message... IFF the
                        //service is written to not ignore it of course!
                        if (errMsg != null && errMsg.toUpperCase().indexOf("DEADLOCK") >= 0) {
                            // it's a deadlock! retry...
                            String retryMsg = "RETRYING SERVICE [" + modelService.name + "]: Deadlock error found in message [" + errMsg + "]; retry [" + (LOCK_RETRIES - lockRetriesRemaining) + "] of [" + LOCK_RETRIES + "]";

                            // make sure the old transaction is rolled back, and then start a new one

                            // if there is an exception in these things, let the big overall thing handle it
                            TransactionUtil.rollback(beganTrans, retryMsg, null);

                            beganTrans = TransactionUtil.begin(modelService.transactionTimeout);
                            // enlist for XAResource debugging
                            if (beganTrans && TransactionUtil.debugResources()) {
                                DebugXaResource dxa = new DebugXaResource(modelService.name);
                                try {
                                    dxa.enlist();
                                } catch (Exception e) {
                                    Debug.logError(e, module);
                                }
                            }

                            if (!beganTrans) {
                                // just log and let things roll through, will be considered an error and ECAs, etc will run according to that
                                Debug.logError("After rollback attempt for lock retry did not begin a new transaction!", module);
                            } else {
                                // deadlocks can be resolved by retring immediately as conflicting operations in the other thread will have cleared
                                needsLockRetry = true;

                                // reset state variables
                                result = new HashMap<String, Object>();
                                isFailure = false;
                                isError = false;

                                Debug.logWarning(retryMsg, module);
                            }

                            // look for lock wait timeout error, retry in a different way by running after the parent transaction finishes, ie attach to parent tx
                            // - Derby 10.2.2.0 lock wait timeout string: "A lock could not be obtained within the time requested"
                            // - MySQL ? lock wait timeout string: "Lock wait timeout exceeded; try restarting transaction"
                            if (errMsg != null && (errMsg.indexOf("A lock could not be obtained within the time requested") >= 0 ||
                                    errMsg.indexOf("Lock wait timeout exceeded") >= 0)) {
                                // TODO: add to run after parent tx
                            }
                        }
                    }
                } while (needsLockRetry && lockRetriesRemaining > 0);

                // create a new context with the results to pass to ECA services; necessary because caller may reuse this context
                ecaContext = new HashMap<String, Object>();
                ecaContext.putAll(context);
                // copy all results: don't worry parameters that aren't allowed won't be passed to the ECA services
                ecaContext.putAll(result);

                // setup default OUT values
                modelService.updateDefaultValues(context, ModelService.OUT_PARAM);

                // validate the result
                if (modelService.validate && validateOut) {
                    // pre-out-validate ECA
                    if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "out-validate", ctx, ecaContext, result, isError, isFailure);
                    try {
                        modelService.validate(result, ModelService.OUT_PARAM, locale);
                    } catch (ServiceValidationException e) {
                        throw new GenericServiceException("Outgoing result (in runSync : " + modelService.name + ") does not match expected requirements", e);
                    }
                }

                // pre-commit ECA
                if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "commit", ctx, ecaContext, result, isError, isFailure);

                // check for pre-commit failure/errors
                isFailure = ServiceUtil.isFailure(result);
                isError = ServiceUtil.isError(result);

                // global-commit-post-run ECA, like global-commit but gets the context after the service is run
                if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "global-commit-post-run", ctx, ecaContext, result, isError, isFailure);

                // check for failure and log on info level; this is used for debugging
                if (isFailure) {
                    Debug.logWarning("Service Failure [" + modelService.name + "]: " + ServiceUtil.getErrorMessage(result), module);
                }
            } catch (Throwable t) {
                if (Debug.timingOn()) {
                    UtilTimer.closeTimer(localName + " / " + modelService.name, "Sync service failed...", module);
                }
                String errMsg = "Service [" + modelService.name + "] threw an unexpected exception/error";
                engine.sendCallbacks(modelService, context, t, GenericEngine.SYNC_MODE);
                try {
                    TransactionUtil.rollback(beganTrans, errMsg, t);
                } catch (GenericTransactionException te) {
                    Debug.logError(te, "Cannot rollback transaction", module);
                }
                rs.setEndStamp();
                if (t instanceof ServiceAuthException) {
                    throw (ServiceAuthException) t;
                } else if (t instanceof ServiceValidationException) {
                    throw (ServiceValidationException) t;
                } else if (t instanceof GenericServiceException) {
                    throw (GenericServiceException) t;
                } else {
                    throw new GenericServiceException("Service [" + modelService.name + "] Failed" + modelService.debugInfo() , t);
                }
            } finally {
                // if there was an error, rollback transaction, otherwise commit
                if (isError) {
                    String errMsg = "Error in Service [" + modelService.name + "]: " + ServiceUtil.getErrorMessage(result);
                    Debug.logError(errMsg, module);

                    // rollback the transaction
                    try {
                        TransactionUtil.rollback(beganTrans, errMsg, null);
                    } catch (GenericTransactionException e) {
                        Debug.logError(e, "Could not rollback transaction: " + e.toString(), module);
                    }
                } else {
                    // commit the transaction
                    try {
                        TransactionUtil.commit(beganTrans);
                    } catch (GenericTransactionException e) {
                        GenericDelegator.popUserIdentifier();
                        String errMsg = "Could not commit transaction for service [" + modelService.name + "] call";
                        Debug.logError(e, errMsg, module);
                        if (e.getMessage() != null) {
                            errMsg = errMsg + ": " + e.getMessage();
                        }
                        throw new GenericServiceException(errMsg);
                    }
                }

                // call notifications -- event is determined from the result (success, error, fail)
                modelService.evalNotifications(this.getLocalContext(localName), context, result);

                // clear out the EE userIdentifier
                GenericDelegator.popUserIdentifier();
            }
        } catch (GenericTransactionException te) {
            Debug.logError(te, "Problems with the transaction", module);
            throw new GenericServiceException("Problems with the transaction.", te.getNested());
        } finally {
            if (lock != null) {
                // release the semaphore lock
                try {
                    lock.release();
                } catch (GenericServiceException e) {
                    Debug.logWarning(e, "Exception thrown while unlocking semaphore: ", module);
                }
            }

            // resume the parent transaction
            if (parentTransaction != null) {
                try {
                    TransactionUtil.resume(parentTransaction);
                } catch (GenericTransactionException ite) {
                    Debug.logWarning(ite, "Transaction error, not resumed", module);
                    throw new GenericServiceException("Resume transaction exception, see logs");
                }
            }
        }

        // pre-return ECA
        if (eventMap != null) ServiceEcaUtil.evalRules(modelService.name, eventMap, "return", ctx, ecaContext, result, isError, isFailure);

        rs.setEndStamp();

        long timeToRun = System.currentTimeMillis() - serviceStartTime;
        long showServiceDurationThreshold = UtilProperties.getPropertyAsLong("service", "showServiceDurationThreshold", 0);
        long showSlowServiceThreshold = UtilProperties.getPropertyAsLong("service", "showSlowServiceThreshold", 1000);
                
        if (Debug.timingOn() && timeToRun > showServiceDurationThreshold) {
            Debug.logTiming("Sync service [" + localName + "/" + modelService.name + "] finished in [" + timeToRun + "] milliseconds", module);
        } else if (Debug.infoOn() && timeToRun > showSlowServiceThreshold) {
            Debug.logTiming("Slow sync service execution detected: service [" + localName + "/" + modelService.name + "] finished in [" + timeToRun + "] milliseconds", module);
        }
        if ((Debug.verboseOn() || modelService.debug) && timeToRun > 50 && !modelService.hideResultInLog) {
            // Sanity check - some service results can be multiple MB in size. Limit message size to 10K.
            String resultStr = result.toString();
            if (resultStr.length() > 10240) {
                resultStr = resultStr.substring(0, 10226) + "...[truncated]";
            }
            Debug.logVerbose("Sync service [" + localName + "/" + modelService.name + "] finished with response [" + resultStr + "]", module);
        }
        if (modelService.metrics != null) {
            modelService.metrics.recordServiceRate(1, timeToRun);
        }
        return result;
    }

    /**
     * Run the service asynchronously, passing an instance of GenericRequester that will receive the result.
     * @param localName Name of the context to use.
     * @param service Service model object.
     * @param params Map of name, value pairs composing the parameters.
     * @param requester Object implementing GenericRequester interface which will receive the result.
     * @param persist True for store/run; False for run.
     * @throws ServiceAuthException
     * @throws ServiceValidationException
     * @throws GenericServiceException
     */
    public void runAsync(String localName, ModelService service, Map<String, ? extends Object> params, GenericRequester requester, boolean persist) throws ServiceAuthException, ServiceValidationException, GenericServiceException {
        if (Debug.timingOn()) {
            UtilTimer.timerLog(localName + " / " + service.name, "ASync service started...", module);
        }
        if (Debug.verboseOn() || service.debug) {
            Debug.logVerbose("[ServiceDispatcher.runAsync] : preparing service " + service.name + " [" + service.location + "/" + service.invoke +
                "] (" + service.engineName + ")", module);
        }

        Map<String, Object> context = new HashMap<String, Object>();
        if (params != null) {
            context.putAll(params);
        }
        // setup the result map
        Map<String, Object> result = new HashMap<String, Object>();
        boolean isFailure = false;
        boolean isError = false;

        // set up the running service log
        this.logService(localName, service, GenericEngine.ASYNC_MODE);

        // check the locale
        Locale locale = this.checkLocale(context);

        // setup the engine and context
        DispatchContext ctx = localContext.get(localName);
        GenericEngine engine = this.getGenericEngine(service.engineName);

        // for isolated transactions
        Transaction parentTransaction = null;
        // start the transaction
        boolean beganTrans = false;

        try {
            if (service.useTransaction) {
                if (TransactionUtil.isTransactionInPlace()) {
                    // if a new transaction is needed, do it here; if not do nothing, just use current tx
                    if (service.requireNewTransaction) {
                        parentTransaction = TransactionUtil.suspend();
                        // now start a new transaction
                        beganTrans = TransactionUtil.begin(service.transactionTimeout);
                    }
                } else {
                    beganTrans = TransactionUtil.begin(service.transactionTimeout);
                }
                // enlist for XAResource debugging
                if (beganTrans && TransactionUtil.debugResources()) {
                    DebugXaResource dxa = new DebugXaResource(service.name);
                    try {
                        dxa.enlist();
                    } catch (Exception e) {
                        Debug.logError(e, module);
                    }
                }
            }

            try {
                // get eventMap once for all calls for speed, don't do event calls if it is null
                Map<String, List<ServiceEcaRule>> eventMap = ServiceEcaUtil.getServiceEventMap(service.name);

                // pre-auth ECA
                if (eventMap != null) ServiceEcaUtil.evalRules(service.name, eventMap, "auth", ctx, context, result, isError, isFailure);

                context = checkAuth(localName, context, service);
                Object userLogin = context.get("userLogin");

                if (service.auth && userLogin == null) {
                    throw new ServiceAuthException("User authorization is required for this service: " + service.name + service.debugInfo());
                }

                // pre-validate ECA
                if (eventMap != null) ServiceEcaUtil.evalRules(service.name, eventMap, "in-validate", ctx, context, result, isError, isFailure);

                // check for pre-validate failure/errors
                isFailure = ModelService.RESPOND_FAIL.equals(result.get(ModelService.RESPONSE_MESSAGE));
                isError = ModelService.RESPOND_ERROR.equals(result.get(ModelService.RESPONSE_MESSAGE));

                // validate the context
                if (service.validate && !isError && !isFailure) {
                    try {
                        service.validate(context, ModelService.IN_PARAM, locale);
                    } catch (ServiceValidationException e) {
                        Debug.logError(e, "Incoming service context (in runAsync: " + service.name + ") does not match expected requirements", module);
                        throw e;
                    }
                }

                // run the service
                if (!isError && !isFailure) {
                    if (requester != null) {
                        engine.runAsync(localName, service, context, requester, persist);
                    } else {
                        engine.runAsync(localName, service, context, persist);
                    }
                    engine.sendCallbacks(service, context, GenericEngine.ASYNC_MODE);
                }

                if (Debug.timingOn()) {
                    UtilTimer.closeTimer(localName + " / " + service.name, "ASync service finished...", module);
                }
            } catch (Throwable t) {
                if (Debug.timingOn()) {
                    UtilTimer.closeTimer(localName + " / " + service.name, "ASync service failed...", module);
                }
                String errMsg = "Service [" + service.name + "] threw an unexpected exception/error";
                Debug.logError(t, errMsg, module);
                engine.sendCallbacks(service, context, t, GenericEngine.ASYNC_MODE);
                try {
                    TransactionUtil.rollback(beganTrans, errMsg, t);
                } catch (GenericTransactionException te) {
                    Debug.logError(te, "Cannot rollback transaction", module);
                }
                if (t instanceof ServiceAuthException) {
                    throw (ServiceAuthException) t;
                } else if (t instanceof ServiceValidationException) {
                    throw (ServiceValidationException) t;
                } else if (t instanceof GenericServiceException) {
                    throw (GenericServiceException) t;
                } else {
                    throw new GenericServiceException("Service [" + service.name + "] Failed" + service.debugInfo() , t);
                }
            } finally {
                // always try to commit the transaction since we don't know in this case if its was an error or not
                try {
                    TransactionUtil.commit(beganTrans);
                } catch (GenericTransactionException e) {
                    Debug.logError(e, "Could not commit transaction", module);
                    throw new GenericServiceException("Commit transaction failed");
                }
            }
        } catch (GenericTransactionException se) {
            Debug.logError(se, "Problems with the transaction", module);
            throw new GenericServiceException("Problems with the transaction: " + se.getMessage() + "; See logs for more detail");
        } finally {
            // resume the parent transaction
            if (parentTransaction != null) {
                try {
                    TransactionUtil.resume(parentTransaction);
                } catch (GenericTransactionException ise) {
                    Debug.logError(ise, "Trouble resuming parent transaction", module);
                    throw new GenericServiceException("Resume transaction exception: " + ise.getMessage() + "; See logs for more detail");
                }
            }
        }
    }

    /**
     * Run the service asynchronously and IGNORE the result.
     * @param localName Name of the context to use.
     * @param service Service model object.
     * @param context Map of name, value pairs composing the context.
     * @param persist True for store/run; False for run.
     * @throws ServiceAuthException
     * @throws ServiceValidationException
     * @throws GenericServiceException
     */
    public void runAsync(String localName, ModelService service, Map<String, ? extends Object> context, boolean persist) throws ServiceAuthException, ServiceValidationException, GenericServiceException {
        this.runAsync(localName, service, context, null, persist);
    }

    /**
     * Gets the GenericEngine instance that corresponds to the given name
     * @param engineName Name of the engine
     * @return GenericEngine instance that corresponds to the engineName
     */
    public GenericEngine getGenericEngine(String engineName) throws GenericServiceException {
        return factory.getGenericEngine(engineName);
    }

    /**
     * Gets the JobManager associated with this dispatcher
     * @return JobManager that is associated with this dispatcher
     */
    public JobManager getJobManager() {
        return this.jm;
    }

    /**
     * Gets the JmsListenerFactory which holds the message listeners.
     * @return JmsListenerFactory
     */
    public JmsListenerFactory getJMSListenerFactory() {
        return this.jlf;
    }

    /**
     * Gets the Delegator associated with this dispatcher
     * @return Delegator associated with this dispatcher
     */
    public Delegator getDelegator() {
        return this.delegator;
    }

    /**
     * Gets the Security object associated with this dispatcher
     * @return Security object associated with this dispatcher
     */
    public Security getSecurity() {
        return this.security;
    }

    /**
     * Gets the local context from a name
     * @param name of the context to find.
     */
    public DispatchContext getLocalContext(String name) {
        return localContext.get(name);
    }

    /**
     * Gets the local dispatcher from a name
     * @param name of the LocalDispatcher to find.
     * @return LocalDispatcher matching the loader name
     */
    public LocalDispatcher getLocalDispatcher(String name) {
        return localContext.get(name).getDispatcher();
    }

    /**
     * Test if this dispatcher instance contains the local context.
     * @param name of the local context
     * @return true if the local context is found in this dispatcher.
     */
    public boolean containsContext(String name) {
        return localContext.containsKey(name);
    }

    protected void shutdown() throws GenericServiceException {
        Debug.logImportant("Shutting down the service engine...", module);
        if (jlf != null) {
            // shutdown JMS listeners
            jlf.closeListeners();
        }
    }

    // checks if parameters were passed for authentication
    private Map<String, Object> checkAuth(String localName, Map<String, Object> context, ModelService origService) throws ServiceAuthException, GenericServiceException {
        String service = null;
        try {
            service = ServiceConfigUtil.getServiceEngine().getAuthorization().getServiceName();
        } catch (GenericConfigException e) {
            throw new GenericServiceException(e.getMessage(), e);
        }

        if (service == null) {
            throw new GenericServiceException("No Authentication Service Defined");
        }
        if (service.equals(origService.name)) {
            // manually calling the auth service, don't continue...
            return context;
        }

        if (UtilValidate.isNotEmpty(context.get("login.username"))) {
            // check for a username/password, if there log the user in and make the userLogin object
            String username = (String) context.get("login.username");

            if (UtilValidate.isNotEmpty(context.get("login.password"))) {
                String password = (String) context.get("login.password");

                context.put("userLogin", getLoginObject(service, localName, username, password, (Locale) context.get("locale")));
                context.remove("login.password");
            } else {
                context.put("userLogin", getLoginObject(service, localName, username, null, (Locale) context.get("locale")));
            }
            context.remove("login.username");
        } else {
            // if a userLogin object is there, make sure the given username/password exists in our local database
            GenericValue userLogin = (GenericValue) context.get("userLogin");

            if (userLogin != null) {
                // Because of encrypted passwords we can't just pass in the encrypted version of the password from the data, so we'll do something different and not run the userLogin service...

                //The old way: GenericValue newUserLogin = getLoginObject(service, localName, userLogin.getString("userLoginId"), userLogin.getString("currentPassword"), (Locale) context.get("locale"));
                GenericValue newUserLogin = null;
                try {
                    newUserLogin = this.getDelegator().findOne("UserLogin", true, "userLoginId", userLogin.get("userLoginId"));
                } catch (GenericEntityException e) {
                    Debug.logError(e, "Error looking up service authentication UserLogin: " + e.toString(), module);
                    // leave newUserLogin null, will be handled below
                }

                if (newUserLogin == null) {
                    // uh oh, couldn't validate that one...
                    // we'll have to remove it from the incoming context which will cause an auth error later if auth is required
                    Debug.logInfo("Service auth failed for userLoginId [" + userLogin.get("userLoginId") + "] because UserLogin record not found.", module);
                    context.remove("userLogin");
                } else if (newUserLogin.getString("currentPassword") != null && !newUserLogin.getString("currentPassword").equals(userLogin.getString("currentPassword"))) {
                    // passwords didn't match, remove the userLogin for failed auth
                    Debug.logInfo("Service auth failed for userLoginId [" + userLogin.get("userLoginId") + "] because UserLogin record currentPassword fields did not match; note that the UserLogin object passed into a service may need to have the currentPassword encrypted.", module);
                    context.remove("userLogin");
                }
            }
        }

        // evaluate permissions for the service or throw exception if fail.
        DispatchContext dctx = this.getLocalContext(localName);
        if (UtilValidate.isNotEmpty(origService.permissionServiceName)) {
            Map<String, Object> permResp = origService.evalPermission(dctx, context);
            Boolean hasPermission = (Boolean) permResp.get("hasPermission");
            if (hasPermission == null) {
                throw new ServiceAuthException("ERROR: the permission-service [" + origService.permissionServiceName + "] did not return a result. Not running the service [" + origService.name + "]");
            }
            if (hasPermission.booleanValue()) {
                context.putAll(permResp);
                context = origService.makeValid(context, ModelService.IN_PARAM);
            } else {
                String message = (String) permResp.get("failMessage");
                if (UtilValidate.isEmpty(message)) {
                    message = ServiceUtil.getErrorMessage(permResp);
                }
                if (UtilValidate.isEmpty(message)) {
                    message = "You do not have permission to invoke the service [" + origService.name + "]";
                }
                throw new ServiceAuthException(message);
            }
        } else {
            if (!origService.evalPermissions(dctx, context)) {
                throw new ServiceAuthException("You do not have permission to invoke the service [" + origService.name + "]");
            }
        }

        return context;
    }

    // gets a value object from name/password pair
    private GenericValue getLoginObject(String service, String localName, String username, String password, Locale locale) throws GenericServiceException {
        Map<String, Object> context = UtilMisc.toMap("login.username", username, "login.password", password, "isServiceAuth", true, "locale", locale);

        if (Debug.verboseOn()) Debug.logVerbose("[ServiceDispathcer.authenticate] : Invoking UserLogin Service", module);

        // get the dispatch context and service model
        DispatchContext dctx = getLocalContext(localName);
        ModelService model = dctx.getModelService(service);

        // get the service engine
        GenericEngine engine = getGenericEngine(model.engineName);

        // invoke the service and get the UserLogin value object
        Map<String, Object> result = engine.runSync(localName, model, context);
        return (GenericValue) result.get("userLogin");
    }

    // checks the locale object in the context
    private Locale checkLocale(Map<String, Object> context) {
        Object locale = context.get("locale");
        Locale newLocale = null;

        if (locale != null) {
            if (locale instanceof Locale) {
                return (Locale) locale;
            } else if (locale instanceof String) {
                // en_US = lang_COUNTRY
                newLocale = UtilMisc.parseLocale((String) locale);
            }
        }

        if (newLocale == null) {
            newLocale = Locale.getDefault();
        }
        context.put("locale", newLocale);
        return newLocale;
    }

    // run startup services
    private synchronized int runStartupServices() {
        if (!enableSvcs || jm == null) {
            return 0;
        }
        int servicesScheduled = 0;
        List<StartupService> startupServices = null;
        try {
            startupServices = ServiceConfigUtil.getServiceEngine().getStartupServices();
        } catch (GenericConfigException e) {
            Debug.logWarning(e, "Exception thrown while getting service config: ", module);
            return 0;
        }
        for (StartupService startupService : startupServices) {
            String serviceName = startupService.getName();
            String runtimeDataId = startupService.getRuntimeDataId();
            int runtimeDelay = startupService.getRuntimeDelay();
            String sendToPool = startupService.getRunInPool();
            if (UtilValidate.isEmpty(sendToPool)) {
                try {
                    sendToPool = ServiceConfigUtil.getServiceEngine().getThreadPool().getSendToPool();
                } catch (GenericConfigException e) {
                    Debug.logError(e, "Unable to get send pool in service [" + serviceName + "]: ", module);
                }
            }
            // current time + 1 sec delay + extended delay
            long runtime = System.currentTimeMillis() + 1000 + runtimeDelay;
            try {
                jm.schedule(sendToPool, serviceName, runtimeDataId, runtime);
            } catch (JobManagerException e) {
                Debug.logError(e, "Unable to schedule service [" + serviceName + "]", module);
            }
        }

        return servicesScheduled;
    }

    private RunningService logService(String localName, ModelService modelService, int mode) {
        // set up the running service log
        RunningService rs = new RunningService(localName, modelService, mode);
        runLog.put(rs, this);
        return rs;
    }

    /**
     * Enables/Disables the Job Manager/Scheduler globally
     * (this will not effect any dispatchers already running)
     * @param enable
     */
    public static void enableJM(boolean enable) {
        ServiceDispatcher.enableJM = enable;
    }

    /**
     * Enables/Disables the JMS listeners globally
     * (this will not effect any dispatchers already running)
     * @param enable
     */
    public static void enableJMS(boolean enable) {
        ServiceDispatcher.enableJMS = enable;
    }

    /**
     * Enables/Disables the startup services globally
     * (this will not effect any dispatchers already running)
     * @param enable
     */
    public static void enableSvcs(boolean enable) {
        ServiceDispatcher.enableSvcs = enable;
    }

    public static Map<RunningService, ServiceDispatcher> getServiceLogMap() {
        return runLog;
    }

}