RESOLVED - issue BATCH-745: strong typing in AggregateItemReader

Added AggregateItem<T>

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/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.item.support;
+
+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-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemFieldSetMapper.java

@@ -0,0 +1,101 @@
+/*
+ * 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 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
+	 * @param lineNum the current line number if known
+	 * 
+	 * @return an {@link AggregateItem} that wraps the return value from the
+	 * delegate
+	 */
+	public AggregateItem<T> mapLine(FieldSet fieldSet, int lineNum) {
+
+		if (fieldSet.readString(0).equals(begin)) {
+			return AggregateItem.getHeader();
+		}
+		if (fieldSet.readString(0).equals(end)) {
+			return AggregateItem.getFooter();
+		}
+
+		return new AggregateItem<T>(delegate.mapLine(fieldSet, lineNum));
+
+	}
+
+}

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

@@ -28,32 +28,26 @@ import org.springframework.batch.item.ResetFailedException;
 /**
  * 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. Usually this class will be used as a wrapper for a custom
+ * 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 with the constant values ({@link AggregateItemReader#BEGIN_RECORD}
- * and {@link AggregateItemReader#END_RECORD}).<br/>
+ * 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 implements ItemReader<List<?>> {
+public class AggregateItemReader<T> implements ItemReader<List<T>> {
 
 	private static final Log log = LogFactory.getLog(AggregateItemReader.class);
 
-	/**
-	 * Marker for the end of a multi-object record.
-	 */
-	public static final Object END_RECORD = new Object();
-
-	/**
-	 * Marker for the beginning of a multi-object record.
-	 */
-	public static final Object BEGIN_RECORD = new Object();
-
-	private ItemReader<?> itemReader;
+	private ItemReader<AggregateItem<T>> itemReader;
 
 	/**
 	 * Get the next list of records.
@@ -61,7 +55,7 @@ public class AggregateItemReader implements ItemReader<List<?>> {
 	 * 
 	 * @see org.springframework.batch.item.ItemReader#read()
 	 */
-	public List<?> read() throws Exception {
+	public List<T> read() throws Exception {
 		ResultHolder holder = new ResultHolder();
 
 		while (process(itemReader.read(), holder)) {
@@ -76,7 +70,7 @@ public class AggregateItemReader implements ItemReader<List<?>> {
 		}
 	}
 
-	private boolean process(Object value, ResultHolder holder) {
+	private boolean process(AggregateItem<T> value, ResultHolder holder) {
 		// finish processing if we hit the end of file
 		if (value == null) {
 			log.debug("Exhausted ItemReader");
@@ -85,20 +79,20 @@ public class AggregateItemReader implements ItemReader<List<?>> {
 		}
 
 		// start a new collection
-		if (value == AggregateItemReader.BEGIN_RECORD) {
+		if (value.isHeader()) {
 			log.debug("Start of new record detected");
 			return true;
 		}
 
 		// mark we are finished with current collection
-		if (value == AggregateItemReader.END_RECORD) {
+		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);
+		holder.records.add(value.getItem());
 		return true;
 	}
 
@@ -110,7 +104,7 @@ public class AggregateItemReader implements ItemReader<List<?>> {
 		itemReader.reset();
 	}
 
-	public void setItemReader(ItemReader<?> itemReader) {
+	public void setItemReader(ItemReader<AggregateItem<T>> itemReader) {
 		this.itemReader = itemReader;
 	}
 
@@ -122,7 +116,8 @@ public class AggregateItemReader implements ItemReader<List<?>> {
 	 * 
 	 */
 	private class ResultHolder {
-		List<Object> records = new ArrayList<Object>();
+		List<T> records = new ArrayList<T>();
+
 		boolean exhausted = false;
 	}
 

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemFieldSetMapperTests.java

@@ -0,0 +1,63 @@
+package org.springframework.batch.item.support;
+
+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;
+
+public class AggregateItemFieldSetMapperTests {
+
+	private AggregateItemFieldSetMapper<String> mapper = new AggregateItemFieldSetMapper<String>();
+
+	@Test
+	public void testDefaultBeginRecord() throws Exception {
+		assertTrue(mapper.mapLine(new DefaultFieldSet(new String[] { "BEGIN" }), -1).isHeader());
+		assertFalse(mapper.mapLine(new DefaultFieldSet(new String[] { "BEGIN" }), -1).isFooter());
+	}
+
+	@Test
+	public void testSetBeginRecord() throws Exception {
+		mapper.setBegin("FOO");
+		assertTrue(mapper.mapLine(new DefaultFieldSet(new String[] { "FOO" }), -1).isHeader());
+	}
+
+	@Test
+	public void testDefaultEndRecord() throws Exception {
+		assertFalse(mapper.mapLine(new DefaultFieldSet(new String[] { "END" }), -1).isHeader());
+		assertTrue(mapper.mapLine(new DefaultFieldSet(new String[] { "END" }), -1).isFooter());
+	}
+
+	@Test
+	public void testSetEndRecord() throws Exception {
+		mapper.setEnd("FOO");
+		assertTrue(mapper.mapLine(new DefaultFieldSet(new String[] { "FOO" }), -1).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 mapLine(FieldSet fs, int lineNum) {
+				return "foo";
+			}
+		});
+		assertEquals("foo", mapper.mapLine(new DefaultFieldSet(new String[] { "FOO" }), -1).getItem());
+	}
+
+
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemReaderTests.java

@@ -12,27 +12,27 @@ import org.springframework.batch.item.ItemReader;
 
 public class AggregateItemReaderTests {
 
-	private ItemReader<Object> input;
+	private ItemReader<AggregateItem<String>> input;
 
-	private AggregateItemReader provider;
+	private AggregateItemReader<String> provider;
 
 	@Before
 	public void setUp() {
 		// create mock for input
-		input = new AbstractItemReader<Object>() {
+		input = new AbstractItemReader<AggregateItem<String>>() {
 
 			private int count = 0;
 
-			public Object read() {
+			public AggregateItem<String> read() {
 				switch (count++) {
 				case 0:
-					return AggregateItemReader.BEGIN_RECORD;
+					return AggregateItem.getHeader();
 				case 1:
 				case 2:
 				case 3:
-					return "line";
+					return new AggregateItem<String>("line");
 				case 4:
-					return AggregateItemReader.END_RECORD;
+					return AggregateItem.getFooter();
 				default:
 					return null;
 				}
@@ -40,7 +40,7 @@ public class AggregateItemReaderTests {
 
 		};
 		// create provider
-		provider = new AggregateItemReader();
+		provider = new AggregateItemReader<String>();
 		provider.setItemReader(input);
 	}
 

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemTests.java

@@ -0,0 +1,68 @@
+/*
+ * 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.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import org.junit.Test;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class AggregateItemTests {
+
+	/**
+	 * Test method for {@link org.springframework.batch.item.support.AggregateItem#getFooter()}.
+	 */
+	@Test
+	public void testGetFooter() {
+		assertTrue(AggregateItem.getFooter().isFooter());
+		assertFalse(AggregateItem.getFooter().isHeader());
+	}
+
+	/**
+	 * Test method for {@link org.springframework.batch.item.support.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
+		}
+	}
+
+}

spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeFieldSetMapper.java

@@ -18,32 +18,19 @@ package org.springframework.batch.sample.domain.trade.internal;
 
 import org.springframework.batch.item.file.mapping.FieldSet;
 import org.springframework.batch.item.file.mapping.FieldSetMapper;
-import org.springframework.batch.item.support.AggregateItemReader;
 import org.springframework.batch.sample.domain.trade.Trade;
 
 
 
-@SuppressWarnings("unchecked")
-/**
- * TODO type safety
- */
-public class TradeFieldSetMapper implements FieldSetMapper {
+public class TradeFieldSetMapper implements FieldSetMapper<Trade> {
 	
 	public static final int ISIN_COLUMN = 0;
 	public static final int QUANTITY_COLUMN = 1;
 	public static final int PRICE_COLUMN = 2;
 	public static final int CUSTOMER_COLUMN = 3;
 	
-    public Object mapLine(FieldSet fieldSet, int lineNum) {
+    public Trade mapLine(FieldSet fieldSet, int lineNum) {
     	
-    	if ("BEGIN".equals(fieldSet.readString(0))) {
-    		return AggregateItemReader.BEGIN_RECORD;
-    	}
-    	
-    	if ("END".equals(fieldSet.readString(0))) {
-    		return AggregateItemReader.END_RECORD;
-    	}
-
     	Trade trade = new Trade();
     	trade.setIsin(fieldSet.readString(0));
         trade.setQuantity(fieldSet.readLong(1));

spring-batch-samples/src/main/java/org/springframework/batch/sample/support/ExceptionThrowingItemReaderProxy.java

@@ -41,10 +41,6 @@ public class ExceptionThrowingItemReaderProxy<T> extends DelegatingItemReader<T>
 		this.throwExceptionOnRecordNumber = throwExceptionOnRecordNumber;
 	}
 	
-	public int getThrowExceptionOnRecordNumber() {
-		return throwExceptionOnRecordNumber;
-	}
-
 	public T read() throws Exception {
 		
 		counter++;

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

@@ -37,7 +37,11 @@
 		<property name="resource"
 			value="classpath:data/multilineJob/input/20070122.teststream.multilineStep.txt" />
 		<property name="lineTokenizer" ref="fixedFileDescriptor" />
-		<property name="fieldSetMapper" ref="tradeLineMapper" />
+		<property name="fieldSetMapper">
+			<bean class="org.springframework.batch.item.support.AggregateItemFieldSetMapper">
+				<property name="delegate" ref="tradeLineMapper" />
+			</bean>
+		</property>
 		<!-- <property name="validator" ref="fixedValidator" /> -->
 	</bean>
 

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

@@ -60,9 +60,9 @@
 				class="org.springmodules.validation.valang.ValangValidator">
 				<property name="valang">
 					<value>
-						<![CDATA[
-										      { isin : length(?) < 13 : 'ISIN too long' : 'isin_length' : 12}
-										    ]]>
+						<![CDATA[
+						{ isin : length(?) < 13 : 'ISIN too long' : 'isin_length' : 12}
+						]]>
 					</value>
 				</property>
 			</bean>

spring-batch-samples/src/test/java/org/springframework/batch/sample/RestartFunctionalTests.java

@@ -52,7 +52,6 @@ public class RestartFunctionalTests extends AbstractBatchLauncherTests {
 		this.jdbcTemplate = new JdbcTemplate(dataSource);
 	}
 
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.test.AbstractSingleSpringContextTests#onTearDown()
@@ -76,7 +75,7 @@ public class RestartFunctionalTests extends AbstractBatchLauncherTests {
 		int before = jdbcTemplate.queryForInt("SELECT COUNT(*) FROM TRADE");
 
 		try {
-			runJob();
+			runJobForRestartTest();
 			fail("First run of the job is expected to fail.");
 		}
 		catch (UnexpectedJobExecutionException expected) {
@@ -89,7 +88,7 @@ public class RestartFunctionalTests extends AbstractBatchLauncherTests {
 		// assert based on commit interval = 2
 		assertEquals(before + 2, medium);
 
-		runJob();
+		runJobForRestartTest();
 
 		int after = jdbcTemplate.queryForInt("SELECT COUNT(*) FROM TRADE");
 
@@ -97,9 +96,11 @@ public class RestartFunctionalTests extends AbstractBatchLauncherTests {
 	}
 
 	// load the application context and launch the job
-	private void runJob() throws Exception {
+	private void runJobForRestartTest() throws Exception {
+		// The second time we run the job it needs to be a new instance so we
+		// need to make the parameters unique...
 		launcher.run(getJob(), new DefaultJobParametersConverter().getJobParameters(PropertiesConverter
-				.stringToProperties("restart=true")));
+				.stringToProperties("force.new.job.parameters=true")));
 	}
 
 }

spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/TradeFieldSetMapperTests.java

@@ -1,16 +1,11 @@
 package org.springframework.batch.sample.domain.trade.internal;
 
-import static org.junit.Assert.assertEquals;
-
 import java.math.BigDecimal;
 
-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.item.support.AggregateItemReader;
 import org.springframework.batch.sample.domain.trade.Trade;
-import org.springframework.batch.sample.domain.trade.internal.TradeFieldSetMapper;
 import org.springframework.batch.sample.support.AbstractFieldSetMapperTests;
 
 public class TradeFieldSetMapperTests extends AbstractFieldSetMapperTests {
@@ -48,16 +43,4 @@ public class TradeFieldSetMapperTests extends AbstractFieldSetMapperTests {
 		return mapper;
 	}
 
-	@Test
-	public void testBeginRecord() throws Exception {
-		assertEquals(AggregateItemReader.BEGIN_RECORD, fieldSetMapper().mapLine(
-				new DefaultFieldSet(new String[] { "BEGIN" }), -1));
-	}
-
-	@Test
-	public void testEndRecord() throws Exception {
-		assertEquals(AggregateItemReader.END_RECORD, fieldSetMapper().mapLine(
-				new DefaultFieldSet(new String[] { "END" }), -1));
-	}
-
 }