001/*
002 * Copyright 2006-2008 the original author or authors.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *      https://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016package org.springframework.batch.core.step.job;
017
018import org.springframework.batch.core.BatchStatus;
019import org.springframework.batch.core.ExitStatus;
020import org.springframework.batch.core.Job;
021import org.springframework.batch.core.JobExecution;
022import org.springframework.batch.core.JobParameters;
023import org.springframework.batch.core.Step;
024import org.springframework.batch.core.StepExecution;
025import org.springframework.batch.core.UnexpectedJobExecutionException;
026import org.springframework.batch.core.launch.JobLauncher;
027import org.springframework.batch.core.step.AbstractStep;
028import org.springframework.batch.item.ExecutionContext;
029import org.springframework.util.Assert;
030
031/**
032 * A {@link Step} that delegates to a {@link Job} to do its work. This is a
033 * great tool for managing dependencies between jobs, and also to modularise
034 * complex step logic into something that is testable in isolation. The job is
035 * executed with parameters that can be extracted from the step execution, hence
036 * this step can also be usefully used as the worker in a parallel or
037 * partitioned execution.
038 * 
039 * @author Dave Syer
040 * 
041 */
042public class JobStep extends AbstractStep {
043
044        /**
045         * The key for the job parameters in the step execution context. Needed for
046         * restarts.
047         */
048        private static final String JOB_PARAMETERS_KEY = JobStep.class.getName() + ".JOB_PARAMETERS";
049
050        private Job job;
051
052        private JobLauncher jobLauncher;
053
054        private JobParametersExtractor jobParametersExtractor = new DefaultJobParametersExtractor();
055
056        @Override
057        public void afterPropertiesSet() throws Exception {
058                super.afterPropertiesSet();
059                Assert.state(jobLauncher != null, "A JobLauncher must be provided");
060                Assert.state(job != null, "A Job must be provided");
061        }
062
063        /**
064         * The {@link Job} to delegate to in this step.
065         * 
066         * @param job a {@link Job}
067         */
068        public void setJob(Job job) {
069                this.job = job;
070        }
071
072        /**
073         * A {@link JobLauncher} is required to be able to run the enclosed
074         * {@link Job}.
075         * 
076         * @param jobLauncher the {@link JobLauncher} to set
077         */
078        public void setJobLauncher(JobLauncher jobLauncher) {
079                this.jobLauncher = jobLauncher;
080        }
081
082        /**
083         * The {@link JobParametersExtractor} is used to extract
084         * {@link JobParametersExtractor} from the {@link StepExecution} to run the
085         * {@link Job}. By default an instance will be provided that simply copies
086         * the {@link JobParameters} from the parent job.
087         * 
088         * @param jobParametersExtractor the {@link JobParametersExtractor} to set
089         */
090        public void setJobParametersExtractor(JobParametersExtractor jobParametersExtractor) {
091                this.jobParametersExtractor = jobParametersExtractor;
092        }
093
094        /**
095         * Execute the job provided by delegating to the {@link JobLauncher} to
096         * prevent duplicate executions. The job parameters will be generated by the
097         * {@link JobParametersExtractor} provided (if any), otherwise empty. On a
098         * restart, the job parameters will be the same as the last (failed)
099         * execution.
100         * 
101         * @see AbstractStep#doExecute(StepExecution)
102         */
103        @Override
104        protected void doExecute(StepExecution stepExecution) throws Exception {
105
106                ExecutionContext executionContext = stepExecution.getExecutionContext();
107
108                executionContext.put(STEP_TYPE_KEY, this.getClass().getName());
109
110                JobParameters jobParameters;
111                if (executionContext.containsKey(JOB_PARAMETERS_KEY)) {
112                        jobParameters = (JobParameters) executionContext.get(JOB_PARAMETERS_KEY);
113                }
114                else {
115                        jobParameters = jobParametersExtractor.getJobParameters(job, stepExecution);
116                        executionContext.put(JOB_PARAMETERS_KEY, jobParameters);
117                }
118
119                JobExecution jobExecution = jobLauncher.run(job, jobParameters);
120                
121                stepExecution.setExitStatus(determineStepExitStatus(stepExecution, jobExecution));
122
123                if (jobExecution.getStatus().isUnsuccessful()) {
124                        // AbstractStep will take care of the step execution status
125                        throw new UnexpectedJobExecutionException("Step failure: the delegate Job failed in JobStep.");
126                }
127                else if(jobExecution.getStatus().equals(BatchStatus.STOPPED)) {
128                        stepExecution.setStatus(BatchStatus.STOPPED);
129                }
130        }
131        
132        /**
133         * Determines the {@link ExitStatus} taking into consideration the {@link ExitStatus} from
134         * the {@link StepExecution}, which invoked the {@link JobStep}, and the {@link JobExecution}.
135         * 
136         * @param stepExecution the {@link StepExecution} which invoked the {@link JobExecution}
137         * @param jobExecution the {@link JobExecution}
138         * @return the final {@link ExitStatus}
139         */
140        private ExitStatus determineStepExitStatus(StepExecution stepExecution, JobExecution jobExecution) {
141                ExitStatus exitStatus = stepExecution.getExitStatus() != null ? stepExecution.getExitStatus() : ExitStatus.COMPLETED;
142                
143                return exitStatus.and(jobExecution.getExitStatus());
144        }
145
146}