diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java b/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java index fa3b6ae46..f92712765 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java @@ -21,11 +21,12 @@ package org.springframework.batch.core; * Enumeration representing the status of a an Execution. * * @author Lucas Ward + * @author Dave Syer */ public enum BatchStatus { - COMPLETED, STARTED, STARTING, FAILED, STOPPING, STOPPED, UNKNOWN, PAUSED; + COMPLETED, STARTED, STARTING, PAUSED, FAILED, STOPPING, STOPPED, UNKNOWN; public static BatchStatus max(BatchStatus status1, BatchStatus status2) { if (status1.compareTo(status2)<0) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java index e689b9b29..5550facc6 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java @@ -193,6 +193,24 @@ public class JobExecution extends Entity { status = BatchStatus.STOPPING; } + /** + * Signal that this job execution wishes to be paused. + */ + public void pause() { + status = BatchStatus.PAUSED; + } + + /** + * Test if the {@link JobExecution} has been paused. + * + * @see #pause() + * + * @return true if this instance is paused + */ + public boolean isPaused() { + return status==BatchStatus.PAUSED; + } + /** * Sets the {@link ExecutionContext} for this execution * diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java index cf5d9294c..ae1b9f1a8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java @@ -217,14 +217,22 @@ public abstract class AbstractJob implements Job, BeanNameAware, InitializingBea if (execution.getStatus() != BatchStatus.STOPPING) { execution.setStartTime(new Date()); - updateStatus(execution, BatchStatus.STARTING); + // If paused we need to retain the status so that subclasses can + // handle the resume, otherwise we mark it as started... + if (!execution.isPaused()) { + updateStatus(execution, BatchStatus.STARTED); + } listener.beforeJob(execution); StepExecution lastStepExecution = doExecute(execution); if (lastStepExecution != null) { - execution.setStatus(lastStepExecution.getStatus()); + if (!execution.isPaused()) { + // If the subclass wants to pause don't change the + // status. + execution.setStatus(lastStepExecution.getStatus()); + } execution.setExitStatus(lastStepExecution.getExitStatus()); } } @@ -232,27 +240,17 @@ public abstract class AbstractJob implements Job, BeanNameAware, InitializingBea // The job was already stopped before we even got this far. Deal // with it in the same way as any other interruption. - if (execution.getStatus() == BatchStatus.PAUSED) { - // do nothing - } - else { + execution.setStatus(BatchStatus.STOPPED); + execution.setExitStatus(ExitStatus.FINISHED); - execution.setStatus(BatchStatus.STOPPED); - execution.setExitStatus(ExitStatus.FINISHED); - } } } catch (JobInterruptedException e) { logger.error(e); - if (execution.getStatus() == BatchStatus.PAUSED) { - // do nothing - } - else { - execution.setExitStatus(ExitStatus.FAILED); - execution.setStatus(BatchStatus.STOPPED); - execution.addFailureException(e); - } + execution.setExitStatus(ExitStatus.FAILED); + execution.setStatus(BatchStatus.STOPPED); + execution.addFailureException(e); } catch (Throwable t) { logger.error(t); @@ -313,7 +311,6 @@ public abstract class AbstractJob implements Job, BeanNameAware, InitializingBea if (shouldStart(jobInstance, step)) { - updateStatus(execution, BatchStatus.STARTED); currentStepExecution = execution.createStepExecution(step.getName()); StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName()); @@ -327,7 +324,7 @@ public abstract class AbstractJob implements Job, BeanNameAware, InitializingBea else { currentStepExecution.setExecutionContext(new ExecutionContext()); } - + jobRepository.add(currentStepExecution); step.execute(currentStepExecution); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/DecisionState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/DecisionState.java new file mode 100644 index 000000000..8d3cac558 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/DecisionState.java @@ -0,0 +1,26 @@ +package org.springframework.batch.core.job.flow; + +import org.springframework.batch.flow.AbstractState; + +/** + * @author Dave Syer + * + */ +public class DecisionState extends AbstractState { + + private final JobExecutionDecider decider; + + /** + * @param name + */ + DecisionState(String name, JobExecutionDecider decider) { + super(name); + this.decider = decider; + } + + @Override + public String handle(JobFlowExecutor context) throws Exception { + return decider.decide(context.getJobExecution()); + } + +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java new file mode 100644 index 000000000..3c12cedb7 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java @@ -0,0 +1,117 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.core.job.flow; + +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StartLimitExceededException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.job.AbstractJob; +import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.flow.Flow; +import org.springframework.batch.flow.FlowExecution; +import org.springframework.batch.flow.FlowExecutionException; +import org.springframework.util.Assert; + + +/** + * @author Dave Syer + * + */ +public class FlowJob extends AbstractJob { + + private Flow flow; + + /** + * Public setter for the flow. + * @param flow the flow to set + */ + public void setFlow(Flow flow) { + this.flow = flow; + } + + /** + * @see AbstractJob#doExecute(JobExecution) + */ + @Override + protected StepExecution doExecute(final JobExecution execution) throws JobExecutionException { + try { + FlowExecution result = flow.start(new JobFlowExecutor() { + public String executeStep(Step step) throws JobInterruptedException, JobRestartException, StartLimitExceededException { + StepExecution stepExecution = handleStep(step, execution); + return stepExecution==null ? FlowExecution.COMPLETED : stepExecution.getExitStatus().getExitCode(); + } + public JobExecution getJobExecution() { + return execution; + } + }); + return getLastStepExecution(execution, result); + } + catch (FlowExecutionException e) { + if (e.getCause() instanceof JobExecutionException) { + throw (JobExecutionException) e.getCause(); + } + throw new JobExecutionException("Flow execution ended unexpectedly", e); + } + } + + /** + * @param execution the current {@link JobExecution} + * @param result the result of the flow execution + * @return a {@link StepExecution} with matching properties to the result + */ + private StepExecution getLastStepExecution(JobExecution execution, FlowExecution result) { + StepExecution value = null; + StepExecution backup = null; + for (StepExecution stepExecution : execution.getStepExecutions()) { + if (stepExecution.getStepName().equals(result.getName()) + && stepExecution.getExitStatus().getExitCode().equals(result.getStatus())) { + value = stepExecution; + } + if (isLater(backup,stepExecution)) { + backup = stepExecution; + } + } + if (value==null) { + value = backup; + } + Assert.state(value != null, String.format( + "Could not locate step execution matching expected properties: flowExecution=%s, stepExecutions=%s", + result, execution.getStepExecutions())); + return value; + } + + /** + * @param first + * @param second + * @return true if the first is deemed to be executed after the second + */ + private boolean isLater(StepExecution first, StepExecution second) { + if (first==null) { + return true; + } + if (first.getEndTime()==null) { + return first.getStartTime().after(second.getStartTime()); + } + if (second.getEndTime()==null) { + return false; + } + return first.getEndTime().after(second.getEndTime()); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobExecutionDecider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobExecutionDecider.java new file mode 100644 index 000000000..a154c459f --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobExecutionDecider.java @@ -0,0 +1,36 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.core.job.flow; + +import org.springframework.batch.core.JobExecution; + +/** + * @author Dave Syer + * + */ +public interface JobExecutionDecider { + + /** + * Strategy for branching an execution based on the state of an ongoing + * {@link JobExecution}. The return value will be used as a status to + * determine the next step in the job. + * + * @param jobExecution a job execution + * @return the exit status code + */ + String decide(JobExecution jobExecution); + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java new file mode 100644 index 000000000..5a9702e66 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java @@ -0,0 +1,48 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.core.job.flow; + +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StartLimitExceededException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.flow.Flow; + +/** + * Context and execution strategy for {@link FlowJob} to allow it to delegate + * its execution step by step. + * + * @author Dave Syer + * + */ +public interface JobFlowExecutor { + + /** + * @param step a {@link Step} to execute + * @return the exit status that drives the surrounding {@link Flow} + * @throws StartLimitExceededException + * @throws JobRestartException + * @throws JobInterruptedException + */ + String executeStep(Step step) throws JobInterruptedException, JobRestartException, StartLimitExceededException; + + /** + * @return the current {@link JobExecution} + */ + JobExecution getJobExecution(); + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/PauseState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/PauseState.java new file mode 100644 index 000000000..27a4dd832 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/PauseState.java @@ -0,0 +1,40 @@ +package org.springframework.batch.core.job.flow; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.flow.AbstractState; +import org.springframework.batch.flow.FlowExecution; + +/** + * @author Dave Syer + * + */ +public class PauseState extends AbstractState { + + /** + * @param name + */ + PauseState(String name) { + super(name); + } + + @Override + public String handle(JobFlowExecutor context) throws Exception { + + JobExecution jobExecution = context.getJobExecution(); + + // This state is just a toggle for the status of the job execution. If + // not already paused we pause it, and expect the flow to respect the + // status. + if (!jobExecution.isPaused()) { + jobExecution.pause(); + return FlowExecution.PAUSED; + } + + // ...otherwise set the status to show that it has resumed + jobExecution.setStatus(BatchStatus.STARTED); + return FlowExecution.COMPLETED; + + } + +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/StepState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/StepState.java new file mode 100644 index 000000000..4afcf7106 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/StepState.java @@ -0,0 +1,31 @@ +package org.springframework.batch.core.job.flow; + +import org.springframework.batch.core.Step; +import org.springframework.batch.flow.AbstractState; +import org.springframework.batch.flow.State; + +/** + * {@link State} implementation that delegates to a {@link JobFlowExecutor} to + * execute the specified {@link Step}. + * + * @author Dave Syer + * + */ +public class StepState extends AbstractState { + + private final Step step; + + /** + * @param step the step that will be executed + */ + StepState(Step step) { + super(step.getName()); + this.step = step; + } + + @Override + public String handle(JobFlowExecutor context) throws Exception { + return context.executeStep(step); + } + +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobLauncher.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobLauncher.java index 03dd171fe..07f40d364 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobLauncher.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobLauncher.java @@ -15,13 +15,12 @@ */ package org.springframework.batch.core.launch; -import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; import org.springframework.batch.core.repository.JobRestartException; -import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; /** * Simple interface for controlling jobs, including possible ad-hoc executions, @@ -39,13 +38,11 @@ public interface JobLauncher { /** * Start a job execution for the given {@link Job} and {@link JobParameters} - * . If a JobExecution was able to be created successfully, it will always - * be returned by this method, regardless of whether or not the execution - * was successful. - * - * If there exists a past {@link JobExecution} and its status is - * {@link BatchStatus#PAUSED}, the same JobExecution should be continued - * instead of new one created. + * . If a {@link JobExecution} was able to be created successfully, it will + * always be returned by this method, regardless of whether or not the + * execution was successful. If there is a past {@link JobExecution} which + * has paused, the same {@link JobExecution} is returned instead of a new + * one created. * * @return the {@link JobExecution} if it returns synchronously. If the * implementation is asynchronous, the status might well be unknown. diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobLauncher.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobLauncher.java index fe90338c4..12c7ffdba 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobLauncher.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobLauncher.java @@ -17,16 +17,15 @@ package org.springframework.batch.core.launch.support; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.launch.JobLauncher; -import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; -import org.springframework.batch.core.repository.JobRestartException; import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.JobRestartException; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.core.task.TaskExecutor; @@ -85,7 +84,7 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean { final JobExecution jobExecution; JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters); if (lastExecution != null) { - if (lastExecution.getStatus() == BatchStatus.PAUSED) { + if (lastExecution.isPaused()) { jobExecution = lastExecution; // this execution will be continued => delete the end time jobExecution.setEndTime(null); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java index 6e64656bd..ad93a0ae3 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java @@ -370,7 +370,7 @@ public class SimpleJobOperator implements JobOperator, InitializingBean { public boolean pause(long executionId) throws NoSuchJobExecutionException { JobExecution jobExecution = findExecutionById(executionId); - jobExecution.setStatus(BatchStatus.PAUSED); + jobExecution.pause(); jobRepository.update(jobExecution); return true; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java index cd17d1ee9..d358b2039 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java @@ -322,7 +322,7 @@ public class SimpleJobRepository implements JobRepository { private void checkForInterruption(StepExecution stepExecution){ JobExecution jobExecution = stepExecution.getJobExecution(); jobExecutionDao.synchronizeStatus(jobExecution); - if(jobExecution.getStatus() == BatchStatus.STOPPING || jobExecution.getStatus() == BatchStatus.PAUSED){ + if(jobExecution.getStatus() == BatchStatus.STOPPING){ stepExecution.setTerminateOnly(); } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/ConditionalJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/ConditionalJobTests.java deleted file mode 100644 index 486c3a0bb..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/ConditionalJobTests.java +++ /dev/null @@ -1,221 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * 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 org.springframework.batch.core.job; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; - -import java.util.Arrays; -import java.util.Collections; - -import org.junit.Before; -import org.junit.Test; -import org.springframework.batch.core.BatchStatus; -import org.springframework.batch.core.JobExecution; -import org.springframework.batch.core.JobExecutionException; -import org.springframework.batch.core.JobInterruptedException; -import org.springframework.batch.core.JobParameters; -import org.springframework.batch.core.StepExecution; -import org.springframework.batch.core.UnexpectedJobExecutionException; -import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; -import org.springframework.batch.core.step.StepSupport; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.support.transaction.ResourcelessTransactionManager; - -/** - * @author Dave Syer - * - */ -public class ConditionalJobTests { - - private ConditionalJob job = new ConditionalJob("job"); - - private JobExecution jobExecution; - - @Before - public void setUp() throws Exception { - MapJobRepositoryFactoryBean.clear(); - MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean(); - factory.setTransactionManager(new ResourcelessTransactionManager()); - factory.afterPropertiesSet(); - JobRepository jobRepository = (JobRepository) factory.getObject(); - job.setJobRepository(jobRepository); - jobExecution = jobRepository.createJobExecution("job", new JobParameters()); - } - - @Test(expected = IllegalArgumentException.class) - public void testEmptySteps() throws Exception { - job.setStepTransitions(Collections. emptySet()); - job.afterPropertiesSet(); - } - - @Test(expected = IllegalArgumentException.class) - public void testNoNextStepSpecified() throws Exception { - job.setStepTransitions(Collections.singleton(new StepTransition(new StepSupport("step"), "*", "foo"))); - job.afterPropertiesSet(); - } - - @Test(expected = IllegalArgumentException.class) - public void testNoStartStep() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StepSupport("step"), "FAILED", "step"), - new StepTransition(new StepSupport("step"), "*"))); - job.afterPropertiesSet(); - } - - @Test(expected = IllegalArgumentException.class) - public void testNoEndStep() throws Exception { - job.setStepTransitions(Collections.singleton(new StepTransition(new StepSupport("step"), "FAILED", "step"))); - job.setStartStepName("step"); - job.afterPropertiesSet(); - } - - @Test(expected = IllegalArgumentException.class) - public void testMultipleStartSteps() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StubStep("step1"), "*"), new StepTransition( - new StubStep("step2"), "*"))); - job.afterPropertiesSet(); - } - - @Test - public void testNoMatchForNextStep() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StubStep("step1"), "FOO", "step2"), - new StepTransition(new StubStep("step2"), "*"))); - job.afterPropertiesSet(); - try { - job.doExecute(jobExecution); - fail("Expected JobExecutionException"); - } - catch (JobExecutionException e) { - // expected - String message = e.getMessage(); - assertTrue("Wrong message: " + message, message.toLowerCase().contains("next step not found")); - } - } - - @Test - public void testOneStep() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StubStep("step1"), "*"))); - job.afterPropertiesSet(); - StepExecution stepExecution = job.doExecute(jobExecution); - assertEquals(ExitStatus.FINISHED, stepExecution.getExitStatus()); - assertEquals(1, jobExecution.getStepExecutions().size()); - } - - @Test - public void testExplicitStartStep() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StubStep("step"), "FAILED", "step"), - new StepTransition(new StubStep("step"), "*"))); - job.setStartStepName("step"); - job.afterPropertiesSet(); - StepExecution stepExecution = job.doExecute(jobExecution); - assertEquals(ExitStatus.FINISHED, stepExecution.getExitStatus()); - assertEquals(1, jobExecution.getStepExecutions().size()); - } - - @Test - public void testTwoSteps() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StubStep("step1"), "*", "step2"), - new StepTransition(new StubStep("step2"), "*"))); - job.afterPropertiesSet(); - StepExecution stepExecution = job.doExecute(jobExecution); - assertEquals(ExitStatus.FINISHED, stepExecution.getExitStatus()); - assertEquals(2, jobExecution.getStepExecutions().size()); - } - - @Test - public void testFailedStep() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StubStep("step1") { - @Override - public void execute(StepExecution stepExecution) throws JobInterruptedException, - UnexpectedJobExecutionException { - stepExecution.setStatus(BatchStatus.FAILED); - stepExecution.setExitStatus(ExitStatus.FAILED); - } - }, "*", "step2"), new StepTransition(new StubStep("step2"), "*"))); - job.afterPropertiesSet(); - StepExecution stepExecution = job.doExecute(jobExecution); - assertEquals(ExitStatus.FINISHED, stepExecution.getExitStatus()); - assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); - assertEquals(2, jobExecution.getStepExecutions().size()); - } - - @Test - public void testStoppingStep() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StubStep("step1") { - @Override - public void execute(StepExecution stepExecution) throws JobInterruptedException, - UnexpectedJobExecutionException { - stepExecution.setStatus(BatchStatus.STOPPED); - } - }, "*", "step2"), - new StepTransition(new StubStep("step2"), "*"))); - job.afterPropertiesSet(); - try { - job.doExecute(jobExecution); - fail("Expected JobInterruptedException"); - } catch (JobInterruptedException e) { - // expected - } - assertEquals(1, jobExecution.getStepExecutions().size()); - } - - @Test - public void testBranching() throws Exception { - job.setStepTransitions(Arrays.asList(new StepTransition(new StubStep("step1"), "*", "step2"), - new StepTransition(new StubStep("step1"), "COMPLETED", "step3"), new StepTransition(new StubStep( - "step2"), "*"), new StepTransition(new StubStep("step3"), "*"))); - job.afterPropertiesSet(); - StepExecution stepExecution = job.doExecute(jobExecution); - assertEquals(ExitStatus.FINISHED, stepExecution.getExitStatus()); - assertEquals(2, jobExecution.getStepExecutions().size()); - assertEquals("step3", stepExecution.getStepName()); - } - - /** - * @author Dave Syer - * - */ - private static class StubStep extends StepSupport { - - /** - * - */ - public StubStep() { - super(); - } - - /** - * @param string - */ - public StubStep(String string) { - super(string); - } - - /** - * @see StepSupport#execute(StepExecution) - */ - @Override - public void execute(StepExecution stepExecution) throws JobInterruptedException, - UnexpectedJobExecutionException { - stepExecution.setStatus(BatchStatus.COMPLETED); - stepExecution.setExitStatus(ExitStatus.FINISHED); - } - - } - -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/StepTransitionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/StepTransitionTests.java deleted file mode 100644 index 98eb2e7a1..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/StepTransitionTests.java +++ /dev/null @@ -1,129 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * 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 org.springframework.batch.core.job; - -import static org.junit.Assert.*; - -import org.junit.Test; -import org.springframework.batch.core.step.StepSupport; -import org.springframework.batch.repeat.ExitStatus; - -/** - * @author Dave Syer - * - */ -public class StepTransitionTests { - - @Test - public void testIsEnd() { - StepTransition transition = new StepTransition(new StepSupport(), ""); - assertTrue(transition.isEnd()); - assertNull(transition.getNext()); - } - - @Test - public void testMatchesStar() { - StepTransition transition = new StepTransition(new StepSupport(), "*", "start"); - assertTrue(transition.matches(ExitStatus.CONTINUABLE)); - } - - @Test - public void testMatchesNull() { - StepTransition transition = new StepTransition(new StepSupport(), null, "start"); - assertTrue(transition.matches(ExitStatus.CONTINUABLE)); - } - - @Test - public void testMatchesEmpty() { - StepTransition transition = new StepTransition(new StepSupport(), "", "start"); - assertTrue(transition.matches(ExitStatus.CONTINUABLE)); - } - - @Test - public void testMatchesExact() { - StepTransition transition = new StepTransition(new StepSupport(), "CONTINUABLE", "start"); - assertTrue(transition.matches(ExitStatus.CONTINUABLE)); - } - - @Test - public void testMatchesWildcard() { - StepTransition transition = new StepTransition(new StepSupport(), "CONTIN*", "start" ); - assertTrue(transition.matches(ExitStatus.CONTINUABLE)); - } - - @Test - public void testMatchesPlaceholder() { - StepTransition transition = new StepTransition(new StepSupport(), "CONTIN???LE", "start"); - assertTrue(transition.matches(ExitStatus.CONTINUABLE)); - } - - @Test - public void testSimpleOrderingEqual() { - StepTransition transition = new StepTransition(new StepSupport(), "CONTIN???LE", "start"); - assertEquals(0, transition.compareTo(transition)); - } - - @Test - public void testSimpleOrderingMoreGeneral() { - StepTransition transition = new StepTransition(new StepSupport(), "CONTIN???LE", "start"); - StepTransition other = new StepTransition(new StepSupport(), "CONTINUABLE", "start"); - assertEquals(1, transition.compareTo(other)); - assertEquals(-1, other.compareTo(transition)); - } - - @Test - public void testSimpleOrderingMostGeneral() { - StepTransition transition = new StepTransition(new StepSupport(), "*", "start"); - StepTransition other = new StepTransition(new StepSupport(), "CONTINUABLE", "start"); - assertEquals(1, transition.compareTo(other)); - assertEquals(-1, other.compareTo(transition)); - } - - @Test - public void testSubstringAndWildcard() { - StepTransition transition = new StepTransition(new StepSupport(), "CONTIN*", "start"); - StepTransition other = new StepTransition(new StepSupport(), "CONTINUABLE", "start"); - assertEquals(1, transition.compareTo(other)); - assertEquals(-1, other.compareTo(transition)); - } - - @Test - public void testSimpleOrderingMostToNextGeneral() { - StepTransition transition = new StepTransition(new StepSupport(), "*", "start"); - StepTransition other = new StepTransition(new StepSupport(), "C?", "start"); - assertEquals(1, transition.compareTo(other)); - assertEquals(-1, other.compareTo(transition)); - } - - @Test - public void testSimpleOrderingAdjacent() { - StepTransition transition = new StepTransition(new StepSupport(), "CON*", "start"); - StepTransition other = new StepTransition(new StepSupport(), "CON?", "start"); - assertEquals(1, transition.compareTo(other)); - assertEquals(-1, other.compareTo(transition)); - } - - @Test - public void testToString() { - StepTransition transition = new StepTransition(new StepSupport(), "CONTIN???LE", "start"); - String string = transition.toString(); - assertTrue("Wrong string: " + string, string.contains("StepTransition")); - assertTrue("Wrong string: " + string, string.contains("start")); - assertTrue("Wrong string: " + string, string.contains("CONTIN???LE")); - assertTrue("Wrong string: " + string, string.contains("next=")); - } - -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java new file mode 100644 index 000000000..bde04f0ba --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java @@ -0,0 +1,228 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.core.job.flow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.UnexpectedJobExecutionException; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.StepSupport; +import org.springframework.batch.flow.SimpleFlow; +import org.springframework.batch.flow.StateTransition; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; + +/** + * @author Dave Syer + * + */ +public class FlowJobTests { + + private FlowJob job = new FlowJob(); + + private JobExecution jobExecution; + + private JobRepository jobRepository; + + @Before + public void setUp() throws Exception { + MapJobRepositoryFactoryBean.clear(); + MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean(); + factory.setTransactionManager(new ResourcelessTransactionManager()); + factory.afterPropertiesSet(); + jobRepository = (JobRepository) factory.getObject(); + job.setJobRepository(jobRepository); + jobExecution = jobRepository.createJobExecution("job", new JobParameters()); + } + + @Test + public void testTwoSteps() throws Exception { + SimpleFlow flow = new SimpleFlow("job"); + Collection> transitions = new ArrayList>(); + transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2")); + transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step2")))); + flow.setStateTransitions(transitions); + job.setFlow(flow); + job.afterPropertiesSet(); + StepExecution stepExecution = job.doExecute(jobExecution); + assertEquals(ExitStatus.FINISHED, stepExecution.getExitStatus()); + assertEquals(2, jobExecution.getStepExecutions().size()); + } + + @Test + public void testFailedStep() throws Exception { + SimpleFlow flow = new SimpleFlow("job"); + Collection> transitions = new ArrayList>(); + transitions.add(StateTransition.createStateTransition(new StepState(new StepSupport("step1") { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException, + UnexpectedJobExecutionException { + stepExecution.setStatus(BatchStatus.FAILED); + stepExecution.setExitStatus(ExitStatus.FAILED); + } + }), "step2")); + transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step2")))); + flow.setStateTransitions(transitions); + job.setFlow(flow); + job.afterPropertiesSet(); + StepExecution stepExecution = job.doExecute(jobExecution); + assertEquals(ExitStatus.FINISHED, stepExecution.getExitStatus()); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, jobExecution.getStepExecutions().size()); + } + + @Test + public void testStoppingStep() throws Exception { + SimpleFlow flow = new SimpleFlow("job"); + Collection> transitions = new ArrayList>(); + transitions.add(StateTransition.createStateTransition(new StepState(new StepSupport("step1") { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException, + UnexpectedJobExecutionException { + stepExecution.setStatus(BatchStatus.STOPPED); + } + }), "step2")); + transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step2")))); + flow.setStateTransitions(transitions); + job.setFlow(flow); + job.afterPropertiesSet(); + try { + job.doExecute(jobExecution); + fail("Expected JobInterruptedException"); + } + catch (JobInterruptedException e) { + // expected + } + assertEquals(1, jobExecution.getStepExecutions().size()); + } + + @Test + public void testBranching() throws Exception { + SimpleFlow flow = new SimpleFlow("job"); + Collection> transitions = new ArrayList>(); + transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2")); + transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "COMPLETED", "step3")); + transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step2")))); + transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step3")))); + flow.setStateTransitions(transitions); + job.setFlow(flow); + job.afterPropertiesSet(); + StepExecution stepExecution = job.doExecute(jobExecution); + assertEquals(ExitStatus.FINISHED, stepExecution.getExitStatus()); + assertEquals(2, jobExecution.getStepExecutions().size()); + assertEquals("step3", stepExecution.getStepName()); + } + + @Test + public void testBasicFlow() throws Throwable { + SimpleFlow flow = new SimpleFlow("job"); + Step step = new StubStep("step"); + flow.setStateTransitions(Collections.singleton(StateTransition.createEndStateTransition(new StepState(step), + "*"))); + job.setFlow(flow); + job.execute(jobExecution); + if (!jobExecution.getAllFailureExceptions().isEmpty()) { + throw jobExecution.getAllFailureExceptions().get(0); + } + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + } + + @Test + public void testDecisionFlow() throws Throwable { + SimpleFlow flow = new SimpleFlow("job"); + JobExecutionDecider decider = new JobExecutionDecider() { + public String decide(JobExecution jobExecution) { + return "SWITCH"; + } + }; + Collection> transitions = new ArrayList>(); + transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "*", "decision")); + transitions.add(StateTransition.createStateTransition(new DecisionState("decision", decider), "*", "step2")); + transitions.add(StateTransition + .createStateTransition(new DecisionState("decision", decider), "SWITCH", "step3")); + transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step2")), "*")); + transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step3")), "*")); + flow.setStateTransitions(transitions); + job.setFlow(flow); + StepExecution stepExecution = job.doExecute(jobExecution); + if (!jobExecution.getAllFailureExceptions().isEmpty()) { + throw jobExecution.getAllFailureExceptions().get(0); + } + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, jobExecution.getStepExecutions().size()); + assertEquals("step3", stepExecution.getStepName()); + } + + @Test + public void testPauseFlow() throws Throwable { + + SimpleFlow flow = new SimpleFlow("job"); + Collection> transitions = new ArrayList>(); + transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "*", "pause")); + transitions.add(StateTransition.createStateTransition(new PauseState("pause"), "*", "step2")); + transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step2")), "*")); + flow.setStateTransitions(transitions); + job.setFlow(flow); + + job.execute(jobExecution); + if (!jobExecution.getAllFailureExceptions().isEmpty()) { + throw jobExecution.getAllFailureExceptions().get(0); + } + assertEquals(BatchStatus.PAUSED, jobExecution.getStatus()); + assertEquals(1, jobExecution.getStepExecutions().size()); + + job.execute(jobExecution); + if (!jobExecution.getAllFailureExceptions().isEmpty()) { + throw jobExecution.getAllFailureExceptions().get(0); + } + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + assertEquals(2, jobExecution.getStepExecutions().size()); + + } + + /** + * @author Dave Syer + * + */ + private class StubStep extends StepSupport { + + private StubStep(String name) { + super(name); + } + + public void execute(StepExecution stepExecution) throws JobInterruptedException { + stepExecution.setStatus(BatchStatus.COMPLETED); + stepExecution.setExitStatus(ExitStatus.FINISHED); + jobRepository.update(stepExecution); + } + + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/SimpleJobLauncherTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/SimpleJobLauncherTests.java index a3e126451..2c242b600 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/SimpleJobLauncherTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/SimpleJobLauncherTests.java @@ -30,7 +30,6 @@ import java.util.List; import org.junit.Before; import org.junit.Test; -import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; @@ -195,7 +194,7 @@ public class SimpleJobLauncherTests { public void testResumePausedInstance() throws Exception { long id = 9; JobExecution jobExecution = new JobExecution(null, id); - jobExecution.setStatus(BatchStatus.PAUSED); + jobExecution.pause(); expect(jobRepository.getLastJobExecution(job.getName(), jobParameters)).andReturn(jobExecution); replay(jobRepository); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/AbstractState.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/AbstractState.java new file mode 100644 index 000000000..7b3f5cc3b --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/AbstractState.java @@ -0,0 +1,40 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + + +/** + * @author Dave Syer + * + */ +public abstract class AbstractState implements State { + + private final String name; + + /** + * + */ + public AbstractState(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public abstract String handle(T context) throws Exception; + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/Flow.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/Flow.java new file mode 100644 index 000000000..39e51e5e3 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/Flow.java @@ -0,0 +1,44 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + + +/** + * @author Dave Syer + * + * @param + */ +public interface Flow { + + /** + * @return the name of the flow + */ + String getName(); + + /** + * @throws FlowExecutionException + */ + FlowExecution start(T context) throws FlowExecutionException; + + /** + * @param stateName the name of the {@link State} to resume on + * @param context the context to be passed into each {@link State} executed + * @return a {@link FlowExecution} containing the exit status of the flow + * @throws FlowExecutionException + */ + FlowExecution resume(String stateName, T context) throws FlowExecutionException; + +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecution.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecution.java new file mode 100644 index 000000000..44cd9e541 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecution.java @@ -0,0 +1,111 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +/** + * @author Dave Syer + * + */ +public class FlowExecution implements Comparable { + + /** + * Special well-known status value. + */ + public static final String COMPLETED = Status.COMPLETED.toString(); + + /** + * Special well-known status value. + */ + public static final String PAUSED = Status.PAUSED.toString(); + + /** + * Special well-known status value. + */ + public static final String FAILED = Status.FAILED.toString(); + + /** + * Special well-known status value. + */ + public static final String UNKNOWN = Status.UNKNOWN.toString(); + + private final String name; + + private final String status; + + private enum Status { + + COMPLETED, PAUSED, FAILED, UNKNOWN; + + static Status match(String value) { + for (int i = 0; i < values().length; i++) { + Status status = values()[i]; + if (value.startsWith(status.toString())) { + return status; + } + } + // Default match should be the lowest priority + return COMPLETED; + } + + }; + + /** + * + */ + public FlowExecution(String name, String status) { + this.name = name; + this.status = status; + } + + /** + * @return the name of the end state reached + */ + public String getName() { + return name; + } + + /** + * @return the exit status + */ + public String getStatus() { + return status; + } + + /** + * Create an ordering on {@link FlowExecution} instances by comparing their + * statuses. + * + * @see Comparable#compareTo(Object) + * + * @param other + * @return negative, zero or positive as per the contract + */ + public int compareTo(FlowExecution other) { + Status one = Status.match(this.getStatus()); + Status two = Status.match(other.getStatus()); + int comparison = one.compareTo(two); + if (comparison==0) { + return this.getStatus().compareTo(other.getStatus()); + } + return comparison; + } + + @Override + public String toString() { + return String.format("FlowExecution: name=%s, status=%s", name, status); + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionAggregator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionAggregator.java new file mode 100644 index 000000000..c3bdde6c1 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionAggregator.java @@ -0,0 +1,35 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import java.util.Collection; + +/** + * Strategy interface for aggregating {@link FlowExecution} instances into a + * single exit status. + * + * @author Dave Syer + * + */ +public interface FlowExecutionAggregator { + + /** + * @param executions the executions to aggregate + * @return a summary status for the whole lot + */ + String aggregate(Collection executions); + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionException.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionException.java new file mode 100644 index 000000000..a7184daf1 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionException.java @@ -0,0 +1,39 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +/** + * @author Dave Syer + * + */ +public class FlowExecutionException extends Exception { + + /** + * @param message + */ + public FlowExecutionException(String message) { + super(message); + } + + /** + * @param message + * @param cause + */ + public FlowExecutionException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/MaxValueFlowExecutionAggregator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/MaxValueFlowExecutionAggregator.java new file mode 100644 index 000000000..3332f97b7 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/MaxValueFlowExecutionAggregator.java @@ -0,0 +1,37 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import java.util.Collection; +import java.util.Collections; + +/** + * @author Dave Syer + * + */ +public class MaxValueFlowExecutionAggregator implements FlowExecutionAggregator { + + /** + * @see FlowExecutionAggregator#aggregate(Collection) + */ + public String aggregate(Collection executions) { + if (executions==null || executions.size()==0) { + return FlowExecution.UNKNOWN; + } + return Collections.max(executions).getStatus(); + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/PatternMatcher.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/PatternMatcher.java new file mode 100644 index 000000000..c7620b078 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/PatternMatcher.java @@ -0,0 +1,162 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +/** + * @author Dave Syer + * + */ +public class PatternMatcher { + + /** + * Lifted from AntPathMatcher in Spring Core. Tests whether or not a string + * matches against a pattern. The pattern may contain two special + * characters:
+ * '*' means zero or more characters
+ * '?' means one and only one character + * @param pattern pattern to match against. Must not be null. + * @param str string which must be matched against the pattern. Must not be + * null. + * @return true if the string matches against the pattern, or + * false otherwise. + */ + public static boolean match(String pattern, String str) { + char[] patArr = pattern.toCharArray(); + char[] strArr = str.toCharArray(); + int patIdxStart = 0; + int patIdxEnd = patArr.length - 1; + int strIdxStart = 0; + int strIdxEnd = strArr.length - 1; + char ch; + + boolean containsStar = pattern.contains("*"); + + if (!containsStar) { + // No '*'s, so we make a shortcut + if (patIdxEnd != strIdxEnd) { + return false; // Pattern and string do not have the same size + } + for (int i = 0; i <= patIdxEnd; i++) { + ch = patArr[i]; + if (ch != '?') { + if (ch != strArr[i]) { + return false;// Character mismatch + } + } + } + return true; // String matches against pattern + } + + if (patIdxEnd == 0) { + return true; // Pattern contains only '*', which matches anything + } + + // Process characters before first star + while ((ch = patArr[patIdxStart]) != '*' && strIdxStart <= strIdxEnd) { + if (ch != '?') { + if (ch != strArr[strIdxStart]) { + return false;// Character mismatch + } + } + patIdxStart++; + strIdxStart++; + } + if (strIdxStart > strIdxEnd) { + // All characters in the string are used. Check if only '*'s are + // left in the pattern. If so, we succeeded. Otherwise failure. + for (int i = patIdxStart; i <= patIdxEnd; i++) { + if (patArr[i] != '*') { + return false; + } + } + return true; + } + + // Process characters after last star + while ((ch = patArr[patIdxEnd]) != '*' && strIdxStart <= strIdxEnd) { + if (ch != '?') { + if (ch != strArr[strIdxEnd]) { + return false;// Character mismatch + } + } + patIdxEnd--; + strIdxEnd--; + } + if (strIdxStart > strIdxEnd) { + // All characters in the string are used. Check if only '*'s are + // left in the pattern. If so, we succeeded. Otherwise failure. + for (int i = patIdxStart; i <= patIdxEnd; i++) { + if (patArr[i] != '*') { + return false; + } + } + return true; + } + + // process pattern between stars. padIdxStart and patIdxEnd point + // always to a '*'. + while (patIdxStart != patIdxEnd && strIdxStart <= strIdxEnd) { + int patIdxTmp = -1; + for (int i = patIdxStart + 1; i <= patIdxEnd; i++) { + if (patArr[i] == '*') { + patIdxTmp = i; + break; + } + } + if (patIdxTmp == patIdxStart + 1) { + // Two stars next to each other, skip the first one. + patIdxStart++; + continue; + } + // Find the pattern between padIdxStart & padIdxTmp in str between + // strIdxStart & strIdxEnd + int patLength = (patIdxTmp - patIdxStart - 1); + int strLength = (strIdxEnd - strIdxStart + 1); + int foundIdx = -1; + strLoop: for (int i = 0; i <= strLength - patLength; i++) { + for (int j = 0; j < patLength; j++) { + ch = patArr[patIdxStart + j + 1]; + if (ch != '?') { + if (ch != strArr[strIdxStart + i + j]) { + continue strLoop; + } + } + } + + foundIdx = strIdxStart + i; + break; + } + + if (foundIdx == -1) { + return false; + } + + patIdxStart = patIdxTmp; + strIdxStart = foundIdx + patLength; + } + + // All characters in the string are used. Check if only '*'s are left + // in the pattern. If so, we succeeded. Otherwise failure. + for (int i = patIdxStart; i <= patIdxEnd; i++) { + if (patArr[i] != '*') { + return false; + } + } + + return true; + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SimpleFlow.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SimpleFlow.java new file mode 100644 index 000000000..b8b4a70c8 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SimpleFlow.java @@ -0,0 +1,265 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; + +import org.springframework.beans.factory.InitializingBean; + +import com.sun.org.apache.xerces.internal.impl.xpath.XPath.Step; + +/** + * A {@link Flow} that branches conditionally depending on the exit status of + * the last {@link State}. The input parameters are the state transitions (in no + * particular order). The start state name can be specified explicitly (and must + * exist in the set of transitions), or computed from the existing transitions, + * if unambiguous. + * + * @author Dave Syer + * + */ +public class SimpleFlow implements Flow { + + private State startState; + + private Map>> transitionMap = new HashMap>>(); + + private Map> stateMap = new HashMap>(); + + private String startStateName; + + private Collection> stateTransitions = new HashSet>(); + + private final String name; + + /** + * Create a flow with the given name. + * + * @param name the name of the flow + */ + public SimpleFlow(String name) { + this.name = name; + } + + /** + * Get the name for this flow. + * + * @see Flow#getName() + */ + public String getName() { + return name; + } + + /** + * Public setter for the start state name. + * @param startStateName the name of the start state + */ + public void setStartStateName(String startStateName) { + this.startStateName = startStateName; + } + + /** + * Public setter for the stateTransitions. + * @param stateTransitions the stateTransitions to set + */ + public void setStateTransitions(Collection> stateTransitions) { + + this.stateTransitions = stateTransitions; + } + + /** + * Locate start step and pre-populate data structures needed for execution. + * + * @see InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + initializeTransitions(); + } + + /** + * @see Flow#start(Object) + */ + public FlowExecution start(T context) throws FlowExecutionException { + if (startState == null) { + initializeTransitions(); + } + State state = startState; + String stateName = state.getName(); + return resume(stateName, context); + } + + /** + * @see Flow#resume(String, Object) + */ + public FlowExecution resume(String stateName, T context) throws FlowExecutionException { + String status = FlowExecution.UNKNOWN; + State state = stateMap.get(stateName); + // Terminate if there are no more states + while (state != null) { + stateName = state.getName(); + try { + status = state.handle(context); + } + catch (Exception e) { + throw new FlowExecutionException(String.format("Ended flow=%s at state=%s with exception", name, + stateName), e); + } + state = nextState(stateName, status); + } + return new FlowExecution(stateName, status); + } + + /** + * @return the next {@link Step} (or null if this is the end) + * @throws JobExecutionException + */ + private State nextState(String stepName, String status) throws FlowExecutionException { + + // Special status value indicating that a state wishes to pause + // execution + if (status.equals(FlowExecution.PAUSED)) { + return null; + } + + Set> set = transitionMap.get(stepName); + + if (set == null) { + throw new FlowExecutionException(String.format("No transitions found in flow=%s for state=%s", getName(), + stepName)); + } + + String next = null; + for (StateTransition stateTransition : set) { + if (stateTransition.matches(status)) { + if (stateTransition.isEnd()) { + // End of job + return null; + } + next = stateTransition.getNext(); + break; + } + } + + if (next == null) { + throw new FlowExecutionException(String.format( + "Next state not found in flow=%s for step=%s with exit status=%s", getName(), stepName, status)); + } + + if (!stateMap.containsKey(next)) { + throw new FlowExecutionException(String.format("Next state not specified in flow=%s for next=%s", + getName(), next)); + } + + return stateMap.get(next); + + } + + /** + * Analyse the transitions provided and generate all the information needed + * to execute the flow. + */ + private void initializeTransitions() { + startState = null; + transitionMap.clear(); + stateMap.clear(); + boolean hasEndStep = false; + + for (StateTransition stepTransition : stateTransitions) { + State step = stepTransition.getState(); + stateMap.put(step.getName(), step); + } + + for (StateTransition stateTransition : stateTransitions) { + + State state = stateTransition.getState(); + + if (!stateTransition.isEnd()) { + + String next = stateTransition.getNext(); + + if (!stateMap.containsKey(next)) { + throw new IllegalArgumentException("Missing step for [" + stateTransition + "]"); + } + + } + else { + hasEndStep = true; + } + + String name = state.getName(); + + SortedSet> set = transitionMap.get(name); + if (set == null) { + set = new TreeSet>(); + transitionMap.put(name, set); + } + set.add(stateTransition); + + } + + if (!hasEndStep) { + throw new IllegalArgumentException( + "No end step was found. You must specify at least one transition with no next state."); + } + + if (startStateName != null) { + + startState = stateMap.get(startStateName); + if (startState == null) { + throw new IllegalArgumentException( + "Start state does not exist (if you specify a startStateName make sure " + + "a state with that name is in one of the transitions): [" + startStateName + "]"); + } + + } + else { + + // Try and locate a transition with no incoming links + + Set nextStateNames = new HashSet(); + + for (StateTransition stepTransition : stateTransitions) { + nextStateNames.add(stepTransition.getNext()); + } + + for (StateTransition stepTransition : stateTransitions) { + State state = stepTransition.getState(); + if (!nextStateNames.contains(state.getName())) { + if (startState != null && !startState.getName().equals(state.getName())) { + throw new IllegalArgumentException(String.format( + "Multiple possible start steps found: [%s, %s]. " + + "Please specify one explicitly with the startStateName property.", startState + .getName(), state.getName())); + } + startState = state; + } + } + + if (startState == null) { + throw new IllegalArgumentException( + "No start state could be located (no transition without incoming links)"); + } + + } + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SplitState.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SplitState.java new file mode 100644 index 000000000..fde283e32 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SplitState.java @@ -0,0 +1,103 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.concurrent.Callable; +import java.util.concurrent.FutureTask; + +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.core.task.TaskRejectedException; + +/** + * A {@link State} implementation that splits a {@link Flow} into multiple + * parallel subflows. + * + * @author Dave Syer + * + */ +public class SplitState extends AbstractState { + + private final Collection> flows; + + private TaskExecutor taskExecutor = new SyncTaskExecutor(); + + private FlowExecutionAggregator aggregator = new MaxValueFlowExecutionAggregator(); + + /** + * @param name + */ + public SplitState(String name, Collection> flows) { + super(name); + this.flows = flows; + } + + /** + * Public setter for the taskExecutor. + * @param taskExecutor the taskExecutor to set + */ + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + /** + * Execute the flows in parallel by passing them to the {@link TaskExecutor} + * and wait for all of them to finish before proceeding. + * + * @see State#handle(Object) + */ + @Override + public String handle(final T context) throws Exception { + + Collection> tasks = new ArrayList>(); + + for (final Flow flow : flows) { + + final FutureTask task = new FutureTask(new Callable() { + public FlowExecution call() throws Exception { + return flow.start(context); + } + }); + + tasks.add(task); + + try { + taskExecutor.execute(new Runnable() { + public void run() { + task.run(); + } + }); + } + catch (TaskRejectedException e) { + throw new FlowExecutionException("TaskExecutor rejected task for flow=" + flow.getName()); + } + + } + + Collection results = new ArrayList(); + + // TODO: could use a CompletionSerice? + for (FutureTask task : tasks) { + results.add(task.get()); + } + + return aggregator.aggregate(results); + + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/State.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/State.java new file mode 100644 index 000000000..16021aabc --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/State.java @@ -0,0 +1,47 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + + +/** + * @author Dave Syer + * + */ +public interface State { + + /** + * The name of the state. Should be unique within a flow. + * + * @return the name of this state + */ + String getName(); + + /** + * Handle some business or processing logic and return a status that can be + * used to drive a flow to the next {@link State}. The status can be any + * string, but special meaning is assigned to the static constants in + * {@link FlowExecution}. The context can be used by implementations to do + * whatever they need to do. The same context will be passed to all + * {@link State} instances, so implementations should be careful that the + * context is thread safe, or used in a thread safe manner. + * + * @param context the context passed in by the caller + * @return a status for the execution + * @throws Exception if anything goes wrong + */ + String handle(T context) throws Exception; + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/StateTransition.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/StateTransition.java new file mode 100644 index 000000000..75479a31d --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/StateTransition.java @@ -0,0 +1,180 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.util.StringUtils; + +/** + * Value object representing a potential transition from one {@link State} to + * another. The originating State name and the next {@link State} to execute are + * linked by a pattern for the {@link ExitStatus#getExitCode() exit code} of an + * execution of the originating State. + * + * @author Dave Syer + * + */ +public class StateTransition implements Comparable> { + + private final State state; + + private final String pattern; + + private final String next; + + /** + * Create a new end state {@link StateTransition} specification. This + * transition explicitly goes unconditionally to an end state (i.e. no more + * executions). + * + * @param state the {@link State} used to generate the outcome for this + * transition + */ + public static StateTransition createEndStateTransition(State state) { + return createStateTransition(state, null, null); + } + + /** + * Create a new end state {@link StateTransition} specification. This + * transition explicitly goes to an end state (i.e. no more processing) if + * the outcome matches the pattern. + * + * @param state the {@link State} used to generate the outcome for this + * transition + * @param pattern the pattern to match in the exit status of the + * {@link State} + */ + public static StateTransition createEndStateTransition(State state, String pattern) { + return createStateTransition(state, pattern, null); + } + + /** + * Create a new state {@link StateTransition} specification with a wildcard + * pattern that matches all outcomes. + * + * @param state the {@link State} used to generate the outcome for this + * transition + * @param next the name of the next {@link State} to execute + */ + public static StateTransition createStateTransition(State state, String next) { + return createStateTransition(state, null, next); + } + + /** + * Create a new {@link StateTransition} specification from one {@link State} + * to another (by name). + * + * @param state the {@link State} used to generate the outcome for this + * transition + * @param pattern the pattern to match in the exit status of the + * {@link State} + * @param next the name of the next {@link State} to execute + */ + public static StateTransition createStateTransition(State state, String pattern, String next) { + return new StateTransition(state, pattern, next); + } + + private StateTransition(State state, String pattern, String next) { + super(); + if (!StringUtils.hasText(pattern)) { + this.pattern = "*"; + } + else { + this.pattern = pattern; + } + this.next = next; + this.state = state; + } + + /** + * Public getter for the State. + * @return the State + */ + public State getState() { + return state; + } + + /** + * Public getter for the next State name. + * @return the next + */ + public String getNext() { + return next; + } + + /** + * Check if the provided status matches the pattern, signalling that the + * next State should be executed. + * + * @param status the status to compare + * @return true if the pattern matches this status + */ + public boolean matches(String status) { + return PatternMatcher.match(pattern, status); + } + + /** + * Check for a special next State signalling the end of a job. + * + * @return true if this transition goes nowhere (there is no next) + */ + public boolean isEnd() { + return next == null; + } + + /** + * Sorts by decreasing specificity of pattern, based on just counting + * wildcards (with * taking precedence over ?). If wildcard counts are equal + * then falls back to alphabetic comparison. Hence * > foo* > ??? > + * fo? > foo. + * @see Comparable#compareTo(Object) + */ + public int compareTo(StateTransition other) { + String value = other.pattern; + if (pattern.equals(value)) { + return 0; + } + int patternCount = StringUtils.countOccurrencesOf(pattern, "*"); + int valueCount = StringUtils.countOccurrencesOf(value, "*"); + if (patternCount > valueCount) { + return 1; + } + if (patternCount < valueCount) { + return -1; + } + patternCount = StringUtils.countOccurrencesOf(pattern, "?"); + valueCount = StringUtils.countOccurrencesOf(value, "?"); + if (patternCount > valueCount) { + return 1; + } + if (patternCount < valueCount) { + return -1; + } + return pattern.compareTo(value); + } + + /* + * (non-Javadoc) + * + * @see java.lang.Object#toString() + */ + @Override + public String toString() { + return String.format("StateTransition: state=%s, pattern=%s, next=%s", state == null ? null : state.getName(), + pattern, next); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/BasicFlowTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/BasicFlowTests.java new file mode 100644 index 000000000..34105d5b5 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/BasicFlowTests.java @@ -0,0 +1,223 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; + +import org.junit.Test; +import org.springframework.batch.flow.FlowExecution; +import org.springframework.batch.flow.FlowExecutionException; +import org.springframework.batch.flow.SimpleFlow; +import org.springframework.batch.flow.StateTransition; + +/** + * @author Dave Syer + * + */ +public class BasicFlowTests { + + private SimpleFlow flow = new SimpleFlow("job"); + + private String executor = "data"; + + @Test(expected = IllegalArgumentException.class) + public void testEmptySteps() throws Exception { + flow.setStateTransitions(Collections.> emptySet()); + flow.afterPropertiesSet(); + } + + @Test(expected = IllegalArgumentException.class) + public void testNoNextStepSpecified() throws Exception { + flow.setStateTransitions(Collections.singleton(StateTransition.createStateTransition(new StateSupport( + "step"), "foo"))); + flow.afterPropertiesSet(); + } + + @Test(expected = IllegalArgumentException.class) + public void testNoStartStep() throws Exception { + flow.setStateTransitions(collect(StateTransition.createStateTransition(new StateSupport("step"), + FlowExecution.FAILED, "step"), StateTransition + .createEndStateTransition(new StateSupport("step")))); + flow.afterPropertiesSet(); + } + + @Test(expected = IllegalArgumentException.class) + public void testNoEndStep() throws Exception { + flow.setStateTransitions(Collections.singleton(StateTransition.createStateTransition(new StateSupport( + "step"), FlowExecution.FAILED, "step"))); + flow.setStartStateName("step"); + flow.afterPropertiesSet(); + } + + @Test(expected = IllegalArgumentException.class) + public void testMultipleStartSteps() throws Exception { + flow.setStateTransitions(collect(StateTransition.createEndStateTransition(new StubState("step1")), + StateTransition.createEndStateTransition(new StubState("step2")))); + flow.afterPropertiesSet(); + } + + @Test + public void testNoMatchForNextStep() throws Exception { + flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step1"), "FOO", "step2"), + StateTransition.createEndStateTransition(new StubState("step2")))); + flow.afterPropertiesSet(); + try { + flow.start(executor); + fail("Expected JobExecutionException"); + } + catch (FlowExecutionException e) { + // expected + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.toLowerCase().contains("next state not found")); + } + } + + @Test + public void testOneStep() throws Exception { + flow.setStateTransitions(Collections + .singleton(StateTransition.createEndStateTransition(new StubState("step1")))); + flow.afterPropertiesSet(); + FlowExecution execution = flow.start(executor); + assertEquals(FlowExecution.COMPLETED, execution.getStatus()); + assertEquals("step1", execution.getName()); + } + + @Test + public void testExplicitStartStep() throws Exception { + flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step"), + FlowExecution.FAILED, "step"), StateTransition.createEndStateTransition(new StubState("step")))); + flow.setStartStateName("step"); + flow.afterPropertiesSet(); + FlowExecution execution = flow.start(executor); + assertEquals(FlowExecution.COMPLETED, execution.getStatus()); + assertEquals("step", execution.getName()); + } + + @Test + public void testTwoSteps() throws Exception { + flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step1"), "step2"), + StateTransition.createEndStateTransition(new StubState("step2")))); + flow.afterPropertiesSet(); + FlowExecution execution = flow.start(executor); + assertEquals(FlowExecution.COMPLETED, execution.getStatus()); + assertEquals("step2", execution.getName()); + } + + @Test + public void testResume() throws Exception { + flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step1"), "step2"), + StateTransition.createEndStateTransition(new StubState("step2")))); + flow.afterPropertiesSet(); + FlowExecution execution = flow.resume("step2", executor); + assertEquals(FlowExecution.COMPLETED, execution.getStatus()); + assertEquals("step2", execution.getName()); + } + + @Test + public void testFailedStep() throws Exception { + flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step1") { + @Override + public String handle(String executor) { + return FlowExecution.FAILED; + } + }, "step2"), StateTransition.createEndStateTransition(new StubState("step2")))); + flow.afterPropertiesSet(); + FlowExecution execution = flow.start(executor); + assertEquals(FlowExecution.COMPLETED, execution.getStatus()); + assertEquals("step2", execution.getName()); + } + + @Test + public void testBranching() throws Exception { + flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step1"), "step2"), + StateTransition.createStateTransition(new StubState("step1"), FlowExecution.COMPLETED, "step3"), + StateTransition.createEndStateTransition(new StubState("step2")), StateTransition + .createEndStateTransition(new StubState("step3")))); + flow.afterPropertiesSet(); + FlowExecution execution = flow.start(executor); + assertEquals(FlowExecution.COMPLETED, execution.getStatus()); + assertEquals("step3", execution.getName()); + } + + @Test + public void testPause() throws Exception { + flow.setStateTransitions(collect(StateTransition.createStateTransition(new StubState("step1"), "step2"), + StateTransition.createStateTransition(new StubState("step2") { + private boolean paused = false; + + @Override + public String handle(String executor) throws Exception { + if (!paused) { + paused = true; + return FlowExecution.PAUSED; + } + paused = false; + return FlowExecution.COMPLETED; + } + + }, "step3"), StateTransition.createEndStateTransition(new StubState("step3")))); + flow.afterPropertiesSet(); + FlowExecution execution = flow.start(executor); + assertEquals(FlowExecution.PAUSED, execution.getStatus()); + assertEquals("step2", execution.getName()); + execution = flow.resume(execution.getName(), executor); + assertEquals(FlowExecution.COMPLETED, execution.getStatus()); + assertEquals("step3", execution.getName()); + } + + private Collection> collect(StateTransition s1, StateTransition s2) { + Collection> list = new ArrayList>(); + list.add(s1); + list.add(s2); + return list; + } + + private Collection> collect(StateTransition s1, StateTransition s2, + StateTransition s3) { + Collection> list = collect(s1, s2); + list.add(s3); + return list; + } + + private Collection> collect(StateTransition s1, StateTransition s2, + StateTransition s3, StateTransition s4) { + Collection> list = collect(s1, s2, s3); + list.add(s4); + return list; + } + + /** + * @author Dave Syer + * + */ + private static class StubState extends StateSupport { + + /** + * @param string + */ + public StubState(String string) { + super(string); + } + + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/FlowExecutionTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/FlowExecutionTests.java new file mode 100644 index 000000000..22a556f3f --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/FlowExecutionTests.java @@ -0,0 +1,77 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.springframework.batch.flow.FlowExecution; + +/** + * @author Dave Syer + * + */ +public class FlowExecutionTests { + + @Test + public void testBasicProperties() throws Exception { + FlowExecution execution = new FlowExecution("foo", "BAR"); + assertEquals("foo",execution.getName()); + assertEquals("BAR",execution.getStatus()); + } + + @Test + public void testAlphaOrdering() throws Exception { + FlowExecution first = new FlowExecution("foo", "BAR"); + FlowExecution second = new FlowExecution("foo", "SPAM"); + assertTrue("Should be negative",first.compareTo(second)<0); + assertTrue("Should be positive",second.compareTo(first)>0); + } + + @Test + public void testEnumOrdering() throws Exception { + FlowExecution first = new FlowExecution("foo", FlowExecution.COMPLETED); + FlowExecution second = new FlowExecution("foo", FlowExecution.FAILED); + assertTrue("Should be negative",first.compareTo(second)<0); + assertTrue("Should be positive",second.compareTo(first)>0); + } + + @Test + public void testEnumStartsWithOrdering() throws Exception { + FlowExecution first = new FlowExecution("foo", "COMPLETED.BAR"); + FlowExecution second = new FlowExecution("foo", "FAILED.FOO"); + assertTrue("Should be negative",first.compareTo(second)<0); + assertTrue("Should be positive",second.compareTo(first)>0); + } + + @Test + public void testEnumStartsWithAlphaOrdering() throws Exception { + FlowExecution first = new FlowExecution("foo", "COMPLETED.BAR"); + FlowExecution second = new FlowExecution("foo", "COMPLETED.FOO"); + assertTrue("Should be negative",first.compareTo(second)<0); + assertTrue("Should be positive",second.compareTo(first)>0); + } + + @Test + public void testEnumAndAlpha() throws Exception { + FlowExecution first = new FlowExecution("foo", "ZZZZZ"); + FlowExecution second = new FlowExecution("foo", "FAILED.FOO"); + assertTrue("Should be negative",first.compareTo(second)<0); + assertTrue("Should be positive",second.compareTo(first)>0); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/SimpleFlowExecutionAggregatorTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/SimpleFlowExecutionAggregatorTests.java new file mode 100644 index 000000000..8edc3fc6e --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/SimpleFlowExecutionAggregatorTests.java @@ -0,0 +1,50 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.Collections; + +import org.junit.Test; +import org.springframework.batch.flow.FlowExecution; +import org.springframework.batch.flow.MaxValueFlowExecutionAggregator; + +/** + * @author Dave Syer + * + */ +public class SimpleFlowExecutionAggregatorTests { + + private MaxValueFlowExecutionAggregator aggregator = new MaxValueFlowExecutionAggregator(); + + @Test + public void testFailed() throws Exception { + FlowExecution first = new FlowExecution("foo", FlowExecution.COMPLETED); + FlowExecution second = new FlowExecution("foo", FlowExecution.FAILED); + assertTrue("Should be negative", first.compareTo(second)<0); + assertTrue("Should be positive", second.compareTo(first)>0); + assertEquals(FlowExecution.FAILED, aggregator.aggregate(Arrays.asList(first, second))); + } + + @Test + public void testEmpty() throws Exception { + assertEquals(FlowExecution.UNKNOWN, aggregator.aggregate(Collections. emptySet())); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/SplitStateTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/SplitStateTests.java new file mode 100644 index 000000000..8302482dd --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/SplitStateTests.java @@ -0,0 +1,86 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.Collection; + +import org.easymock.EasyMock; +import org.junit.Test; +import org.springframework.batch.flow.Flow; +import org.springframework.batch.flow.FlowExecution; +import org.springframework.batch.flow.SplitState; +import org.springframework.core.task.SimpleAsyncTaskExecutor; + + +/** + * @author Dave Syer + * + */ +public class SplitStateTests { + + @Test + public void testBasicHandling() throws Exception { + + Collection> flows = new ArrayList>(); + @SuppressWarnings("unchecked") + Flow flow1 = EasyMock.createMock(Flow.class); + @SuppressWarnings("unchecked") + Flow flow2 = EasyMock.createMock(Flow.class); + flows.add(flow1); + flows.add(flow2); + + SplitState state = new SplitState("foo", flows); + + EasyMock.expect(flow1.start(null)).andReturn(new FlowExecution("step1", FlowExecution.COMPLETED)); + EasyMock.expect(flow2.start(null)).andReturn(new FlowExecution("step1", FlowExecution.COMPLETED)); + EasyMock.replay(flow1, flow2); + + String result = state.handle(null); + assertEquals(FlowExecution.COMPLETED, result); + + EasyMock.verify(flow1, flow2); + + } + + @Test + public void testConcurrentHandling() throws Exception { + + Collection> flows = new ArrayList>(); + @SuppressWarnings("unchecked") + Flow flow1 = EasyMock.createMock(Flow.class); + @SuppressWarnings("unchecked") + Flow flow2 = EasyMock.createMock(Flow.class); + flows.add(flow1); + flows.add(flow2); + + SplitState state = new SplitState("foo", flows); + state.setTaskExecutor(new SimpleAsyncTaskExecutor()); + + EasyMock.expect(flow1.start(null)).andReturn(new FlowExecution("step1", FlowExecution.COMPLETED)); + EasyMock.expect(flow2.start(null)).andReturn(new FlowExecution("step1", FlowExecution.COMPLETED)); + EasyMock.replay(flow1, flow2); + + String result = state.handle(null); + assertEquals(FlowExecution.COMPLETED, result); + + EasyMock.verify(flow1, flow2); + + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/StateSupport.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/StateSupport.java new file mode 100644 index 000000000..f1289c27a --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/StateSupport.java @@ -0,0 +1,41 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import org.springframework.batch.flow.AbstractState; +import org.springframework.batch.flow.FlowExecution; + +/** + * Base class for {@link State} implementations. + * + * @author Dave Syer + * + */ +public class StateSupport extends AbstractState { + + /** + * @param name + */ + public StateSupport(String name) { + super(name); + } + + @Override + public String handle(T context) throws Exception { + return FlowExecution.COMPLETED; + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/StateTransitionTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/StateTransitionTests.java new file mode 100644 index 000000000..e3e18b5fa --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/flow/StateTransitionTests.java @@ -0,0 +1,130 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * 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 org.springframework.batch.flow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.springframework.batch.flow.StateTransition; + +/** + * @author Dave Syer + * + */ +public class StateTransitionTests { + + @Test + public void testIsEnd() { + StateTransition transition = StateTransition.createEndStateTransition(null, ""); + assertTrue(transition.isEnd()); + assertNull(transition.getNext()); + } + + @Test + public void testMatchesStar() { + StateTransition transition = StateTransition.createStateTransition(null, "*", "start"); + assertTrue(transition.matches("CONTINUABLE")); + } + + @Test + public void testMatchesNull() { + StateTransition transition = StateTransition.createStateTransition(null, null, "start"); + assertTrue(transition.matches("CONTINUABLE")); + } + + @Test + public void testMatchesEmpty() { + StateTransition transition = StateTransition.createStateTransition(null, "", "start"); + assertTrue(transition.matches("CONTINUABLE")); + } + + @Test + public void testMatchesExact() { + StateTransition transition = StateTransition.createStateTransition(null, "CONTINUABLE", "start"); + assertTrue(transition.matches("CONTINUABLE")); + } + + @Test + public void testMatchesWildcard() { + StateTransition transition = StateTransition.createStateTransition(null, "CONTIN*", "start"); + assertTrue(transition.matches("CONTINUABLE")); + } + + @Test + public void testMatchesPlaceholder() { + StateTransition transition = StateTransition.createStateTransition(null, "CONTIN???LE", "start"); + assertTrue(transition.matches("CONTINUABLE")); + } + + @Test + public void testSimpleOrderingEqual() { + StateTransition transition = StateTransition.createStateTransition(null, "CONTIN???LE", "start"); + assertEquals(0, transition.compareTo(transition)); + } + + @Test + public void testSimpleOrderingMoreGeneral() { + StateTransition transition = StateTransition.createStateTransition(null, "CONTIN???LE", "start"); + StateTransition other = StateTransition.createStateTransition(null, "CONTINUABLE", "start"); + assertEquals(1, transition.compareTo(other)); + assertEquals(-1, other.compareTo(transition)); + } + + @Test + public void testSimpleOrderingMostGeneral() { + StateTransition transition = StateTransition.createStateTransition(null, "*", "start"); + StateTransition other = StateTransition.createStateTransition(null, "CONTINUABLE", "start"); + assertEquals(1, transition.compareTo(other)); + assertEquals(-1, other.compareTo(transition)); + } + + @Test + public void testSubstringAndWildcard() { + StateTransition transition = StateTransition.createStateTransition(null, "CONTIN*", "start"); + StateTransition other = StateTransition.createStateTransition(null, "CONTINUABLE", "start"); + assertEquals(1, transition.compareTo(other)); + assertEquals(-1, other.compareTo(transition)); + } + + @Test + public void testSimpleOrderingMostToNextGeneral() { + StateTransition transition = StateTransition.createStateTransition(null, "*", "start"); + StateTransition other = StateTransition.createStateTransition(null, "C?", "start"); + assertEquals(1, transition.compareTo(other)); + assertEquals(-1, other.compareTo(transition)); + } + + @Test + public void testSimpleOrderingAdjacent() { + StateTransition transition = StateTransition.createStateTransition(null, "CON*", "start"); + StateTransition other = StateTransition.createStateTransition(null, "CON?", "start"); + assertEquals(1, transition.compareTo(other)); + assertEquals(-1, other.compareTo(transition)); + } + + @Test + public void testToString() { + StateTransition transition = StateTransition.createStateTransition(null, "CONTIN???LE", "start"); + String string = transition.toString(); + assertTrue("Wrong string: " + string, string.contains("Transition")); + assertTrue("Wrong string: " + string, string.contains("start")); + assertTrue("Wrong string: " + string, string.contains("CONTIN???LE")); + assertTrue("Wrong string: " + string, string.contains("next=")); + } + +}