| 1 | /* |
| 2 | * Copyright 2006-2007 the original author or authors. |
| 3 | * |
| 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | * you may not use this file except in compliance with the License. |
| 6 | * You may obtain a copy of the License at |
| 7 | * |
| 8 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | * |
| 10 | * Unless required by applicable law or agreed to in writing, software |
| 11 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | * See the License for the specific language governing permissions and |
| 14 | * limitations under the License. |
| 15 | */ |
| 16 | |
| 17 | package org.springframework.batch.core.job; |
| 18 | |
| 19 | import java.util.Date; |
| 20 | import java.util.Iterator; |
| 21 | import java.util.List; |
| 22 | |
| 23 | import org.springframework.batch.core.BatchStatus; |
| 24 | import org.springframework.batch.core.StartLimitExceededException; |
| 25 | import org.springframework.batch.core.UnexpectedJobExecutionException; |
| 26 | import org.springframework.batch.core.JobExecution; |
| 27 | import org.springframework.batch.core.JobExecutionException; |
| 28 | import org.springframework.batch.core.JobInstance; |
| 29 | import org.springframework.batch.core.JobInterruptedException; |
| 30 | import org.springframework.batch.core.JobExecutionListener; |
| 31 | import org.springframework.batch.core.Step; |
| 32 | import org.springframework.batch.core.StepExecution; |
| 33 | import org.springframework.batch.core.listener.CompositeExecutionJobListener; |
| 34 | import org.springframework.batch.core.repository.JobRepository; |
| 35 | import org.springframework.batch.item.ExecutionContext; |
| 36 | import org.springframework.batch.repeat.ExitStatus; |
| 37 | |
| 38 | /** |
| 39 | * Simple implementation of (@link Job} interface providing the ability to run a |
| 40 | * {@link JobExecution}. Sequentially executes a job by iterating through its |
| 41 | * list of steps. |
| 42 | * |
| 43 | * @author Lucas Ward |
| 44 | * @author Dave Syer |
| 45 | */ |
| 46 | public class SimpleJob extends AbstractJob { |
| 47 | |
| 48 | private JobRepository jobRepository; |
| 49 | |
| 50 | private CompositeExecutionJobListener listener = new CompositeExecutionJobListener(); |
| 51 | |
| 52 | /** |
| 53 | * Public setter for injecting {@link JobExecutionListener}s. They will all |
| 54 | * be given the listener callbacks at the appropriate point in the job. |
| 55 | * |
| 56 | * @param listeners the listeners to set. |
| 57 | */ |
| 58 | public void setJobExecutionListeners(JobExecutionListener[] listeners) { |
| 59 | for (int i = 0; i < listeners.length; i++) { |
| 60 | this.listener.register(listeners[i]); |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | /** |
| 65 | * Register a single listener for the {@link JobExecutionListener} |
| 66 | * callbacks. |
| 67 | * |
| 68 | * @param listener a {@link JobExecutionListener} |
| 69 | */ |
| 70 | public void registerJobExecutionListener(JobExecutionListener listener) { |
| 71 | this.listener.register(listener); |
| 72 | } |
| 73 | |
| 74 | /** |
| 75 | * Run the specified job by looping through the steps and delegating to the |
| 76 | * {@link Step}. |
| 77 | * |
| 78 | * @see org.springframework.batch.core.Job#execute(org.springframework.batch.core.JobExecution) |
| 79 | * @throws StartLimitExceededException if start limit of one of the steps |
| 80 | * was exceeded |
| 81 | */ |
| 82 | public void execute(JobExecution execution) throws JobExecutionException { |
| 83 | |
| 84 | JobInstance jobInstance = execution.getJobInstance(); |
| 85 | |
| 86 | StepExecution currentStepExecution = null; |
| 87 | int startedCount = 0; |
| 88 | List steps = getSteps(); |
| 89 | |
| 90 | try { |
| 91 | |
| 92 | // The job was already stopped before we even got this far. Deal |
| 93 | // with it in the same way as any other interruption. |
| 94 | if (execution.getStatus() == BatchStatus.STOPPING) { |
| 95 | throw new JobInterruptedException("JobExecution already stopped before being executed."); |
| 96 | } |
| 97 | |
| 98 | execution.setStartTime(new Date()); |
| 99 | updateStatus(execution, BatchStatus.STARTING); |
| 100 | |
| 101 | listener.beforeJob(execution); |
| 102 | |
| 103 | for (Iterator i = steps.iterator(); i.hasNext();) { |
| 104 | |
| 105 | Step step = (Step) i.next(); |
| 106 | |
| 107 | if (shouldStart(jobInstance, step)) { |
| 108 | |
| 109 | startedCount++; |
| 110 | updateStatus(execution, BatchStatus.STARTED); |
| 111 | currentStepExecution = execution.createStepExecution(step); |
| 112 | |
| 113 | StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step); |
| 114 | |
| 115 | boolean isRestart = (jobRepository.getStepExecutionCount(jobInstance, step) > 0 && !lastStepExecution |
| 116 | .getExitStatus().equals(ExitStatus.FINISHED)) ? true : false; |
| 117 | |
| 118 | if (isRestart && lastStepExecution != null) { |
| 119 | currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); |
| 120 | } |
| 121 | else { |
| 122 | currentStepExecution.setExecutionContext(new ExecutionContext()); |
| 123 | } |
| 124 | |
| 125 | step.execute(currentStepExecution); |
| 126 | |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | updateStatus(execution, BatchStatus.COMPLETED); |
| 131 | |
| 132 | listener.afterJob(execution); |
| 133 | |
| 134 | } |
| 135 | catch (JobInterruptedException e) { |
| 136 | execution.setStatus(BatchStatus.STOPPED); |
| 137 | listener.onInterrupt(execution); |
| 138 | rethrow(e); |
| 139 | } |
| 140 | catch (Throwable t) { |
| 141 | execution.setStatus(BatchStatus.FAILED); |
| 142 | listener.onError(execution, t); |
| 143 | rethrow(t); |
| 144 | } |
| 145 | finally { |
| 146 | ExitStatus status = ExitStatus.FAILED; |
| 147 | if (startedCount == 0) { |
| 148 | if (steps.size() > 0) { |
| 149 | status = ExitStatus.NOOP |
| 150 | .addExitDescription("All steps already completed. No processing was done."); |
| 151 | } |
| 152 | else { |
| 153 | status = ExitStatus.NOOP.addExitDescription("No steps configured for this job."); |
| 154 | } |
| 155 | } |
| 156 | else if (currentStepExecution != null) { |
| 157 | status = currentStepExecution.getExitStatus(); |
| 158 | } |
| 159 | |
| 160 | execution.setEndTime(new Date()); |
| 161 | execution.setExitStatus(status); |
| 162 | jobRepository.saveOrUpdate(execution); |
| 163 | } |
| 164 | |
| 165 | } |
| 166 | |
| 167 | private void updateStatus(JobExecution jobExecution, BatchStatus status) { |
| 168 | jobExecution.setStatus(status); |
| 169 | jobRepository.saveOrUpdate(jobExecution); |
| 170 | } |
| 171 | |
| 172 | /* |
| 173 | * Given a step and configuration, return true if the step should start, |
| 174 | * false if it should not, and throw an exception if the job should finish. |
| 175 | */ |
| 176 | private boolean shouldStart(JobInstance jobInstance, Step step) throws JobExecutionException { |
| 177 | |
| 178 | BatchStatus stepStatus; |
| 179 | // if the last execution is null, the step has never been executed. |
| 180 | StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step); |
| 181 | if (lastStepExecution == null) { |
| 182 | stepStatus = BatchStatus.STARTING; |
| 183 | } |
| 184 | else { |
| 185 | stepStatus = lastStepExecution.getStatus(); |
| 186 | } |
| 187 | |
| 188 | if (stepStatus == BatchStatus.UNKNOWN) { |
| 189 | throw new JobExecutionException("Cannot restart step from UNKNOWN status. " |
| 190 | + "The last execution ended with a failure that could not be rolled back, " |
| 191 | + "so it may be dangerous to proceed. " + "Manual intervention is probably necessary."); |
| 192 | } |
| 193 | |
| 194 | if (stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) { |
| 195 | // step is complete, false should be returned, indicating that the |
| 196 | // step should not be started |
| 197 | return false; |
| 198 | } |
| 199 | |
| 200 | if (jobRepository.getStepExecutionCount(jobInstance, step) < step.getStartLimit()) { |
| 201 | // step start count is less than start max, return true |
| 202 | return true; |
| 203 | } |
| 204 | else { |
| 205 | // start max has been exceeded, throw an exception. |
| 206 | throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName() |
| 207 | + "StartMax: " + step.getStartLimit()); |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | /** |
| 212 | * @param t |
| 213 | */ |
| 214 | private static void rethrow(Throwable t) throws RuntimeException { |
| 215 | if (t instanceof RuntimeException) { |
| 216 | throw (RuntimeException) t; |
| 217 | } |
| 218 | else if (t instanceof Error) { |
| 219 | throw (Error) t; |
| 220 | } |
| 221 | else { |
| 222 | throw new UnexpectedJobExecutionException("Unexpected checked exception in job execution", t); |
| 223 | } |
| 224 | } |
| 225 | |
| 226 | /** |
| 227 | * Public setter for the {@link JobRepository} that is needed to manage the |
| 228 | * state of the batch meta domain (jobs, steps, executions) during the life |
| 229 | * of a job. |
| 230 | * |
| 231 | * @param jobRepository |
| 232 | */ |
| 233 | public void setJobRepository(JobRepository jobRepository) { |
| 234 | this.jobRepository = jobRepository; |
| 235 | } |
| 236 | } |