RESOLVED - issue BATCH-1157: Add max count parameter to counting item readers

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStreamReader.java

@@ -0,0 +1,26 @@
+/*
+ * 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.item;
+
+/**
+ * Convenience interface that combines {@link ItemStream} and {@link ItemReader}
+ * .
+ * @author Dave Syer
+ * 
+ */
+public interface ItemStreamReader<T> extends ItemStream, ItemReader<T> {
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java

@@ -38,8 +38,12 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
 
 	private static final String READ_COUNT = "read.count";
 
+	private static final String READ_COUNT_MAX = "read.count.max";
+
 	private int currentItemCount = 0;
 
+	private int maxItemCount = Integer.MAX_VALUE;
+
 	private ExecutionContextUserSupport ecSupport = new ExecutionContextUserSupport();
 
 	private boolean saveState = true;
@@ -68,11 +72,14 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
 	 */
 	protected void jumpToItem(int itemIndex) throws Exception {
 		for (int i = 0; i < itemIndex; i++) {
-			doRead();
+			read();
 		}
 	}
 
-	public T read() throws Exception, UnexpectedInputException, ParseException {
+	public final T read() throws Exception, UnexpectedInputException, ParseException {
+		if (currentItemCount >= maxItemCount-1) {
+			return null;
+		}
 		currentItemCount++;
 		return doRead();
 	}
@@ -104,17 +111,23 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
 			throw new ItemStreamException("Failed to initialize the reader", e);
 		}
 
+		if (executionContext.containsKey(ecSupport.getKey(READ_COUNT_MAX))) {
+			maxItemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT_MAX));
+		}
+
 		if (executionContext.containsKey(ecSupport.getKey(READ_COUNT))) {
 			int itemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT));
 
-			try {
-				jumpToItem(itemCount);
+			if (itemCount < maxItemCount) {
+				try {
+					jumpToItem(itemCount);
+				}
+				catch (Exception e) {
+					throw new ItemStreamException("Could not move to stored position on restart", e);
+				}
 			}
-			catch (Exception e) {
-				throw new ItemStreamException("Could not move to stored position on restart", e);
-			}
-
 			currentItemCount = itemCount;
+
 		}
 
 	}
@@ -123,6 +136,9 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements Ite
 		if (saveState) {
 			Assert.notNull(executionContext, "ExecutionContext must not be null");
 			executionContext.putInt(ecSupport.getKey(READ_COUNT), currentItemCount);
+			if (maxItemCount < Integer.MAX_VALUE) {
+				executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX), maxItemCount);
+			}
 		}
 
 	}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemStream.java

@@ -42,6 +42,19 @@ public class CompositeItemStream implements ItemStream {
 		this.streams = Arrays.asList(listeners);
 	}
 
+	/**
+	 * Register a {@link ItemStream} as one of the interesting providers under
+	 * the provided key.
+	 * 
+	 */
+	public void register(ItemStream stream) {
+		synchronized (streams) {
+			if (!streams.contains(stream)) {
+				streams.add(stream);
+			}
+		}
+	}
+
 	/**
 	 * 
 	 */
@@ -63,19 +76,6 @@ public class CompositeItemStream implements ItemStream {
 		}
 	}
 
-	/**
-	 * Register a {@link ItemStream} as one of the interesting providers under
-	 * the provided key.
-	 * 
-	 */
-	public void register(ItemStream stream) {
-		synchronized (streams) {
-			if (!streams.contains(stream)) {
-				streams.add(stream);
-			}
-		}
-	}
-
 	/**
 	 * Broadcast the call to close.
 	 * @throws ItemStreamException

spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/MethodArgumentsKeyGenerator.java

@@ -15,11 +15,9 @@
  */
 package org.springframework.batch.retry.interceptor;
 
-import org.springframework.batch.item.ItemReader;
-
 /**
- * Extension of the {@link ItemReader} interface that allows items to be
- * identified and tagged by a unique key.
+ * Interface that allows method parameters to be identified and tagged by a
+ * unique key.
  * 
  * @author Dave Syer
  * 

spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/DefaultRetryState.java

@@ -44,6 +44,7 @@ public class DefaultRetryState implements RetryState {
 	 * @param key the key for the state to allow this retry attempt to be
 	 * recognised
 	 * @param forceRefresh true if the attempt is known to be a brand new state
+	 * (could not have previously failed)
 	 * @param rollbackClassifier the rollback classifier to set. The rollback
 	 * classifier answers true if the exception provided should cause a
 	 * rollback.
@@ -55,7 +56,7 @@ public class DefaultRetryState implements RetryState {
 	}
 
 	/**
-	 * Defaults the rollback classifier to null.
+	 * Defaults the force refresh flag to false.
 	 * @see DefaultRetryState#DefaultRetryState(Object, boolean, Classifier)
 	 */
 	public DefaultRetryState(Object key, Classifier<? super Throwable, Boolean> rollbackClassifier) {
@@ -80,22 +81,30 @@ public class DefaultRetryState implements RetryState {
 		this(key, false, null);
 	}
 
-	/* (non-Javadoc)
+	/*
+	 * (non-Javadoc)
+	 * 
 	 * @see org.springframework.batch.retry.IRetryState#getKey()
 	 */
 	public Object getKey() {
 		return key;
 	}
 
-	/* (non-Javadoc)
+	/*
+	 * (non-Javadoc)
+	 * 
 	 * @see org.springframework.batch.retry.IRetryState#isForceRefresh()
 	 */
 	public boolean isForceRefresh() {
 		return forceRefresh;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.batch.retry.IRetryState#rollbackFor(java.lang.Exception)
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see
+	 * org.springframework.batch.retry.IRetryState#rollbackFor(java.lang.Exception
+	 * )
 	 */
 	public boolean rollbackFor(Exception exception) {
 		if (rollbackClassifier == null) {