| /* | |
| * 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.oodt.cas.workflow.engine.processor; | |
| //JDK imports | |
| import java.util.List; | |
| import java.util.Vector; | |
| import java.util.logging.Logger; | |
| //OODT imports | |
| import org.apache.oodt.cas.workflow.engine.ChangeType; | |
| import org.apache.oodt.cas.workflow.lifecycle.WorkflowLifecycleManager; | |
| import org.apache.oodt.cas.workflow.lifecycle.WorkflowLifecycleStage; | |
| import org.apache.oodt.cas.workflow.lifecycle.WorkflowState; | |
| import org.apache.oodt.cas.workflow.structs.WorkflowInstance; | |
| /** | |
| * | |
| * The new Apache OODT workflow style of processor. These processors are | |
| * responsible for returning the set of underlying tasks, or conditions that can | |
| * run. A sequential version will return only a single sub-processor (condition | |
| * or task, or even workflow); a parallel version will return many sub | |
| * processors to run. | |
| * | |
| * @since Apache OODT 0.4. | |
| * | |
| * @author mattmann | |
| * @author bfoster | |
| * | |
| */ | |
| public abstract class WorkflowProcessor implements WorkflowProcessorListener, | |
| Comparable<WorkflowProcessor> { | |
| private static final Logger LOG = Logger.getLogger(WorkflowProcessor.class | |
| .getName()); | |
| private WorkflowInstance workflowInstance; | |
| private WorkflowProcessor preConditions; | |
| private WorkflowProcessor postConditions; | |
| private List<String> excusedSubProcessorIds; // FIXME: read this in | |
| // PackagedRepo: flow through | |
| // instance | |
| private List<WorkflowProcessor> subProcessors; | |
| private List<WorkflowProcessorListener> listeners; | |
| private int minReqSuccessfulSubProcessors; // FIXME: read this in | |
| // PackagedRepo: flow through | |
| // instance | |
| protected WorkflowLifecycleManager lifecycleManager; | |
| protected WorkflowProcessorHelper helper; | |
| public WorkflowProcessor(WorkflowLifecycleManager lifecycleManager, | |
| WorkflowInstance workflowInstance) { | |
| this.subProcessors = new Vector<WorkflowProcessor>(); | |
| this.listeners = new Vector<WorkflowProcessorListener>(); | |
| this.excusedSubProcessorIds = new Vector<String>(); | |
| this.minReqSuccessfulSubProcessors = -1; | |
| this.lifecycleManager = lifecycleManager; | |
| this.workflowInstance = workflowInstance; | |
| this.helper = new WorkflowProcessorHelper(lifecycleManager); | |
| WorkflowState initState = helper.getLifecycleForProcessor(this) | |
| .createState("Null", "initial", | |
| "Instance created by workflow processor."); | |
| this.workflowInstance.setState(initState); | |
| } | |
| /** | |
| * @return the workflowInstance | |
| */ | |
| public WorkflowInstance getWorkflowInstance() { | |
| return workflowInstance; | |
| } | |
| /** | |
| * @param workflowInstance | |
| * the workflowInstance to set | |
| */ | |
| public void setWorkflowInstance(WorkflowInstance workflowInstance) { | |
| this.workflowInstance = workflowInstance; | |
| } | |
| /** | |
| * @return the excusedSubProcessorIds | |
| */ | |
| public List<String> getExcusedSubProcessorIds() { | |
| return excusedSubProcessorIds; | |
| } | |
| /** | |
| * @param excusedSubProcessorIds | |
| * the excusedSubProcessorIds to set | |
| */ | |
| public void setExcusedSubProcessorIds(List<String> excusedSubProcessorIds) { | |
| this.excusedSubProcessorIds = excusedSubProcessorIds; | |
| } | |
| /** | |
| * @return the subProcessors | |
| */ | |
| public List<WorkflowProcessor> getSubProcessors() { | |
| return subProcessors; | |
| } | |
| /** | |
| * @param subProcessors | |
| * the subProcessors to set | |
| */ | |
| public void setSubProcessors(List<WorkflowProcessor> subProcessors) { | |
| this.subProcessors = subProcessors; | |
| } | |
| /** | |
| * @return the listeners | |
| */ | |
| public List<WorkflowProcessorListener> getListeners() { | |
| return listeners; | |
| } | |
| /** | |
| * @param listeners | |
| * the listeners to set | |
| */ | |
| public void setListeners(List<WorkflowProcessorListener> listeners) { | |
| this.listeners = listeners; | |
| } | |
| /** | |
| * @return the minReqSuccessfulSubProcessors | |
| */ | |
| public int getMinReqSuccessfulSubProcessors() { | |
| return minReqSuccessfulSubProcessors; | |
| } | |
| /** | |
| * @param minReqSuccessfulSubProcessors | |
| * the minReqSuccessfulSubProcessors to set | |
| */ | |
| public void setMinReqSuccessfulSubProcessors(int minReqSuccessfulSubProcessors) { | |
| this.minReqSuccessfulSubProcessors = minReqSuccessfulSubProcessors; | |
| } | |
| /** | |
| * @return the lifecycleManager | |
| */ | |
| public WorkflowLifecycleManager getLifecycleManager() { | |
| return lifecycleManager; | |
| } | |
| /** | |
| * @param lifecycleManager | |
| * the lifecycleManager to set | |
| */ | |
| public void setLifecycleManager(WorkflowLifecycleManager lifecycleManager) { | |
| this.lifecycleManager = lifecycleManager; | |
| } | |
| /** | |
| * @return the preConditions | |
| */ | |
| public WorkflowProcessor getPreConditions() { | |
| return preConditions; | |
| } | |
| /** | |
| * @param preConditions | |
| * the preConditions to set | |
| */ | |
| public void setPreConditions(WorkflowProcessor preConditions) { | |
| this.preConditions = preConditions; | |
| } | |
| /** | |
| * @return the postConditions | |
| */ | |
| public WorkflowProcessor getPostConditions() { | |
| return postConditions; | |
| } | |
| /** | |
| * @param postConditions | |
| * the postConditions to set | |
| */ | |
| public void setPostConditions(WorkflowProcessor postConditions) { | |
| this.postConditions = postConditions; | |
| } | |
| /* | |
| * (non-Javadoc) | |
| * | |
| * @see java.lang.Comparable#compareTo(java.lang.Object) | |
| */ | |
| @Override | |
| public int compareTo(WorkflowProcessor workflowProcessor) { | |
| return this.getWorkflowInstance().getPriority() | |
| .compareTo(workflowProcessor.getWorkflowInstance().getPriority()); | |
| } | |
| /* | |
| * (non-Javadoc) | |
| * | |
| * @see | |
| * org.apache.oodt.cas.workflow.engine.WorkflowProcessorListener#notifyChange | |
| * (org.apache.oodt.cas.workflow.engine.WorkflowProcessor, | |
| * org.apache.oodt.cas.workflow.engine.ChangeType) | |
| */ | |
| @Override | |
| public void notifyChange(WorkflowProcessor processor, ChangeType changeType) { | |
| for (WorkflowProcessorListener listener : this.getListeners()) | |
| listener.notifyChange(this, changeType); | |
| } | |
| public synchronized List<TaskProcessor> getRunnableWorkflowProcessors() { | |
| Vector<TaskProcessor> runnableTasks = new Vector<TaskProcessor>(); | |
| // evaluate pre-conditions | |
| if (!this.passedPreConditions()) { | |
| for (WorkflowProcessor subProcessor : this.getPreConditions() | |
| .getRunnableSubProcessors()) { | |
| for (TaskProcessor tp : subProcessor.getRunnableWorkflowProcessors()) { | |
| runnableTasks.add(tp); | |
| } | |
| } | |
| } else if (this.isDone().getName().equals("ResultsFailure")) { | |
| // do nothing -- this workflow failed!!! | |
| } else if (this.isDone().getName().equals("ResultsBail")) { | |
| for (WorkflowProcessor subProcessor : this.getRunnableSubProcessors()) | |
| runnableTasks.addAll(subProcessor.getRunnableWorkflowProcessors()); | |
| } else if (!this.passedPostConditions()) { | |
| for (WorkflowProcessor subProcessor : this.getPostConditions() | |
| .getRunnableSubProcessors()) { | |
| for (TaskProcessor tp : subProcessor.getRunnableWorkflowProcessors()) { | |
| runnableTasks.add(tp); | |
| } | |
| } | |
| } | |
| return runnableTasks; | |
| } | |
| /** | |
| * Advances this WorkflowProcessor to its next {@link WorkflowState}. | |
| */ | |
| public void nextState() { | |
| if (this.workflowInstance != null | |
| && this.workflowInstance.getState() != null) { | |
| WorkflowState currState = this.workflowInstance.getState(); | |
| WorkflowState nextState = null; | |
| if (currState.getName().equals("Null")) { | |
| nextState = this.helper.getLifecycleForProcessor(this).createState( | |
| "Loaded", | |
| "initial", | |
| "Workflow Processor: nextState: " + "loading workflow instance: [" | |
| + this.workflowInstance.getId() + "]"); | |
| } else if (currState.getName().equals("Loaded")) { | |
| nextState = this.helper.getLifecycleForProcessor(this).createState( | |
| "Queued", | |
| "initial", | |
| "Workflow Processor: nextState: " + "queueing instance: [" | |
| + this.workflowInstance.getId() + "]"); | |
| } else if (currState.getName().equals("Queued")) { | |
| if (!this.passedPreConditions()) { | |
| nextState = this.helper.getLifecycleForProcessor(this).createState( | |
| "PreConditionEval", | |
| "running", | |
| "Workflow Processor: nextState: " | |
| + "running preconditiosn for workflow instance: [" | |
| + this.workflowInstance.getId() + "]"); | |
| } else { | |
| if (this.isDone().getName().equals("ResultsSuccess")) { | |
| nextState = this.helper.getLifecycleForProcessor(this).createState( | |
| "Success", | |
| "done", | |
| "Workflow Processor: nextState: " + "workflow instance: [" | |
| + this.workflowInstance.getId() | |
| + "] completed successfully"); | |
| } | |
| } | |
| } else if (currState.getName().equals("Executing")) { | |
| if(this.isDone().getName().equals("ResultsSuccess")){ | |
| nextState = this.helper.getLifecycleForProcessor(this).createState( | |
| "Success", | |
| "done", | |
| "Workflow Processor: nextState: " + "workflow instance: [" | |
| + this.workflowInstance.getId() + "] completed successfully"); | |
| } | |
| } | |
| else if(currState.getName().equals("ExecutionComplete")){ | |
| nextState = this.helper.getLifecycleForProcessor(this).createState( | |
| "Success", | |
| "done", | |
| "Workflow Processor: nextState: " + "workflow instance: [" | |
| + this.workflowInstance.getId() + "] completed successfully"); | |
| } | |
| if (nextState != null) { | |
| this.workflowInstance.setState(nextState); | |
| } | |
| } else { | |
| this.workflowInstance.setState(helper.getLifecycleForProcessor(this) | |
| .createState( | |
| "Unknown", | |
| "holding", | |
| "The Workflow Processor for instance : [" | |
| + this.getWorkflowInstance().getId() + "] " | |
| + "had a null state")); | |
| } | |
| } | |
| /** | |
| * Evaluates whether or not this processor's {@link WorkflowState} | |
| * is in any of the provided state names. | |
| * | |
| * @param states The names of states to check this processor's | |
| * {@link WorkflowState} against. | |
| * | |
| * @return True, if any of the state names provided is the name of | |
| * this processor's internal {@link WorkflowState}, False otherwise. | |
| */ | |
| public boolean isAnyState(String... states) { | |
| for (String state : states) { | |
| if (this.getWorkflowInstance().getState().getName().equals(state)) { | |
| return true; | |
| } | |
| } | |
| return false; | |
| } | |
| /** | |
| * Evaluates whether or not this processor's {@link WorkflowLifecycleStage} | |
| * is in any of the provided category names. | |
| * | |
| * @param categories The names of categories to check this processor's | |
| * {@link WorkflowLifecycleStage} against. | |
| * | |
| * @return True, if any of the category names provided is the name of | |
| * this processor's internal {@link WorkflowLifecycleStage}, False otherwise. | |
| */ | |
| public boolean isAnyCategory(String... categories) { | |
| for (String category : categories) { | |
| if (this.getWorkflowInstance().getState().getCategory().getName() | |
| .equals(category)) { | |
| return true; | |
| } | |
| } | |
| return false; | |
| } | |
| protected boolean passedPreConditions() { | |
| if (this.getPreConditions() != null) { | |
| return this.getPreConditions().getWorkflowInstance().getState().getName() | |
| .equals("Success"); | |
| } else { | |
| return true; | |
| } | |
| } | |
| protected boolean passedPostConditions() { | |
| if (this.getPostConditions() != null) { | |
| return this.getPostConditions().getWorkflowInstance().getState() | |
| .getName().equals("Success"); | |
| } else { | |
| return true; | |
| } | |
| } | |
| /** | |
| * First checks to see if any of this Processor's {@link #subProcessors} have | |
| * arrived in a state within the done category. If so the method determines if | |
| * any of the done {@link #subProcessors} are in Failure state. If so, the | |
| * method compares the number of Failed sub-processors against | |
| * {@link #minReqSuccessfulSubProcessors}, and if it is greater than it, | |
| * returns a ResultsFailure {@link WorkflowState}. Otherwise, the method scans | |
| * the failed sub-processors, and checks to see if all of them have been | |
| * excused. If they haven't, then a ResultFailure state is returned. Finally, | |
| * the method checks to ensure that all sub processors are in the done | |
| * category. If they are, a ResultsSuccess {@link WorkflowState} is returned, | |
| * otherwise, a ResultsBail state is returned. | |
| * | |
| * @return A {@link WorkflowState}, according to the method description. | |
| */ | |
| protected WorkflowState isDone() { | |
| if (this.helper.containsCategory(this.getSubProcessors(), "done")) { | |
| List<WorkflowProcessor> failedSubProcessors = this.helper | |
| .getWorkflowProcessorsByState(this.getSubProcessors(), "Failure"); | |
| if (this.minReqSuccessfulSubProcessors != -1 | |
| && failedSubProcessors.size() > (this.getSubProcessors().size() - this.minReqSuccessfulSubProcessors)) | |
| return lifecycleManager.getDefaultLifecycle().createState( | |
| "ResultsFailure", "results", | |
| "More than the allowed number of sub-processors failed"); | |
| for (WorkflowProcessor subProcessor : failedSubProcessors) { | |
| if (!this.getExcusedSubProcessorIds().contains( | |
| subProcessor.getWorkflowInstance().getId())) { | |
| return lifecycleManager.getDefaultLifecycle().createState( | |
| "ResultsFailure", | |
| "results", | |
| "Sub processor: [" + subProcessor.getWorkflowInstance().getId() | |
| + "] failed."); | |
| } | |
| } | |
| if (this.helper | |
| .allProcessorsSameCategory(this.getSubProcessors(), "done")) | |
| return lifecycleManager.getDefaultLifecycle().createState( | |
| "ResultsSuccess", | |
| "results", | |
| "Workflow Processor: processing instance id: [" | |
| + workflowInstance.getId() + "] is Done."); | |
| } | |
| return lifecycleManager.getDefaultLifecycle().createState( | |
| "ResultsBail", | |
| "results", | |
| "All sub-processors for Workflow Processor handling workflow id: [" | |
| + workflowInstance.getId() + "] are " + "not complete"); | |
| } | |
| /** | |
| * This is the core method of the WorkflowProcessor class in the new Wengine | |
| * style workflows. Instead of requiring that a processor actually walk | |
| * through the underlying {@link Workflow}, these style WorkflowProcessors | |
| * actually require their implementing sub-classes to return the current set | |
| * of Runnable sub-processors (which could be tasks, conditions, even | |
| * {@link Workflow}s themselves. | |
| * | |
| * The Parallel sub-class returns a list of task or condition processors that | |
| * are able to run at a given time. The Sequential sub-class returns only a | |
| * single task or condition processor to run, and so forth. | |
| * | |
| * @return The list of WorkflowProcessors able to currently run. | |
| */ | |
| protected abstract List<WorkflowProcessor> getRunnableSubProcessors(); | |
| protected abstract void handleSubProcessorMetadata( | |
| WorkflowProcessor workflowProcessor); | |
| } |