RESOLVED - issue BATCH-541: pull commit interval from the job parameters

Added StepExecutionSimpleCompletionPolicy (a la StepExecution*) as s one-off solution to the commit interval problem initially posed

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();
 		}
 
-	}	
+	}
 
 }

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();
+	}
+
+}

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);
+	}
+
 }