diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItem.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItem.java new file mode 100644 index 000000000..961f555e0 --- /dev/null +++ b/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 { + + @SuppressWarnings("unchecked") + private static final AggregateItem FOOTER = new AggregateItem(false, true) { + @Override + public Object getItem() throws ItemReaderException { + throw new IllegalStateException("Footer record has no item."); + } + }; + + /** + * @param the type of item nominally wrapped + * @return a static {@link AggregateItem} that is a footer. + */ + @SuppressWarnings("unchecked") + public static final AggregateItem getFooter() { + return FOOTER; + } + + @SuppressWarnings("unchecked") + private static final AggregateItem HEADER = new AggregateItem(true, false) { + @Override + public Object getItem() throws ItemReaderException { + throw new IllegalStateException("Header record has no item."); + } + }; + + /** + * @param the type of item nominally wrapped + * @return a static {@link AggregateItem} that is a header. + */ + @SuppressWarnings("unchecked") + public static final AggregateItem 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; + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemFieldSetMapper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemFieldSetMapper.java new file mode 100644 index 000000000..790962237 --- /dev/null +++ b/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 implements FieldSetMapper>, InitializingBean { + + private FieldSetMapper delegate; + + private String end = "END"; + + private String begin = "BEGIN"; + + /** + * Public setter for the delegate. + * @param delegate the delegate to set + */ + public void setDelegate(FieldSetMapper 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 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(delegate.mapLine(fieldSet, lineNum)); + + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemReader.java index 490b4b277..d800f7c4d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemReader.java +++ b/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}).
+ * should mark the beginning and end of records by returning an + * {@link AggregateItem} which responds true to its query methods + * is*().

* * 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> { +public class AggregateItemReader implements ItemReader> { 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> itemReader; /** * Get the next list of records. @@ -61,7 +55,7 @@ public class AggregateItemReader implements ItemReader> { * * @see org.springframework.batch.item.ItemReader#read() */ - public List read() throws Exception { + public List read() throws Exception { ResultHolder holder = new ResultHolder(); while (process(itemReader.read(), holder)) { @@ -76,7 +70,7 @@ public class AggregateItemReader implements ItemReader> { } } - private boolean process(Object value, ResultHolder holder) { + private boolean process(AggregateItem 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> { } // 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> { itemReader.reset(); } - public void setItemReader(ItemReader itemReader) { + public void setItemReader(ItemReader> itemReader) { this.itemReader = itemReader; } @@ -122,7 +116,8 @@ public class AggregateItemReader implements ItemReader> { * */ private class ResultHolder { - List records = new ArrayList(); + List records = new ArrayList(); + boolean exhausted = false; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemFieldSetMapperTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemFieldSetMapperTests.java new file mode 100644 index 000000000..a9c4e4b07 --- /dev/null +++ b/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 mapper = new AggregateItemFieldSetMapper(); + + @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() { + public String mapLine(FieldSet fs, int lineNum) { + return "foo"; + } + }); + assertEquals("foo", mapper.mapLine(new DefaultFieldSet(new String[] { "FOO" }), -1).getItem()); + } + + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemReaderTests.java index 68f320949..8d0cace50 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemReaderTests.java +++ b/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 input; + private ItemReader> input; - private AggregateItemReader provider; + private AggregateItemReader provider; @Before public void setUp() { // create mock for input - input = new AbstractItemReader() { + input = new AbstractItemReader>() { private int count = 0; - public Object read() { + public AggregateItem read() { switch (count++) { case 0: - return AggregateItemReader.BEGIN_RECORD; + return AggregateItem.getHeader(); case 1: case 2: case 3: - return "line"; + return new AggregateItem("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(); provider.setItemReader(input); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/AggregateItemTests.java new file mode 100644 index 000000000..808ba530c --- /dev/null +++ b/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 + } + } + +} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeFieldSetMapper.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeFieldSetMapper.java index 5e177dcfb..1356f05b3 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeFieldSetMapper.java +++ b/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 { 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)); diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/support/ExceptionThrowingItemReaderProxy.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/support/ExceptionThrowingItemReaderProxy.java index a194553bf..4f8cfd443 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/support/ExceptionThrowingItemReaderProxy.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/support/ExceptionThrowingItemReaderProxy.java @@ -41,10 +41,6 @@ public class ExceptionThrowingItemReaderProxy extends DelegatingItemReader this.throwExceptionOnRecordNumber = throwExceptionOnRecordNumber; } - public int getThrowExceptionOnRecordNumber() { - return throwExceptionOnRecordNumber; - } - public T read() throws Exception { counter++; diff --git a/spring-batch-samples/src/main/resources/jobs/multilineJob.xml b/spring-batch-samples/src/main/resources/jobs/multilineJob.xml index 717b575d3..674557c6b 100644 --- a/spring-batch-samples/src/main/resources/jobs/multilineJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/multilineJob.xml @@ -37,7 +37,11 @@ - + + + + + diff --git a/spring-batch-samples/src/main/resources/jobs/restartSample.xml b/spring-batch-samples/src/main/resources/jobs/restartSample.xml index 293e05f5a..ad0192f3d 100644 --- a/spring-batch-samples/src/main/resources/jobs/restartSample.xml +++ b/spring-batch-samples/src/main/resources/jobs/restartSample.xml @@ -60,9 +60,9 @@ class="org.springmodules.validation.valang.ValangValidator"> - + diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RestartFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RestartFunctionalTests.java index b30104147..0256c0660 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RestartFunctionalTests.java +++ b/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"))); } } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/TradeFieldSetMapperTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/TradeFieldSetMapperTests.java index 096cd2e35..10b341523 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/TradeFieldSetMapperTests.java +++ b/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)); - } - }