Start line:  
End line:  

Snippet Preview

Snippet HTML Code

Stack Overflow Questions
  /*
   * Copyright 2012 International Business Machines Corp.
   * 
   * See the NOTICE file distributed with this work for additional information
   * regarding copyright ownership. Licensed 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 com.ibm.jbatch.container.impl;
 
 import java.util.List;
 
 
 
 public class FlowControllerImpl implements IExecutionElementController {
 
 	private final static String CLASSNAME = PartitionedStepControllerImpl.class.getName();
 	private final static Logger logger = Logger.getLogger();
 	
 	
     private IPersistenceManagerService persistenceService = null;
     
     protected Flow flow;
     
     private final Navigator<FlowflowNavigator;
 	
     //
     // The currently executing controller, this will only be set to the 
     // local variable reference when we are ready to accept stop events for
     // this execution.
     
 
     public FlowControllerImpl(RuntimeJobExecutionHelper jobExecutionImplFlow flow) {
         this. = jobExecutionImpl;
         this. = flow;
         
         
          = NavigatorFactory.createFlowNavigator(flow);
     }
 
    
     @Override
     public String execute(List<Stringcontainmentthrows AbortedBeforeStartException {
         final String methodName = "execute";
         if (.isLoggable(.)) {
             .entering(methodName);
         }
 
         try {
             
             // --------------------
             // The same as a simple Job. Loop to complete all steps and decisions in the flow.
             // --------------------
             doExecutionLoop(containment);
 
            return "FLOW_CONTROLLER_RETURN_VALUE";
        } catch (Throwable t) {
                        
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            t.printStackTrace(pw);
            
            if (.isLoggable(.)) {
                .fine( + ": caught exception/error: " + t.getMessage() + " : Stack trace: " + sw.toString());
            }
            
            if (.isLoggable(.)) {
                .fine("Flow failed with exception/error: " + t.getMessage());
            }
            throw new BatchContainerRuntimeException(t);
        } finally {
            // Persist flow status, setting default if not set
            if (.isLoggable(.)) {
                .fine("Flow complete for flow id=" + .getId() + ", executionId="
                        + .getExecutionId() /* + ", batchStatus=" + currentFlowContext.getBatchStatus() + ", exitStatus="
                        + currentFlowContext.getExitStatus()*/);
            }
            try {
                //use the job status service here to persist the status        
            } catch (Throwable t) {
                if (.isLoggable(.)) {
                    StringWriter sw = new StringWriter();
                    t.printStackTrace(new PrintWriter(sw));
                    .warning("Caught Throwable on updating execution status: " + sw.toString());
                }
            }
            if (.isLoggable(.)) {
                .exiting(methodName);
            }
            
        }
        
    	
    }
    private void doExecutionLoop(Navigator<FlowflowNavigatorList<Stringcontainmentthrows Exception {
        final String methodName = "doExecutionLoop";
        ExecutionElement currentExecutionElement = null;
        try {
            currentExecutionElement = flowNavigator.getFirstExecutionElement(.getRestartOn());
        } catch (Exception e) {
            throw new IllegalArgumentException("Flow doesn't contain a step."e);
        }
        if (.isLoggable(.)) {
            .fine("First execution element = " + currentExecutionElement.getId());
        }
        // TODO can the first execution element be a decision ??? seems like
        // it's possible
        StepContextImpl<?, ?> stepContext = null;
        ExecutionElement previousExecutionElement = null;
        
        IExecutionElementController previousElementController = null;
        while (true) {
            if (!(currentExecutionElement instanceof Step) && !(currentExecutionElement instanceof Decision
            		&& !(currentExecutionElement instanceof Flow) && !(currentExecutionElement instanceof Split)) {
                throw new UnsupportedOperationException("Only support step, and decision within a flow");
            }
            if (.isLoggable(.)) {
                .fine("Next execution element = " + currentExecutionElement.getId());
            }
            IExecutionElementController elementController = 
                ExecutionElementControllerFactory.getExecutionElementController(currentExecutionElement);
            // Depending on the execution element new up the associated context
            // and add it to the controller
            if (currentExecutionElement instanceof Decision) {
                if (previousExecutionElement == null) {
                    // only job context is available to the decider since it is
                    // the first execution element in the job
                    // we need to set to null if batch artifacts are reused
                    elementController.setStepContext(null);
                } else if (previousExecutionElement instanceof Decision) {
                    throw new BatchContainerRuntimeException("A decision cannot precede another decision...OR CAN IT???");
                } else if (previousExecutionElement instanceof Step) {
                    
                    StepExecution lastStepExecution = getLastStepExecution((SteppreviousExecutionElement);
                    
                    ((DecisionControllerImpl)elementController).setStepExecution((Step)previousExecutionElementlastStepExecution);
                } else if (previousExecutionElement instanceof Split) {
                	
                	List<StepExecutionstepExecutions = getSplitStepExecutions(previousElementController);
               		
            		((DecisionControllerImpl)elementController).setStepExecutions((Split)previousExecutionElementstepExecutions);
            		
                } else if (previousExecutionElement instanceof Flow) {
                	
                    // get last step in flow
                    Step last = getLastStepInTheFlow(previousExecutionElement);
                    
                    // get last step StepExecution
                    StepExecution lastStepExecution = getLastStepExecution(last);
                    
                    ((DecisionControllerImpl)elementController).setStepExecution((Flow)previousExecutionElementlastStepExecution);
                }
            } else if (currentExecutionElement instanceof Step) {
                String stepId = ((StepcurrentExecutionElement).getId();
                stepContext = new StepContextImpl<ObjectExternalizable>(stepId);
                elementController.setStepContext(stepContext);
            } else if (currentExecutionElement instanceof Flow) {
            	// do nothing
            } else if (currentExecutionElement instanceof Split) {
            	// do nothing
            }
            if (.isLoggable(.)) {
                .fine("Start executing element = " + currentExecutionElement.getId());
            }
            
            /*
             * NOTE:
             * One approach would be to call:  jobStatusService.updateJobCurrentStep()
             * now.  However for something like a flow the element controller (flow controller) will
             * have a better view of what the "current step" is, so let's delegate to it instead. 
             */
            
            this. = elementController;
            String executionElementExitStatus = null;
            try {
                //we need to create a new copy of the containment list to pass around because we
                //don't want to modify the original containment list, since it can get reused
                //multiple times
                ArrayList<StringflowContainment = new ArrayList<String>();
                if (containment != null) {
                    flowContainment.addAll(containment);
                }
                flowContainment.add(.getId());
                executionElementExitStatus = elementController.execute(flowContainment);
            } catch (AbortedBeforeStartException e) {
                if (.isLoggable(.)) {
                    .fine("Execution failed before even getting to execute execution element = " + currentExecutionElement.getId());
                }
                throw new IllegalStateException("Execution failed before even getting to execute execution element = " + 
                        currentExecutionElement.getId() + "; breaking out of execution loop.");                
            }
            
            // set the execution element controller to null so we don't try to
            // call stop
            // on it after the element has finished executing
            this. = null;
            previousElementController = elementController;
            if (.isLoggable(.)) {
                .fine("Done executing element=" + currentExecutionElement.getId() + ", exitStatus=" + executionElementExitStatus);
            }
            Transition nextTransition = flowNavigator.getNextTransition(currentExecutionElementexecutionElementExitStatus);
            // TODO
            if (nextTransition == null) {
                if (.isLoggable(.)) {
                    .fine(methodName + " TODO: is this an expected state or not? ");
                }
                return;
            }
            
            if (nextTransition.getNextExecutionElement() != null) {
                // hold on to the previous execution element for the decider
                // we need it because we need to inject the context of the
                // previous execution element
                // into the decider
                previousExecutionElement = currentExecutionElement;
                currentExecutionElement = nextTransition.getNextExecutionElement();
                      
                if (.isLoggable(.)) {
                    .fine(methodName + " , Looping through to next execution element=" + currentExecutionElement.getId());
                }
            } else if (nextTransition.getControlElement() != null) {
                // TODO - update job status mgr
                ControlElement controlElem = nextTransition.getControlElement();
                if (.isLoggable(.)) {
                    .fine(methodName + " , Looping through to next control element=" + controlElem);
                }
                if (controlElem instanceof Stop) {
                    String restartOn = ((StopcontrolElem).getRestart();
                    if (.isLoggable(.)) {
                        .fine(methodName + " , next control element is a <stop> : " + controlElem + " with restartOn=" + restartOn);
                    }
                    //FIXME jobStatusService.updateJobStatusFromJSLStop(jobInstanceId, restartOn);
                    String newExitStatus = ((StopcontrolElem).getExitStatus();
                    if (newExitStatus != null && !newExitStatus.isEmpty()) { // overrides with exit status in JSL @exit-status
                        if (.isLoggable(.)) {
                            .fine(methodName + " , on stop, setting new JSL-specified exit status to: " + newExitStatus);
                        }
                    } 
                    if (.isLoggable(.)) {
                        .fine(methodName + " Exiting stopped job");
                    }
                    return;
                } else if (controlElem instanceof End) {
                    if (.isLoggable(.)) {
                        .fine(methodName + " , next control element is an <end>: " + controlElem);
                    }
                    String newExitStatus = ((EndcontrolElem).getExitStatus();
                    if (newExitStatus != null && !newExitStatus.isEmpty()) { // overrides with exit status in JSL @exit-status
                        if (.isLoggable(.)) {
                            .fine(methodName + " , on end, setting new JSL-specified exit status to: " + newExitStatus);
                        }
                    } 
                } else if (controlElem instanceof Fail) {
                    if (.isLoggable(.)) {
                        .fine(methodName + " , next control element is a <fail>: " + controlElem);
                    }
                    String newExitStatus = ((FailcontrolElem).getExitStatus();
                    if (newExitStatus != null && !newExitStatus.isEmpty()) { // overrides
                                                                             // with
                                                                 // in
                        if (.isLoggable(.)) {
                            .fine(methodName + " , on fail, setting new JSL-specified exit status to: " + newExitStatus);
                        }
                    } // <fail> @exit-status
                } else {
                    throw new IllegalStateException("Not sure how we'd get here but better than looping.");
                }
                return;
            } else {
                if (.isLoggable(.)) {
                    .fine(methodName + " Exiting as there are no more execution elements= ");
                }
                return;
            }
        }
	}
    
			IExecutionElementController previousElementController) {
		List<StepExecutionstepExecutions = new ArrayList<StepExecution>();
		if(previousElementController != null) {
			SplitControllerImpl controller = (SplitControllerImpl)previousElementController;
			for (BatchWorkUnit batchWorkUnit : controller.getParallelJobExecs()) {
				                			
				StepExecution lastStepExecution = null;
				for (StepExecution stepExecution : stepExecs) {
					lastStepExecution = stepExecution;
				}
				stepExecutions.add(lastStepExecution);
			}
		}
		return stepExecutions;
	}
		StepExecution lastStepExecution = null;
		for (StepExecution stepExecution : stepExecs) {
			if(last.getId().equals(stepExecution.getStepName())) {
				lastStepExecution = stepExecution;
			}
		}
		return lastStepExecution;
	}
	private Step getLastStepInTheFlow(ExecutionElement previousExecutionElement) {
		Flow flow = (Flow)previousExecutionElement;
		Step last = null;
		for (ExecutionElement elem : flow.getExecutionElements()) {
			if(elem instanceof Step) {
				last = (Stepelem;
			}
		}
		return last;
	}
    public void stop() { 
    }
    public void setStepContext(StepContextImpl<?, ? extends SerializablestepContext) {
        throw new BatchContainerRuntimeException("Incorrect usage: step context is not in scope within a flow.");
    }
    public void setAnalyzerQueue(PartitionAnalyzerProxy analyzerProxy) {
        this. = analyzerProxy;
    }
    @Override
    public void setAnalyzerQueue(BlockingQueue<PartitionDataWrapperanalyzerQueue) {
        // no-op
    }
	public void setSubJobExitStatusQueue(Stack<StringsubJobExitStatusQueue) {
		// no-op
	}
New to GrepCode? Check out our FAQ X