diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStreamReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStreamReader.java new file mode 100644 index 000000000..9f04fd6f6 --- /dev/null +++ b/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 extends ItemStream, ItemReader { + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java index 4b0384ccd..b145539c6 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java @@ -38,8 +38,12 @@ public abstract class AbstractItemCountingItemStreamItemReader 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 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 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 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); + } } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemStream.java index 0acebc4b9..7568718ba 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemStream.java +++ b/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 diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/MethodArgumentsKeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/MethodArgumentsKeyGenerator.java index 801a8e849..ddbae85c6 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/MethodArgumentsKeyGenerator.java +++ b/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 * diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/DefaultRetryState.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/DefaultRetryState.java index 212457b25..cdb84c425 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/DefaultRetryState.java +++ b/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 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) { diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/ItemCountingItemStreamItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/ItemCountingItemStreamItemReaderTests.java new file mode 100644 index 000000000..3ad9b7bf5 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/ItemCountingItemStreamItemReaderTests.java @@ -0,0 +1,157 @@ +/* + * 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.support; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.Iterator; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; + +/** + * @author Dave Syer + * + */ +public class ItemCountingItemStreamItemReaderTests { + + private ItemCountingItemStreamItemReader reader = new ItemCountingItemStreamItemReader(); + + @Before + public void setUp() { + reader.setName("foo"); + } + + @Test + public void testJumpToItem() throws Exception { + reader.jumpToItem(2); + assertEquals(2, reader.getCurrentItemCount()); + reader.read(); + assertEquals(3, reader.getCurrentItemCount()); + } + + @Test + public void testGetCurrentItemCount() throws Exception { + assertEquals(0, reader.getCurrentItemCount()); + reader.read(); + assertEquals(1, reader.getCurrentItemCount()); + } + + @Test + public void testClose() { + reader.close(); + assertTrue(reader.closeCalled); + } + + @Test(expected=IllegalArgumentException.class) + public void testOpenWithoutName() { + reader = new ItemCountingItemStreamItemReader(); + reader.open(new ExecutionContext()); + assertFalse(reader.openCalled); + } + + @Test + public void testOpen() { + reader.open(new ExecutionContext()); + assertTrue(reader.openCalled); + } + + @Test + public void testReadToEnd() throws Exception { + reader.read(); + reader.read(); + reader.read(); + assertNull(reader.read()); + } + + @Test + public void testUpdate() throws Exception { + reader.read(); + ExecutionContext context = new ExecutionContext(); + reader.update(context); + assertEquals(1, context.size()); + assertEquals(1, context.getInt("foo.read.count")); + } + + @Test + public void testSetName() throws Exception { + reader.setName("bar"); + reader.read(); + ExecutionContext context = new ExecutionContext(); + reader.update(context); + assertEquals(1, context.getInt("bar.read.count")); + } + + @Test + public void testSetSaveState() throws Exception { + reader.read(); + ExecutionContext context = new ExecutionContext(); + reader.update(context); + assertEquals(1, context.size()); + } + + @Test + public void testReadToEndWithMax() throws Exception { + ExecutionContext context = new ExecutionContext(); + context.putInt("foo.read.count.max", 1); + reader.open(context); + reader.read(); + assertNull(reader.read()); + } + + @Test + public void testUpdateWithMax() throws Exception { + ExecutionContext context = new ExecutionContext(); + context.putInt("foo.read.count.max", 1); + reader.open(context); + reader.update(context); + assertEquals(2, context.size()); + } + + private static class ItemCountingItemStreamItemReader extends AbstractItemCountingItemStreamItemReader { + + private boolean closeCalled = false; + + private boolean openCalled = false; + + private Iterator items = Arrays.asList("a", "b", "c").iterator(); + + @Override + protected void doClose() throws Exception { + closeCalled = true; + } + + @Override + protected void doOpen() throws Exception { + openCalled = true; + } + + @Override + protected String doRead() throws Exception { + if (!items.hasNext()) { + return null; + } + return items.next(); + } + + } + +}