specialpurpose/workflow/src/org/ofbiz/workflow/WorkflowEngine.java

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

import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import javax.transaction.Transaction;
import org.ofbiz.base.util.Debug;
import org.ofbiz.base.util.StringUtil;
import org.ofbiz.base.util.UtilDateTime;
import org.ofbiz.base.util.UtilMisc;
import org.ofbiz.entity.GenericEntityException;
import org.ofbiz.entity.GenericValue;
import org.ofbiz.entity.transaction.GenericTransactionException;
import org.ofbiz.entity.transaction.TransactionUtil;
import org.ofbiz.service.GenericRequester;
import org.ofbiz.service.GenericResultWaiter;
import org.ofbiz.service.GenericServiceException;
import org.ofbiz.service.ModelService;
import org.ofbiz.service.ServiceDispatcher;
import org.ofbiz.service.engine.AbstractEngine;
import org.ofbiz.service.job.AbstractJob;
import org.ofbiz.service.job.Job;
import org.ofbiz.service.job.JobManagerException;

/**
 * WorkflowEngine - Workflow Service Engine
 */
public class WorkflowEngine extends AbstractEngine {

    public static final String module = WorkflowEngine.class.getName();

    public WorkflowEngine(ServiceDispatcher dispatcher) {
        super(dispatcher);
    }

    /**
     * @see org.ofbiz.service.engine.GenericEngine#runSync(java.lang.String, org.ofbiz.service.ModelService, java.util.Map)
     */
    public Map<String, Object> runSync(String localName, ModelService modelService, Map<String, Object> context) throws GenericServiceException {
        GenericResultWaiter waiter = new GenericResultWaiter();
        runAsync(localName, modelService, context, waiter, false);
        return waiter.waitForResult();
    }

    /**
     * @see org.ofbiz.service.engine.GenericEngine#runSyncIgnore(java.lang.String, org.ofbiz.service.ModelService, java.util.Map)
     */
    public void runSyncIgnore(String localName, ModelService modelService, Map<String, Object> context) throws GenericServiceException {
        runAsync(localName, modelService, context, null, false);
    }

    /**
     * @see org.ofbiz.service.engine.GenericEngine#runAsync(java.lang.String, org.ofbiz.service.ModelService, java.util.Map, boolean)
     */
    public void runAsync(String localName, ModelService modelService, Map<String, Object> context, boolean persist) throws GenericServiceException {
        runAsync(localName, modelService, context, null, persist);
    }

    /**
     * @see org.ofbiz.service.engine.GenericEngine#runAsync(java.lang.String, org.ofbiz.service.ModelService, java.util.Map, org.ofbiz.service.GenericRequester, boolean)
     */
    public void runAsync(String localName, ModelService modelService, Map<String, Object> context, GenericRequester requester, boolean persist) throws GenericServiceException {
        Transaction parentTrans = null;
        boolean beganTransaction = false;
        try {
            try {
                parentTrans = TransactionUtil.suspend();
                beganTransaction = TransactionUtil.begin();
                //Debug.logInfo("Suspended transaction; began new: " + beganTransaction, module);
            } catch (GenericTransactionException e) {
                Debug.logError(e, "Cannot begin nested transaction: " + e.getMessage(), module);
            }

            // Build the requester
            WfRequester req = null;
            try {
                req = WfFactory.getWfRequester();
            } catch (WfException e) {
                try {
                    TransactionUtil.rollback(beganTransaction, "Error getting Workflow Requester", e);
                } catch (GenericTransactionException gte) {
                    Debug.logError(gte, "Unable to rollback nested exception.", module);
                }
                throw new GenericServiceException(e.getMessage(), e);
            }

            // Get the package and process ID::VERSION
            String location = this.getLocation(modelService);
            String invoke = modelService.invoke;
            String packageId = this.getSplitPosition(location, 0);
            String packageVersion = this.getSplitPosition(location, 1);
            String processId = this.getSplitPosition(invoke, 0);
            String processVersion = this.getSplitPosition(invoke, 1);

            // Build the process manager
            WfProcessMgr mgr = null;
            try {
                mgr = WfFactory.getWfProcessMgr(dispatcher.getDelegator(), packageId, packageVersion, processId, processVersion);
            } catch (WfException e) {
                String errMsg = "Process manager error";
                Debug.logError(e, errMsg, module);
                try {
                    TransactionUtil.rollback(beganTransaction, errMsg, e);
                } catch (GenericTransactionException gte) {
                    Debug.logError(gte, "Unable to rollback nested exception.", module);
                }
                throw new GenericServiceException(e.getMessage(), e);
            } catch (Exception e) {
                Debug.logError(e, "Un-handled process manager error", module);
                throw new GenericServiceException(e.getMessage(), e);
            }

            // Create the process
            WfProcess process = null;
            try {
                process = mgr.createProcess(req);
            } catch (NotEnabled ne) {
                try {
                    TransactionUtil.rollback(beganTransaction, "Error in create workflow process: Not Enabled", ne);
                } catch (GenericTransactionException gte) {
                    Debug.logError(gte, "Unable to rollback nested exception.", module);
                }
                throw new GenericServiceException(ne.getMessage(), ne);
            } catch (InvalidRequester ir) {
                try {
                    TransactionUtil.rollback(beganTransaction, "Error in create workflow process: Invalid Requester", ir);
                } catch (GenericTransactionException gte) {
                    Debug.logError(gte, "Unable to rollback nested exception.", module);
                }
                throw new GenericServiceException(ir.getMessage(), ir);
            } catch (RequesterRequired rr) {
                try {
                    TransactionUtil.rollback(beganTransaction, "Error in create workflow process: Requester Required", rr);
                } catch (GenericTransactionException gte) {
                    Debug.logError(gte, "Unable to rollback nested exception.", module);
                }
                throw new GenericServiceException(rr.getMessage(), rr);
            } catch (WfException wfe) {
                try {
                    TransactionUtil.rollback(beganTransaction, "Error in create workflow process: general workflow error error", wfe);
                } catch (GenericTransactionException gte) {
                    Debug.logError(gte, "Unable to rollback nested exception.", module);
                }
                throw new GenericServiceException(wfe.getMessage(), wfe);
            } catch (Exception e) {
                Debug.logError(e, "Un-handled process exception", module);
                throw new GenericServiceException(e.getMessage(), e);
            }

            // Assign the owner of the process
            GenericValue userLogin = null;
            if (context.containsKey("userLogin")) {
                userLogin = (GenericValue) context.remove("userLogin");
                try {
                    Map<String, Object> fields = UtilMisc.toMap("partyId", userLogin.getString("partyId"),
                            "roleTypeId", "WF_OWNER", "workEffortId", process.runtimeKey(),
                            "fromDate", UtilDateTime.nowTimestamp());

                    try {
                        GenericValue wepa = dispatcher.getDelegator().makeValue("WorkEffortPartyAssignment", fields);
                        dispatcher.getDelegator().create(wepa);
                    } catch (GenericEntityException e) {
                        String errMsg = "Cannot set ownership of workflow";
                        try {
                            TransactionUtil.rollback(beganTransaction, errMsg, e);
                        } catch (GenericTransactionException gte) {
                            Debug.logError(gte, "Unable to rollback nested exception.", module);
                        }
                        throw new GenericServiceException(errMsg, e);
                    }
                } catch (WfException we) {
                    String errMsg = "Cannot get the workflow process runtime key";
                    try {
                        TransactionUtil.rollback(beganTransaction, errMsg, we);
                    } catch (GenericTransactionException gte) {
                        Debug.logError(gte, "Unable to rollback nested exception.", module);
                    }
                    throw new GenericServiceException(errMsg);
                }
            }

            // Grab the locale from the context
            Locale locale = (Locale) context.remove("locale");

            // Grab the starting activityId from the context
            String startActivityId = (String) context.remove("startWithActivityId");

            // Register the process and set the workflow owner
            try {
                req.registerProcess(process, context, requester);
                if (userLogin != null) {
                    Map<String, Object> pContext = process.processContext();
                    pContext.put("workflowOwnerId", userLogin.getString("userLoginId"));
                    process.setProcessContext(pContext);
                }
            } catch (WfException wfe) {
                try {
                    TransactionUtil.rollback(beganTransaction, wfe.getMessage(), wfe);
                } catch (GenericTransactionException gte) {
                    Debug.logError(gte, "Unable to rollback nested exception.", module);
                }
                throw new GenericServiceException(wfe.getMessage(), wfe);
            }

            // Set the initial locale - (in context)
            if (locale != null) {
                try {
                    Map<String, Object> pContext = process.processContext();
                    pContext.put("initialLocale", locale);
                    process.setProcessContext(pContext);
                } catch (WfException wfe) {
                    try {
                        TransactionUtil.rollback(beganTransaction, wfe.getMessage(), wfe);
                    } catch (GenericTransactionException gte) {
                        Debug.logError(gte, "Unable to rollback nested exception.", module);
                    }
                    throw new GenericServiceException(wfe.getMessage(), wfe);
                }
            }

            // Use the WorkflowRunner to start the workflow in a new thread
            try {
                Job job = new WorkflowRunner(process, requester, startActivityId);
                if (Debug.verboseOn()) Debug.logVerbose("Created WorkflowRunner: " + job, module);
                dispatcher.getJobManager().runJob(job);
            } catch (JobManagerException je) {
                try {
                    TransactionUtil.rollback(beganTransaction, je.getMessage(), je);
                } catch (GenericTransactionException gte) {
                    Debug.logError(gte, "Unable to rollback nested exception.", module);
                }
                throw new GenericServiceException(je.getMessage(), je);
            }

            try {
                TransactionUtil.commit(beganTransaction);
            } catch (GenericTransactionException e) {
                Debug.logError(e, "Cannot commit nested transaction: " + e.getMessage(), module);
            }
        } finally {
            // Resume the parent transaction
            if (parentTrans != null) {
                try {
                    TransactionUtil.resume(parentTrans);
                    //Debug.logInfo("Resumed the parent transaction.", module);
                } catch (GenericTransactionException e) {
                    throw new GenericServiceException("Could not resume transaction: " + e.toString(), e);
                }
            }
        }
    }

    private String getSplitPosition(String splitString, int position) {
        if (splitString.indexOf("::") == -1) {
            if (position == 0)
                return splitString;
            if (position == 1)
                return null;
        }
        List<String> splitList = StringUtil.split(splitString, "::");
        return splitList.get(position);
    }
}

/** Workflow Runner class runs inside its own thread using the Scheduler API */
@SuppressWarnings("serial")
class WorkflowRunner extends AbstractJob {

    GenericRequester requester;
    WfProcess process;
    String startActivityId;

    WorkflowRunner(WfProcess process, GenericRequester requester, String startActivityId) {
        super(process.toString() + "." + System.currentTimeMillis(), process.toString());
        this.process = process;
        this.requester = requester;
        this.startActivityId = startActivityId;
        runtime = new Date().getTime();
    }

    protected void finish() {
        runtime = -1;
    }

    @Override
    public void exec() {
        try {
            if (startActivityId != null)
                process.start(startActivityId);
            else
                process.start();
        } catch (Exception e) {
            Debug.logError(e, module);
            if (requester != null)
                requester.receiveResult(null);
        }
        finish();
    }
}