diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/PeekableItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/PeekableItemReader.java new file mode 100644 index 000000000..995fb64d1 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/PeekableItemReader.java @@ -0,0 +1,49 @@ +/* + * Copyright 2006-2010 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; + +/** + *

+ * A specialisation of {@link ItemReader} that allows the user to look ahead + * into the stream of items. This is useful, for instance, when reading flat + * file data that contains record separator lines which are actually part of the + * next record. + *

+ * + *

+ * The detailed contract for {@link #peek()} has to be defined by the + * implementation because there is no general way to define it in a concurrent + * environment. The definition of "the next read()" operation is tenuous if + * multiple clients are reading concurrently, and the ability to peek implies + * that some state is likely to be stored, so implementations of + * {@link PeekableItemReader} may well be restricted to single threaded use. + *

+ * + * @author Dave Syer + * + */ +public interface PeekableItemReader extends ItemReader { + + /** + * Get the next item that would be returned by {@link #read()}, without + * affecting the result of {@link #read()}. + * + * @return the next item + * @throws Exception if there is a problem + */ + T peek() throws Exception, UnexpectedInputException, ParseException; + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SingleItemPeekableItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SingleItemPeekableItemReader.java new file mode 100644 index 000000000..a57388f82 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SingleItemPeekableItemReader.java @@ -0,0 +1,155 @@ +/* + * Copyright 2006-2010 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 java.util.Map.Entry; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.ItemStreamReader; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.PeekableItemReader; +import org.springframework.batch.item.UnexpectedInputException; + +/** + *

+ * A {@link PeekableItemReader} that allows the user to peek one item ahead. + * Repeated calls to {@link #peek()} will return the same item, and this will be + * the next item returned from {@link #read()}. + *

+ * + *

+ * Intentionally not thread safe: it wouldn't be possible to honour the peek in + * multiple threads because only one of the threads that peeked would get that + * item in the next call to read. + *

+ * + * @author Dave Syer + * + */ +public class SingleItemPeekableItemReader implements ItemStreamReader, PeekableItemReader { + + private ItemReader delegate; + + private T next; + + private ExecutionContext executionContext = new ExecutionContext(); + + /** + * The item reader to use as a delegate. Items are read from the delegate + * and passed to the caller in {@link #read()}. + * + * @param delegate the delegate to set + */ + public void setDelegate(ItemReader delegate) { + this.delegate = delegate; + } + + /** + * Get the next item from the delegate (whether or not it has already been + * peeked at). + * + * @see ItemReader#read() + */ + public T read() throws Exception, UnexpectedInputException, ParseException { + if (next != null) { + T item = next; + next = null; + // executionContext = new ExecutionContext(); + return item; + } + return delegate.read(); + } + + /** + * Peek at the next item, ensuring that if the delegate is an + * {@link ItemStream} the state is stored for the next call to + * {@link #update(ExecutionContext)}. + * + * @return the next item (or null if there is none). + * + * @see PeekableItemReader#peek() + */ + @Override + public T peek() throws Exception, UnexpectedInputException, ParseException { + if (next == null) { + updateDelegate(executionContext); + next = delegate.read(); + } + return next; + } + + /** + * If the delegate is an {@link ItemStream}, just pass the call on, + * otherwise reset the peek cache. + * + * @throws ItemStreamException if there is a problem + * @see ItemStream#close() + */ + public void close() throws ItemStreamException { + next = null; + if (delegate instanceof ItemStream) { + ((ItemStream) delegate).close(); + } + executionContext = new ExecutionContext(); + } + + /** + * If the delegate is an {@link ItemStream}, just pass the call on, + * otherwise reset the peek cache. + * + * @param executionContext the current context + * @throws ItemStreamException if there is a problem + * @see ItemStream#open(ExecutionContext) + */ + public void open(ExecutionContext executionContext) throws ItemStreamException { + next = null; + if (delegate instanceof ItemStream) { + ((ItemStream) delegate).open(executionContext); + } + executionContext = new ExecutionContext(); + } + + /** + * If there is a cached peek, then retrieve the execution context state from + * that point. If there is no peek cached, then call directly to the + * delegate. + * + * @param executionContext the current context + * @throws ItemStreamException if there is a problem + * @see ItemStream#update(ExecutionContext) + */ + public void update(ExecutionContext executionContext) throws ItemStreamException { + if (next != null) { + // Get the last state from the delegate instead of using + // current value. + for (Entry entry : this.executionContext.entrySet()) { + executionContext.put(entry.getKey(), entry.getValue()); + } + return; + } + updateDelegate(executionContext); + } + + private void updateDelegate(ExecutionContext executionContext) { + if (delegate instanceof ItemStream) { + ((ItemStream) delegate).update(executionContext); + } + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SingleItemPeekableItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SingleItemPeekableItemReaderTests.java new file mode 100644 index 000000000..496bb08bb --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SingleItemPeekableItemReaderTests.java @@ -0,0 +1,135 @@ +/* + * Copyright 2006-2010 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 java.util.Arrays; +import java.util.List; + +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; + +/** + * @author Dave Syer + * + */ +public class SingleItemPeekableItemReaderTests { + + private SingleItemPeekableItemReader reader = new SingleItemPeekableItemReader(); + + /** + * Test method for {@link org.springframework.batch.item.support.SingleItemPeekableItemReader#read()}. + */ + @Test + public void testRead() throws Exception { + reader.setDelegate(new CountingListItemReader(Arrays.asList("a", "b"))); + assertEquals("a", reader.read()); + assertEquals("b", reader.read()); + assertEquals(null, reader.read()); + } + + /** + * Test method for {@link org.springframework.batch.item.support.SingleItemPeekableItemReader#peek()}. + */ + @Test + public void testPeek() throws Exception { + reader.setDelegate(new CountingListItemReader(Arrays.asList("a", "b"))); + assertEquals("a", reader.peek()); + assertEquals("a", reader.read()); + assertEquals("b", reader.read()); + assertEquals(null, reader.peek()); + assertEquals(null, reader.read()); + } + + /** + * Test method for {@link org.springframework.batch.item.support.SingleItemPeekableItemReader#close()}. + */ + @Test + public void testCloseAndOpenNoPeek() throws Exception { + reader.setDelegate(new CountingListItemReader(Arrays.asList("a", "b"))); + assertEquals("a", reader.read()); + ExecutionContext executionContext = new ExecutionContext(); + reader.update(executionContext); + reader.close(); + reader.open(executionContext); + assertEquals("b", reader.read()); + } + + /** + * Test method for {@link org.springframework.batch.item.support.SingleItemPeekableItemReader#close()}. + */ + @Test + public void testCloseAndOpenWithPeek() throws Exception { + reader.setDelegate(new CountingListItemReader(Arrays.asList("a", "b", "c"))); + assertEquals("a", reader.read()); + assertEquals("b", reader.peek()); + ExecutionContext executionContext = new ExecutionContext(); + reader.update(executionContext); + reader.close(); + reader.open(executionContext); + assertEquals("b", reader.read()); + } + + @Test + public void testCloseAndOpenWithPeekAndRead() throws Exception { + ExecutionContext executionContext = new ExecutionContext(); + reader.setDelegate(new CountingListItemReader(Arrays.asList("a", "b", "c"))); + assertEquals("a", reader.read()); + assertEquals("b", reader.peek()); + reader.update(executionContext); + reader.close(); + reader.open(executionContext); + assertEquals("b", reader.read()); + assertEquals("c", reader.peek()); + reader.update(executionContext); + reader.close(); + reader.open(executionContext); + assertEquals("c", reader.read()); + } + + public static class CountingListItemReader extends AbstractItemCountingItemStreamItemReader { + + private final List list; + + private int counter = 0; + + public CountingListItemReader(List list) { + this.list = list; + setName("foo"); + } + + @Override + protected void doClose() throws Exception { + counter = 0; + } + + @Override + protected void doOpen() throws Exception { + counter = 0; + } + + @Override + protected T doRead() throws Exception { + if (counter>=list.size()) { + return null; + } + return list.get(counter++); + } + + } + +}