REOPENED - BATCH-745: strong typing in AggregateItemReader

moved AggregateItemReader to samples

spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/multiline/AggregateItem.java

@@ -0,0 +1,112 @@
+/*
+ * 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.sample.domain.multiline;
+
+import org.springframework.batch.item.ItemReaderException;
+
+/**
+ * A wrapper type for an item that is used by {@link AggregateItemReader} to
+ * identify the start and end of an aggregate record.
+ * 
+ * @see AggregateItemReader
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class AggregateItem<T> {
+
+	@SuppressWarnings("unchecked")
+	private static final AggregateItem FOOTER = new AggregateItem<Object>(false, true) {
+		@Override
+		public Object getItem() throws ItemReaderException {
+			throw new IllegalStateException("Footer record has no item.");
+		}
+	};
+
+	/**
+	 * @param <T> the type of item nominally wrapped
+	 * @return a static {@link AggregateItem} that is a footer.
+	 */
+	@SuppressWarnings("unchecked")
+	public static final <T> AggregateItem<T> getFooter() {
+		return FOOTER;
+	}
+
+	@SuppressWarnings("unchecked")
+	private static final AggregateItem HEADER = new AggregateItem<Object>(true, false) {
+		@Override
+		public Object getItem() throws ItemReaderException {
+			throw new IllegalStateException("Header record has no item.");
+		}
+	};
+
+	/**
+	 * @param <T> the type of item nominally wrapped
+	 * @return a static {@link AggregateItem} that is a header.
+	 */
+	@SuppressWarnings("unchecked")
+	public static final <T> AggregateItem<T> getHeader() {
+		return HEADER;
+	}
+
+	private T item;
+
+	private boolean footer = false;
+
+	private boolean header = false;
+
+	/**
+	 * @param item
+	 */
+	public AggregateItem(T item) {
+		super();
+		this.item = item;
+	}
+
+	public AggregateItem(boolean header, boolean footer) {
+		this(null);
+		this.header = header;
+		this.footer = footer;
+	}
+
+	/**
+	 * Accessor for the wrapped item.
+	 * 
+	 * @return the wrapped item
+	 * @throws IllegalStateException if called on a record for which either
+	 * {@link #isHeader()} or {@link #isFooter()} answers true.
+	 */
+	public T getItem() throws IllegalStateException {
+		return item;
+	}
+
+	/**
+	 * Responds true if this record is a footer in an aggregate.
+	 * @return true if this is the end of an aggregate record.
+	 */
+	public boolean isFooter() {
+		return footer;
+	}
+
+	/**
+	 * Responds true if this record is a header in an aggregate.
+	 * @return true if this is the beginning of an aggregate record.
+	 */
+	public boolean isHeader() {
+		return header;
+	}
+
+}

spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/multiline/AggregateItemFieldSetMapper.java

@@ -0,0 +1,99 @@
+/*
+ * 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.sample.domain.multiline;
+
+import org.springframework.batch.item.file.mapping.FieldSet;
+import org.springframework.batch.item.file.mapping.FieldSetMapper;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.util.Assert;
+
+/**
+ * Delegating mapper to convert form a vanilla {@link FieldSetMapper} to one
+ * that returns {@link AggregateItem} instances for consumption by the
+ * {@link AggregateItemReader}.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class AggregateItemFieldSetMapper<T> implements FieldSetMapper<AggregateItem<T>>, InitializingBean {
+
+	private FieldSetMapper<T> delegate;
+
+	private String end = "END";
+
+	private String begin = "BEGIN";
+
+	/**
+	 * Public setter for the delegate.
+	 * @param delegate the delegate to set
+	 */
+	public void setDelegate(FieldSetMapper<T> delegate) {
+		this.delegate = delegate;
+	}
+
+	/**
+	 * Public setter for the end field value. If the {@link FieldSet} input has
+	 * a first field with this value that signals the start of an aggregate
+	 * record.
+	 * 
+	 * @param end the end to set
+	 */
+	public void setEnd(String end) {
+		this.end = end;
+	}
+
+	/**
+	 * Public setter for the begin value. If the {@link FieldSet} input has a
+	 * first field with this value that signals the end of an aggregate record.
+	 * 
+	 * @param begin the begin to set
+	 */
+	public void setBegin(String begin) {
+		this.begin = begin;
+	}
+
+	/**
+	 * Check mandatory properties (delegate).
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(delegate, "A FieldSetMapper delegate must be provided.");
+	}
+
+	/**
+	 * Build an {@link AggregateItem} based on matching the first column in the
+	 * input {@link FieldSet} to check for begin and end delimiters. If the
+	 * current record is neither a begin nor an end marker then it is mapped
+	 * using the delegate.
+	 * @param fieldSet a {@link FieldSet} to map
+	 * 
+	 * @return an {@link AggregateItem} that wraps the return value from the
+	 * delegate
+	 */
+	public AggregateItem<T> process(FieldSet fieldSet) throws Exception {
+
+		if (fieldSet.readString(0).equals(begin)) {
+			return AggregateItem.getHeader();
+		}
+		if (fieldSet.readString(0).equals(end)) {
+			return AggregateItem.getFooter();
+		}
+
+		return new AggregateItem<T>(delegate.process(fieldSet));
+
+	}
+
+}

spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/multiline/AggregateItemReader.java

@@ -0,0 +1,114 @@
+/*
+ * 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.sample.domain.multiline;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ItemReader;
+
+/**
+ * An {@link ItemReader} that delivers a list as its item, storing up objects
+ * from the injected {@link ItemReader} until they are ready to be packed out as
+ * a collection. This class must be used as a wrapper for a custom
+ * {@link ItemReader} that can identify the record boundaries. The custom reader
+ * should mark the beginning and end of records by returning an
+ * {@link AggregateItem} which responds true to its query methods
+ * <code>is*()</code>.<br/><br/>
+ * 
+ * This class is thread safe (it can be used concurrently by multiple threads)
+ * as long as the {@link ItemReader} is also thread safe.
+ * 
+ * @see AggregateItem#isHeader()
+ * @see AggregateItem#isFooter()
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class AggregateItemReader<T> implements ItemReader<List<T>> {
+
+	private static final Log log = LogFactory.getLog(AggregateItemReader.class);
+
+	private ItemReader<AggregateItem<T>> itemReader;
+
+	/**
+	 * Get the next list of records.
+	 * @throws Exception
+	 * 
+	 * @see org.springframework.batch.item.ItemReader#read()
+	 */
+	public List<T> read() throws Exception {
+		ResultHolder holder = new ResultHolder();
+
+		while (process(itemReader.read(), holder)) {
+			continue;
+		}
+
+		if (!holder.exhausted) {
+			return holder.records;
+		}
+		else {
+			return null;
+		}
+	}
+
+	private boolean process(AggregateItem<T> value, ResultHolder holder) {
+		// finish processing if we hit the end of file
+		if (value == null) {
+			log.debug("Exhausted ItemReader");
+			holder.exhausted = true;
+			return false;
+		}
+
+		// start a new collection
+		if (value.isHeader()) {
+			log.debug("Start of new record detected");
+			return true;
+		}
+
+		// mark we are finished with current collection
+		if (value.isFooter()) {
+			log.debug("End of record detected");
+			return false;
+		}
+
+		// add a simple record to the current collection
+		log.debug("Mapping: " + value);
+		holder.records.add(value.getItem());
+		return true;
+	}
+
+	public void setItemReader(ItemReader<AggregateItem<T>> itemReader) {
+		this.itemReader = itemReader;
+	}
+
+	/**
+	 * Private class for temporary state management while item is being
+	 * collected.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 */
+	private class ResultHolder {
+		List<T> records = new ArrayList<T>();
+
+		boolean exhausted = false;
+	}
+
+}

spring-batch-samples/src/main/resources/jobs/multilineJob.xml

@@ -13,7 +13,7 @@
 				<property name="streams" ref="fileItemReader"/>
 				<property name="itemReader">
 					<bean
-						class="org.springframework.batch.item.support.AggregateItemReader">
+						class="org.springframework.batch.sample.domain.multiline.AggregateItemReader">
 						<property name="itemReader"
 							ref="fileItemReader" />
 					</bean>
@@ -40,7 +40,7 @@
 			<bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
 				<property name="lineTokenizer" ref="fixedFileDescriptor" />
 				<property name="fieldSetMapper">
-					<bean class="org.springframework.batch.item.support.AggregateItemFieldSetMapper">
+					<bean class="org.springframework.batch.sample.domain.multiline.AggregateItemFieldSetMapper">
 						<property name="delegate" ref="tradeLineMapper" />
 					</bean>
 				</property>

spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/multiline/AggregateItemFieldSetMapperTests.java

@@ -0,0 +1,64 @@
+package org.springframework.batch.sample.domain.multiline;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import org.junit.Test;
+import org.springframework.batch.item.file.mapping.DefaultFieldSet;
+import org.springframework.batch.item.file.mapping.FieldSet;
+import org.springframework.batch.item.file.mapping.FieldSetMapper;
+import org.springframework.batch.sample.domain.multiline.AggregateItemFieldSetMapper;
+
+public class AggregateItemFieldSetMapperTests {
+
+	private AggregateItemFieldSetMapper<String> mapper = new AggregateItemFieldSetMapper<String>();
+
+	@Test
+	public void testDefaultBeginRecord() throws Exception {
+		assertTrue(mapper.process(new DefaultFieldSet(new String[] { "BEGIN" })).isHeader());
+		assertFalse(mapper.process(new DefaultFieldSet(new String[] { "BEGIN" })).isFooter());
+	}
+
+	@Test
+	public void testSetBeginRecord() throws Exception {
+		mapper.setBegin("FOO");
+		assertTrue(mapper.process(new DefaultFieldSet(new String[] { "FOO" })).isHeader());
+	}
+
+	@Test
+	public void testDefaultEndRecord() throws Exception {
+		assertFalse(mapper.process(new DefaultFieldSet(new String[] { "END" })).isHeader());
+		assertTrue(mapper.process(new DefaultFieldSet(new String[] { "END" })).isFooter());
+	}
+
+	@Test
+	public void testSetEndRecord() throws Exception {
+		mapper.setEnd("FOO");
+		assertTrue(mapper.process(new DefaultFieldSet(new String[] { "FOO" })).isFooter());
+	}
+
+	@Test
+	public void testMandatoryProperties() throws Exception {
+		try {
+			mapper.afterPropertiesSet();
+			fail("Expected IllegalArgumentException");
+		}
+		catch (IllegalArgumentException e) {
+			// expected
+		}
+	}
+
+	@Test
+	public void testDelegate() throws Exception {
+		mapper.setDelegate(new FieldSetMapper<String>() {
+			public String process(FieldSet fs) {
+				return "foo";
+			}
+		});
+		assertEquals("foo", mapper.process(new DefaultFieldSet(new String[] { "FOO" })).getItem());
+	}
+
+
+}

spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/multiline/AggregateItemReaderTests.java

@@ -0,0 +1,68 @@
+package org.springframework.batch.sample.domain.multiline;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Collection;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.sample.domain.multiline.AggregateItem;
+import org.springframework.batch.sample.domain.multiline.AggregateItemReader;
+
+public class AggregateItemReaderTests {
+
+	private ItemReader<AggregateItem<String>> input;
+
+	private AggregateItemReader<String> provider;
+
+	@Before
+	public void setUp() {
+		// create mock for input
+		input = new ItemReader<AggregateItem<String>>() {
+
+			private int count = 0;
+
+			public AggregateItem<String> read() {
+				switch (count++) {
+				case 0:
+					return AggregateItem.getHeader();
+				case 1:
+				case 2:
+				case 3:
+					return new AggregateItem<String>("line");
+				case 4:
+					return AggregateItem.getFooter();
+				default:
+					return null;
+				}
+			}
+
+		};
+		// create provider
+		provider = new AggregateItemReader<String>();
+		provider.setItemReader(input);
+	}
+
+	@Test
+	public void testNext() throws Exception {
+		// read object
+		Object result = provider.read();
+
+		// it should be collection of 3 strings "line"
+		assertTrue(result instanceof Collection);
+		Collection<?> lines = (Collection<?>) result;
+		assertEquals(3, lines.size());
+
+		for (Object line : lines) {
+			assertEquals("line", line);
+		}
+
+		// read object again - it should return null
+		assertNull(provider.read());
+
+	}
+
+}

spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/multiline/AggregateItemTests.java

@@ -0,0 +1,69 @@
+/*
+ * 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.sample.domain.multiline;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import org.junit.Test;
+import org.springframework.batch.sample.domain.multiline.AggregateItem;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class AggregateItemTests {
+
+	/**
+	 * Test method for {@link org.springframework.batch.sample.domain.multiline.AggregateItem#getFooter()}.
+	 */
+	@Test
+	public void testGetFooter() {
+		assertTrue(AggregateItem.getFooter().isFooter());
+		assertFalse(AggregateItem.getFooter().isHeader());
+	}
+
+	/**
+	 * Test method for {@link org.springframework.batch.sample.domain.multiline.AggregateItem#getHeader()}.
+	 */
+	@Test
+	public void testGetHeader() {
+		assertTrue(AggregateItem.getHeader().isHeader());
+		assertFalse(AggregateItem.getHeader().isFooter());
+	}
+
+	@Test
+	public void testBeginRecordHasNoItem() throws Exception {
+		try {
+			AggregateItem.getHeader().getItem();
+			fail("Expected IllegalStateException");
+		} catch(IllegalStateException e) {
+			// expected
+		}
+	}
+
+	@Test
+	public void testEndRecordHasNoItem() throws Exception {
+		try {
+			AggregateItem.getFooter().getItem();
+			fail("Expected IllegalStateException");
+		} catch(IllegalStateException e) {
+			// expected
+		}
+	}
+
+}