diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/ClassPathXmlApplicationContextJobFactory.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/ClassPathXmlApplicationContextJobFactory.java index b61633602..44c49cacf 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/ClassPathXmlApplicationContextJobFactory.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/ClassPathXmlApplicationContextJobFactory.java @@ -34,15 +34,17 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; */ public class ClassPathXmlApplicationContextJobFactory implements JobFactory { - private String beanName; + final private String beanName; - private String path; + final private String path; - private ApplicationContext parent; + final private ApplicationContext parent; /** - * @param beanName - * @param path + * @param beanName the id of the {@link Job} in the application context to + * be created + * @param path the path to the XML configuration containing the {@link Job} + * @param parent the application context to use as a parent (or null) */ public ClassPathXmlApplicationContextJobFactory(String beanName, String path, ApplicationContext parent) { super(); @@ -73,14 +75,16 @@ public class ClassPathXmlApplicationContextJobFactory implements JobFactory { public String getJobName() { return beanName; } - + /** * @author Dave Syer - * + * */ private static class ContextClosingJob implements Job { private Job delegate; + private ConfigurableApplicationContext context; + /** * @param delegate * @param context @@ -90,6 +94,7 @@ public class ClassPathXmlApplicationContextJobFactory implements JobFactory { this.delegate = delegate; this.context = context; } + /** * @param execution * @throws JobExecutionException @@ -98,22 +103,26 @@ public class ClassPathXmlApplicationContextJobFactory implements JobFactory { public void execute(JobExecution execution) throws JobExecutionException { try { delegate.execute(execution); - } finally { + } + finally { context.close(); } } + /** * @see org.springframework.batch.core.Job#getName() */ public String getName() { return delegate.getName(); } + /** * @see org.springframework.batch.core.Job#getSteps() */ public List getSteps() { return delegate.getSteps(); } + /** * @see org.springframework.batch.core.Job#isRestartable() */ @@ -121,6 +130,6 @@ public class ClassPathXmlApplicationContextJobFactory implements JobFactory { return delegate.isRestartable(); } - } + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/resource/StepExecutionSimpleCompletionPolicy.java b/spring-batch-core/src/main/java/org/springframework/batch/core/resource/StepExecutionSimpleCompletionPolicy.java new file mode 100644 index 000000000..b3dd29846 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/resource/StepExecutionSimpleCompletionPolicy.java @@ -0,0 +1,126 @@ +/* + * 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.resource; + +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.listener.StepExecutionListenerSupport; +import org.springframework.batch.repeat.CompletionPolicy; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.util.Assert; + +/** + * A {@link CompletionPolicy} that picks up a commit interval from + * {@link JobParameters} by listening to the start of a step. Use anywhere that + * a {@link CompletionPolicy} can be used (usually at the chunk level in a + * step), and inject as a {@link StepExecutionListener} into the surrounding + * step. N.B. only after the step has started will the completion policy be + * usable. + * + * @author Dave Syer + * + * @see CompletionPolicy + */ +public class StepExecutionSimpleCompletionPolicy extends StepExecutionListenerSupport implements CompletionPolicy { + + private CompletionPolicy delegate; + + private String keyName = "commit.interval"; + + /** + * Public setter for the key name of a Long value in the + * {@link JobParameters} that will contain a commit interval. Defaults to + * "commit.interval". + * @param keyName the keyName to set + */ + public void setKeyName(String keyName) { + this.keyName = keyName; + } + + /** + * Set up a {@link SimpleCompletionPolicy} with a commit interval taken from + * the {@link JobParameters}. If there is a Long parameter with the given + * key name, the intValue of this parameter is used. If not an exception + * will be thrown. + * + * @see org.springframework.batch.core.listener.StepExecutionListenerSupport#beforeStep(org.springframework.batch.core.StepExecution) + */ + public void beforeStep(StepExecution stepExecution) { + JobParameters jobParameters = stepExecution.getJobParameters(); + Assert.state(jobParameters.getLongParameters().containsKey(keyName), + "JobParameters do not contain Long parameter with key=[" + keyName + "]"); + delegate = new SimpleCompletionPolicy(jobParameters.getLong(keyName).intValue()); + } + + /** + * @param context + * @param result + * @return true if the commit interval has been reached or the result + * indicates completion + * @see org.springframework.batch.repeat.CompletionPolicy#isComplete(org.springframework.batch.repeat.RepeatContext, + * org.springframework.batch.repeat.ExitStatus) + */ + public boolean isComplete(RepeatContext context, ExitStatus result) { + Assert.state(delegate != null, "The delegate resource has not been initialised. " + + "Remember to register this object as a StepListener."); + return delegate.isComplete(context, result); + } + + /** + * @param context + * @return if the commit interval has been reached + * @see org.springframework.batch.repeat.CompletionPolicy#isComplete(org.springframework.batch.repeat.RepeatContext) + */ + public boolean isComplete(RepeatContext context) { + Assert.state(delegate != null, "The delegate resource has not been initialised. " + + "Remember to register this object as a StepListener."); + return delegate.isComplete(context); + } + + /** + * @param parent + * @return a new {@link RepeatContext} + * @see org.springframework.batch.repeat.CompletionPolicy#start(org.springframework.batch.repeat.RepeatContext) + */ + public RepeatContext start(RepeatContext parent) { + Assert.state(delegate != null, "The delegate resource has not been initialised. " + + "Remember to register this object as a StepListener."); + return delegate.start(parent); + } + + /** + * @param context + * @see org.springframework.batch.repeat.CompletionPolicy#update(org.springframework.batch.repeat.RepeatContext) + */ + public void update(RepeatContext context) { + Assert.state(delegate != null, "The delegate resource has not been initialised. " + + "Remember to register this object as a StepListener."); + delegate.update(context); + } + + /** + * Delegates to the wrapped {@link CompletionPolicy} if set, otherwise + * returns the value of {@link #setKeyName(String)}. + */ + public String toString() { + return (delegate == null) ? keyName : delegate.toString(); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index 47b509f1c..cfdacfc43 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -15,12 +15,15 @@ */ package org.springframework.batch.core.step.item; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.StepListener; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.repeat.CompletionPolicy; import org.springframework.batch.repeat.exception.DefaultExceptionHandler; import org.springframework.batch.repeat.exception.ExceptionHandler; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; @@ -44,7 +47,11 @@ import org.springframework.util.Assert; */ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { - private int commitInterval = 1; + protected final Log logger = LogFactory.getLog(getClass()); + + private static final int DEFAULT_COMMIT_INTERVAL = 1; + + private int commitInterval = 0; private ItemStream[] streams = new ItemStream[0]; @@ -55,21 +62,35 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { private ItemHandler itemHandler; private RepeatTemplate stepOperations; - + private RepeatTemplate chunkOperations; private ExceptionHandler exceptionHandler = new DefaultExceptionHandler(); - + private CompletionPolicy chunkCompletionPolicy; + /** - * Set the commit interval. + * Set the commit interval. Either set this or the chunkCompletionPolicy but + * not both. * * @param commitInterval 1 by default */ public void setCommitInterval(int commitInterval) { this.commitInterval = commitInterval; } - + + /** + * Public setter for the {@link CompletionPolicy} applying to the chunk + * level. A transaction will be committed when this policy decides to + * complete. Defaults to a {@link SimpleCompletionPolicy} with chunk size + * equal to the commitInterval property. + * + * @param chunkCompletionPolicy the chunkCompletionPolicy to set + */ + public void setChunkCompletionPolicy(CompletionPolicy chunkCompletionPolicy) { + this.chunkCompletionPolicy = chunkCompletionPolicy; + } + /** * The streams to inject into the {@link Step}. Any instance of * {@link ItemStream} can be used, and will then receive callbacks at the @@ -108,7 +129,7 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { protected RepeatTemplate getStepOperations() { return stepOperations; } - + /** * Protected getter for the chunk operations to make them available in * subclasses. @@ -118,7 +139,6 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { return chunkOperations; } - /** * Public setter for the SimpleLimitExceptionHandler. * @param exceptionHandler the exceptionHandler to set @@ -168,8 +188,6 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { protected void applyConfiguration(ItemOrientedStep step) { super.applyConfiguration(step); - - Assert.isTrue(commitInterval > 0); step.setStreams(streams); @@ -195,7 +213,7 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper(); chunkOperations = new RepeatTemplate(); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); + chunkOperations.setCompletionPolicy(getChunkCompletionPolicy()); helper.addChunkListeners(chunkOperations, listeners); step.setChunkOperations(chunkOperations); @@ -226,4 +244,24 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { } + /** + * @return a {@link CompletionPolicy} consistent with the commit interval + * and injected policy (if present). + */ + private CompletionPolicy getChunkCompletionPolicy() { + Assert.state(!(chunkCompletionPolicy != null && commitInterval != 0), + "You must specify either a chunkCompletionPolicy or a commitInterval but not both."); + Assert.state(commitInterval >= 0, + "The commitInterval must be positive or zero (for default value)."); + + if (chunkCompletionPolicy != null) { + return chunkCompletionPolicy; + } + if (commitInterval == 0) { + logger.info("Setting commit interval to default value (" + DEFAULT_COMMIT_INTERVAL + ")"); + commitInterval = DEFAULT_COMMIT_INTERVAL; + } + return new SimpleCompletionPolicy(commitInterval); + } + } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/ClassPathXmlApplicationContextJobFactoryTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/ClassPathXmlApplicationContextJobFactoryTests.java index 240e36c06..d7ed20b6a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/ClassPathXmlApplicationContextJobFactoryTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/ClassPathXmlApplicationContextJobFactoryTests.java @@ -15,17 +15,17 @@ */ package org.springframework.batch.core.configuration.support; -import org.springframework.util.ClassUtils; - import junit.framework.TestCase; +import org.springframework.util.ClassUtils; + /** * @author Dave Syer * */ public class ClassPathXmlApplicationContextJobFactoryTests extends TestCase { - private ClassPathXmlApplicationContextJobFactory factory = new ClassPathXmlApplicationContextJobFactory("test-job", ClassUtils.addResourcePathToPackagePath(getClass(), "test-context.xml"), null); + private ClassPathXmlApplicationContextJobFactory factory = new ClassPathXmlApplicationContextJobFactory("test-job", ClassUtils.addResourcePathToPackagePath(getClass(), "trivial-context.xml"), null); /** * Test method for {@link org.springframework.batch.core.configuration.support.ClassPathXmlApplicationContextJobFactory#createJob()}. diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/resource/StepExecutionSimpleCompletionPolicyTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/resource/StepExecutionSimpleCompletionPolicyTests.java new file mode 100644 index 000000000..fa2ed1d6e --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/resource/StepExecutionSimpleCompletionPolicyTests.java @@ -0,0 +1,73 @@ +/* + * 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.resource; + +import java.io.IOException; + +import junit.framework.TestCase; + +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.JobParametersBuilder; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.step.StepSupport; +import org.springframework.batch.repeat.RepeatContext; + +/** + * Unit tests for {@link StepExecutionSimpleCompletionPolicy} + * + * @author Dave Syer + */ +public class StepExecutionSimpleCompletionPolicyTests extends TestCase { + + /** + * Object under test + */ + private StepExecutionSimpleCompletionPolicy policy = new StepExecutionSimpleCompletionPolicy(); + + private JobInstance jobInstance; + + private StepExecution stepExecution; + + /** + * mock step context + */ + + protected void setUp() throws Exception { + + JobParameters jobParameters = new JobParametersBuilder().addLong("commit.interval", new Long(2L)).toJobParameters(); + jobInstance = new JobInstance(new Long(0), jobParameters, "testJob"); + JobExecution jobExecution = new JobExecution(jobInstance); + Step step = new StepSupport("bar"); + stepExecution = jobExecution.createStepExecution(step); + policy.beforeStep(stepExecution); + + } + + public void testToString() throws Exception { + String msg = policy.toString(); + assertTrue("String does not contain chunk size", msg.indexOf("chunkSize=2")>=0); + } + + public void testKeyName() throws Exception, IOException { + RepeatContext context = policy.start(null); + assertFalse(policy.isComplete(context)); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java index f16683c31..8c6fb59c0 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java @@ -42,6 +42,7 @@ import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.ListItemReader; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.exception.ExceptionHandler; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; @@ -241,24 +242,34 @@ public class SimpleStepFactoryBeanTests extends TestCase { // nothing wrong here factory.getObject(); - // but exception excpected after setting commit interval to value <= 0 - factory.setCommitInterval(0); - try { - factory.getObject(); - fail(); - } - catch (IllegalArgumentException e) { - // expected - } - + // but exception expected after setting commit interval to value < 0 factory.setCommitInterval(-1); try { factory.getObject(); fail(); } - catch (IllegalArgumentException e) { + catch (IllegalStateException e) { // expected } } + /** + * Commit interval specified is not allowed to be zero or negative. + * @throws Exception + */ + public void testCommitIntervalAndCompletionPolicyBothSet() throws Exception { + SimpleStepFactoryBean factory = getStepFactory("foo"); + + // but exception expected after setting commit interval and completion policy + factory.setCommitInterval(1); + factory.setChunkCompletionPolicy(new SimpleCompletionPolicy(2)); + try { + factory.getObject(); + fail(); + } + catch (IllegalStateException e) { + // expected + } + + } } diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/trivial-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/trivial-context.xml new file mode 100644 index 000000000..054b4c96d --- /dev/null +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/trivial-context.xml @@ -0,0 +1,15 @@ + + + + + + diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/policy/SimpleCompletionPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/policy/SimpleCompletionPolicy.java index aeec5dd36..e8e68cf71 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/policy/SimpleCompletionPolicy.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/policy/SimpleCompletionPolicy.java @@ -20,6 +20,7 @@ import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.context.RepeatContextSupport; import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.util.ClassUtils; /** * Policy for terminating a batch after a fixed number of operations. Internal @@ -103,5 +104,12 @@ public class SimpleCompletionPolicy extends DefaultResultCompletionPolicy { return getStartedCount() >= chunkSize; } } + + /* (non-Javadoc) + * @see java.lang.Object#toString() + */ + public String toString() { + return ClassUtils.getShortName(SimpleCompletionPolicy.class)+": chunkSize="+chunkSize; + } } diff --git a/spring-batch-samples/src/main/resources/adhoc-job-launcher-context.xml b/spring-batch-samples/src/main/resources/adhoc-job-launcher-context.xml index 407dee9ee..ca1fff431 100644 --- a/spring-batch-samples/src/main/resources/adhoc-job-launcher-context.xml +++ b/spring-batch-samples/src/main/resources/adhoc-job-launcher-context.xml @@ -45,6 +45,7 @@ + diff --git a/spring-batch-samples/src/main/resources/quartz-job-launcher-context.xml b/spring-batch-samples/src/main/resources/quartz-job-launcher-context.xml index 9999996e5..2bb8303c5 100644 --- a/spring-batch-samples/src/main/resources/quartz-job-launcher-context.xml +++ b/spring-batch-samples/src/main/resources/quartz-job-launcher-context.xml @@ -30,6 +30,8 @@ + + diff --git a/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml b/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml index 4a332dae6..3ca868d02 100644 --- a/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml +++ b/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml @@ -9,17 +9,10 @@ - - - - - -