From d1b84f3e2fea411cee1c113b3c41aaaa06003cab Mon Sep 17 00:00:00 2001 From: Dean de Bree Date: Wed, 27 Feb 2013 08:52:36 +0200 Subject: [PATCH] Added ExecutionContextUserSupport to the ItemStream Support class Changed child classes to use this shared object Changed some classes that previously extended ExecutionContextUserSupport to rather use the same common parent class as everything else. Changes all for issue BATCH-1966 --- .../batch/core/step/item/ChunkMonitor.java | 17 ++++-- .../core/configuration/xml/TestReader.java | 6 +- .../core/partition/ExampleItemReader.java | 18 +++--- .../core/step/item/ChunkMonitorTests.java | 3 +- .../FaultTolerantStepFactoryBeanTests.java | 58 +++++++------------ .../step/item/TaskletStepExceptionTests.java | 1 + .../step/tasklet/AsyncTaskletStepTests.java | 1 + .../core/step/tasklet/TaskletStepTests.java | 16 +++-- .../batch/item/ItemStreamSupport.java | 20 ++++++- .../item/database/JdbcPagingItemReader.java | 4 +- .../batch/item/file/FlatFileItemWriter.java | 19 +++--- .../item/file/MultiResourceItemReader.java | 19 +++--- .../item/file/MultiResourceItemWriter.java | 20 ++++--- .../batch/item/file/ResourcesItemReader.java | 19 +++--- ...tractItemCountingItemStreamItemReader.java | 28 ++++----- .../support/AbstractItemStreamItemReader.java | 3 +- .../support/AbstractItemStreamItemWriter.java | 3 +- .../batch/item/xml/StaxEventItemWriter.java | 22 +++---- .../MultiResourceItemWriterFlatFileTests.java | 8 +-- .../item/file/ResourcesItemReaderTests.java | 4 +- .../support/CompositeItemStreamTests.java | 21 ++++--- 21 files changed, 168 insertions(+), 142 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java index ae99ac162..7959bd344 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java @@ -21,7 +21,9 @@ 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.ItemStreamSupport; import org.springframework.batch.item.support.CompositeItemStream; +import org.springframework.util.ClassUtils; /** * Manage the offset data between the last successful commit and updates made to @@ -32,7 +34,7 @@ import org.springframework.batch.item.support.CompositeItemStream; * @author Dave Syer * @since 2.0 */ -public class ChunkMonitor implements ItemStream { +public class ChunkMonitor extends ItemStreamSupport { private Log logger = LogFactory.getLog(getClass()); @@ -49,7 +51,7 @@ public class ChunkMonitor implements ItemStream { } } - private static final String OFFSET = ChunkMonitor.class.getName() + ".OFFSET"; + private static final String OFFSET = "OFFSET"; private CompositeItemStream stream = new CompositeItemStream(); @@ -57,6 +59,10 @@ public class ChunkMonitor implements ItemStream { private ItemReader reader; + public ChunkMonitor() { + this.setExecutionContextName(ClassUtils.getShortName(this.getClass())); + } + /** * @param stream the stream to set */ @@ -95,6 +101,7 @@ public class ChunkMonitor implements ItemStream { @Override public void close() throws ItemStreamException { + super.close(); holder.set(null); if (streamsRegistered) { stream.close(); @@ -103,9 +110,10 @@ public class ChunkMonitor implements ItemStream { @Override public void open(ExecutionContext executionContext) throws ItemStreamException { + super.open(executionContext); if (streamsRegistered) { stream.open(executionContext); - ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(OFFSET, 0), 0); + ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(getExecutionContextKey(OFFSET), 0), 0); holder.set(data); if (reader == null) { logger.warn("No ItemReader set (must be concurrent step), so ignoring offset data."); @@ -124,6 +132,7 @@ public class ChunkMonitor implements ItemStream { @Override public void update(ExecutionContext executionContext) throws ItemStreamException { + super.update(executionContext); if (streamsRegistered) { ChunkMonitorData data = getData(); if (data.offset == 0) { @@ -132,7 +141,7 @@ public class ChunkMonitor implements ItemStream { stream.update(executionContext); } else { - executionContext.putInt(OFFSET, data.offset); + executionContext.putInt(getExecutionContextKey(OFFSET), data.offset); } } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java index 630725c1a..7aa95a896 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java @@ -43,16 +43,16 @@ public class TestReader extends AbstractTestComponent implements ItemStreamReade } @Override - public void close() throws ItemStreamException { + public void close() { } @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { + public void open(ExecutionContext executionContext) { opened = true; } @Override - public void update(ExecutionContext executionContext) throws ItemStreamException { + public void update(ExecutionContext executionContext) { } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java index 216a80175..131bcf32c 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java @@ -5,11 +5,13 @@ import org.apache.commons.logging.LogFactory; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemStreamReader; +import org.springframework.batch.item.support.AbstractItemStreamItemReader; +import org.springframework.util.ClassUtils; /** * {@link ItemStreamReader} with hard-coded input data. */ -public class ExampleItemReader implements ItemStreamReader { +public class ExampleItemReader extends AbstractItemStreamItemReader { private Log logger = LogFactory.getLog(getClass()); @@ -23,6 +25,10 @@ public class ExampleItemReader implements ItemStreamReader { public static volatile boolean fail = false; + public ExampleItemReader() { + this.setExecutionContextName(ClassUtils.getShortName(this.getClass())); + } + /** * @param min the min to set */ @@ -61,18 +67,16 @@ public class ExampleItemReader implements ItemStreamReader { return input[index++]; } - @Override - public void close() throws ItemStreamException { - } - @Override public void open(ExecutionContext executionContext) throws ItemStreamException { - index = (int) executionContext.getLong("POSITION", min); + super.open(executionContext); + index = (int) executionContext.getLong(getExecutionContextKey("POSITION"), min); } @Override public void update(ExecutionContext executionContext) throws ItemStreamException { - executionContext.putLong("POSITION", index); + super.update(executionContext); + executionContext.putLong(getExecutionContextKey("POSITION"), index); } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java index 13fd1c23a..4b8edf3cc 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java @@ -54,7 +54,8 @@ public class ChunkMonitorTests { }); monitor.registerItemStream(new ItemStreamSupport() { @Override - public void close() throws ItemStreamException { + public void close() { + super.close(); closed = true; } }); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java index 3bdef5d4b..c8e9ee25a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java @@ -49,6 +49,7 @@ import org.springframework.batch.item.ParseException; import org.springframework.batch.item.UnexpectedInputException; import org.springframework.batch.item.WriteFailedException; import org.springframework.batch.item.WriterNotOpenException; +import org.springframework.batch.item.support.AbstractItemStreamItemReader; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.beans.factory.FactoryBean; import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; @@ -830,23 +831,21 @@ public class FaultTolerantStepFactoryBeanTests { public void testItemStreamOpenedEvenWithTaskExecutor() throws Exception { writer.setFailures("4"); - ItemStreamReader reader = new ItemStreamReader() { + ItemStreamReader reader = new AbstractItemStreamItemReader() { @Override - public void close() throws ItemStreamException { + public void close() { + super.close(); closed = true; } @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { + public void open(ExecutionContext executionContext) { + super.open(executionContext); opened = true; } @Override - public void update(ExecutionContext executionContext) throws ItemStreamException { - } - - @Override - public String read() throws Exception, UnexpectedInputException, ParseException { + public String read() { return null; } }; @@ -870,42 +869,29 @@ public class FaultTolerantStepFactoryBeanTests { public void testNestedItemStreamOpened() throws Exception { writer.setFailures("4"); - ItemStreamReader reader = new ItemStreamReader() { - @Override - public void close() throws ItemStreamException { - } + ItemStreamReader reader = new AbstractItemStreamItemReader() { @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { - } - - @Override - public void update(ExecutionContext executionContext) throws ItemStreamException { - } - - @Override - public String read() throws Exception, UnexpectedInputException, ParseException { + public String read() { return null; } }; - ItemStreamReader stream = new ItemStreamReader() { + ItemStreamReader stream = new AbstractItemStreamItemReader() { @Override - public void close() throws ItemStreamException { + public void close() { + super.close(); closed = true; } @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { + public void open(ExecutionContext executionContext) { + super.open(executionContext); opened = true; } @Override - public void update(ExecutionContext executionContext) throws ItemStreamException { - } - - @Override - public String read() throws Exception, UnexpectedInputException, ParseException { + public String read() { return null; } }; @@ -930,23 +916,21 @@ public class FaultTolerantStepFactoryBeanTests { public void testProxiedItemStreamOpened() throws Exception { writer.setFailures("4"); - ItemStreamReader reader = new ItemStreamReader() { + ItemStreamReader reader = new AbstractItemStreamItemReader() { @Override - public void close() throws ItemStreamException { + public void close() { + super.close(); closed = true; } @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { + public void open(ExecutionContext executionContext) { + super.open(executionContext); opened = true; } @Override - public void update(ExecutionContext executionContext) throws ItemStreamException { - } - - @Override - public String read() throws Exception, UnexpectedInputException, ParseException { + public String read() { return null; } }; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java index 41d21f285..d8503620a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java @@ -184,6 +184,7 @@ public class TaskletStepExceptionTests { taskletStep.setStreams(new ItemStream[] { new ItemStreamSupport() { @Override public void close() throws ItemStreamException { + super.close(); throw exception; } } }); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java index 00891d40c..3be1e951a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java @@ -103,6 +103,7 @@ public class AsyncTaskletStepTests { @Override public void update(ExecutionContext executionContext) { + super.update(executionContext); executionContext.putInt("counter", count++); } }); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java index a019bac2a..70bbe524d 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java @@ -429,12 +429,13 @@ public class TaskletStepTests { public void testStreamManager() throws Exception { MockRestartableItemReader reader = new MockRestartableItemReader() { @Override - public String read() throws Exception { + public String read() { return "foo"; } @Override public void update(ExecutionContext executionContext) { + super.update(executionContext); executionContext.putString("foo", "bar"); } }; @@ -457,6 +458,7 @@ public class TaskletStepTests { step.setStreams(new ItemStream[] { new ItemStreamSupport() { @Override public void update(ExecutionContext executionContext) { + super.update(executionContext); executionContext.putString("foo", "bar"); } } }); @@ -500,6 +502,7 @@ public class TaskletStepTests { @Override public void open(ExecutionContext executionContext) throws ItemStreamException { + super.open(executionContext); assertEquals(1, list.size()); } }; @@ -547,7 +550,7 @@ public class TaskletStepTests { }); step.setTasklet(new TestingChunkOrientedTasklet(new MockRestartableItemReader() { @Override - public String read() throws Exception { + public String read() throws RuntimeException { throw new RuntimeException("FOO"); } }, itemWriter)); @@ -562,12 +565,13 @@ public class TaskletStepTests { public void testDirectlyInjectedStreamWhichIsAlsoReader() throws Exception { MockRestartableItemReader reader = new MockRestartableItemReader() { @Override - public String read() throws Exception { + public String read() { return "foo"; } @Override public void update(ExecutionContext executionContext) { + super.update(executionContext); executionContext.putString("foo", "bar"); } }; @@ -740,6 +744,7 @@ public class TaskletStepTests { step.setStreams(new ItemStream[] { new ItemStreamSupport() { @Override public void close() throws ItemStreamException { + super.close(); throw new RuntimeException("Bar"); } } }); @@ -797,7 +802,7 @@ public class TaskletStepTests { public void testRestartAfterFailureInFirstChunk() throws Exception { MockRestartableItemReader reader = new MockRestartableItemReader() { @Override - public String read() throws Exception { + public String read() throws RuntimeException { // fail on the very first item throw new RuntimeException("CRASH!"); } @@ -927,12 +932,13 @@ public class TaskletStepTests { private boolean restoreFromCalled = false; @Override - public String read() throws Exception { + public String read() { return "item"; } @Override public void update(ExecutionContext executionContext) { + super.update(executionContext); getExecutionAttributesCalled = true; executionContext.putString("spam", "bucket"); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStreamSupport.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStreamSupport.java index cebf23566..3663f0df7 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStreamSupport.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStreamSupport.java @@ -15,21 +15,25 @@ */ package org.springframework.batch.item; +import org.springframework.batch.item.util.ExecutionContextUserSupport; /** * Empty method implementation of {@link ItemStream}. * * @author Dave Syer + * @author Dean de Bree * */ public abstract class ItemStreamSupport implements ItemStream { + private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(); + /** * No-op. * @see org.springframework.batch.item.ItemStream#close() */ @Override - public void close() throws ItemStreamException { + public void close() { } /** @@ -37,7 +41,7 @@ public abstract class ItemStreamSupport implements ItemStream { * @see org.springframework.batch.item.ItemStream#open(ExecutionContext) */ @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { + public void open(ExecutionContext executionContext) { } /** @@ -47,5 +51,17 @@ public abstract class ItemStreamSupport implements ItemStream { @Override public void update(ExecutionContext executionContext) { } + + private ExecutionContextUserSupport getExecutionContextUserSupport() { + return executionContextUserSupport; + } + + protected void setExecutionContextName(String name) { + this.getExecutionContextUserSupport().setName(name); + } + + public String getExecutionContextKey(String key) { + return this.getExecutionContextUserSupport().getKey(key); + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java index 126867192..0c4f3fbcf 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java @@ -231,7 +231,7 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme public void update(ExecutionContext executionContext) throws ItemStreamException { super.update(executionContext); if (isSaveState() && startAfterValues != null) { - executionContext.put(getExecutionContextUserSupport().getKey(START_AFTER_VALUE), startAfterValues); + executionContext.put(getExecutionContextKey(START_AFTER_VALUE), startAfterValues); } } @@ -239,7 +239,7 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme @SuppressWarnings("unchecked") public void open(ExecutionContext executionContext) { if (isSaveState()) { - startAfterValues = (Map) executionContext.get(getExecutionContextUserSupport().getKey(START_AFTER_VALUE)); + startAfterValues = (Map) executionContext.get(getExecutionContextKey(START_AFTER_VALUE)); if(startAfterValues == null) { startAfterValues = new LinkedHashMap(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java index 13ad811d5..0034d88d2 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java @@ -34,7 +34,7 @@ import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.WriteFailedException; import org.springframework.batch.item.WriterNotOpenException; import org.springframework.batch.item.file.transform.LineAggregator; -import org.springframework.batch.item.util.ExecutionContextUserSupport; +import org.springframework.batch.item.support.AbstractItemStreamItemWriter; import org.springframework.batch.item.util.FileUtils; import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter; import org.springframework.beans.factory.InitializingBean; @@ -57,7 +57,7 @@ import org.springframework.util.ClassUtils; * @author Dave Syer * @author Michael Minella */ -public class FlatFileItemWriter extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream, +public class FlatFileItemWriter extends AbstractItemStreamItemWriter implements ResourceAwareItemWriterItemStream, InitializingBean { private static final boolean DEFAULT_TRANSACTIONAL = true; @@ -97,7 +97,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement private boolean append = false; public FlatFileItemWriter() { - setName(ClassUtils.getShortName(FlatFileItemWriter.class)); + this.setExecutionContextName(ClassUtils.getShortName(FlatFileItemWriter.class)); } /** @@ -281,6 +281,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement */ @Override public void close() { + super.close(); if (state != null) { try { if (footerCallback != null && state.outputBufferedWriter != null) { @@ -314,7 +315,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement */ @Override public void open(ExecutionContext executionContext) throws ItemStreamException { - + super.open(executionContext); + Assert.notNull(resource, "The resource must be set"); if (!getOutputState().isInitialized()) { @@ -324,7 +326,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement private void doOpen(ExecutionContext executionContext) throws ItemStreamException { OutputState outputState = getOutputState(); - if (executionContext.containsKey(getKey(RESTART_DATA_NAME))) { + if (executionContext.containsKey(getExecutionContextKey(RESTART_DATA_NAME))) { outputState.restoreFrom(executionContext); } try { @@ -351,6 +353,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement */ @Override public void update(ExecutionContext executionContext) { + super.update(executionContext); if (state == null) { throw new ItemStreamException("ItemStream not open or already closed."); } @@ -360,13 +363,13 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement if (saveState) { try { - executionContext.putLong(getKey(RESTART_DATA_NAME), state.position()); + executionContext.putLong(getExecutionContextKey(RESTART_DATA_NAME), state.position()); } catch (IOException e) { throw new ItemStreamException("ItemStream does not return current position properly", e); } - executionContext.putLong(getKey(WRITTEN_STATISTICS_NAME), state.linesWritten); + executionContext.putLong(getExecutionContextKey(WRITTEN_STATISTICS_NAME), state.linesWritten); } } @@ -454,7 +457,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement * @param executionContext */ public void restoreFrom(ExecutionContext executionContext) { - lastMarkedByteOffsetPosition = executionContext.getLong(getKey(RESTART_DATA_NAME)); + lastMarkedByteOffsetPosition = executionContext.getLong(getExecutionContextKey(RESTART_DATA_NAME)); restarted = true; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java index af3918e15..df82c4d2e 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java @@ -23,11 +23,10 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStreamException; -import org.springframework.batch.item.ItemStreamReader; import org.springframework.batch.item.ParseException; import org.springframework.batch.item.ResourceAware; import org.springframework.batch.item.UnexpectedInputException; -import org.springframework.batch.item.util.ExecutionContextUserSupport; +import org.springframework.batch.item.support.AbstractItemStreamItemReader; import org.springframework.core.io.Resource; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -43,14 +42,12 @@ import org.springframework.util.ClassUtils; * @author Robert Kasanicky * @author Lucas Ward */ -public class MultiResourceItemReader implements ItemStreamReader { +public class MultiResourceItemReader extends AbstractItemStreamItemReader { private static final Log logger = LogFactory.getLog(MultiResourceItemReader.class); private static final String RESOURCE_KEY = "resourceIndex"; - private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(); - private ResourceAwareItemReaderItemStream delegate; private Resource[] resources; @@ -86,7 +83,7 @@ public class MultiResourceItemReader implements ItemStreamReader { }; public MultiResourceItemReader() { - executionContextUserSupport.setName(ClassUtils.getShortName(MultiResourceItemReader.class)); + this.setExecutionContextName(ClassUtils.getShortName(MultiResourceItemReader.class)); } /** @@ -151,6 +148,7 @@ public class MultiResourceItemReader implements ItemStreamReader { */ @Override public void close() throws ItemStreamException { + super.close(); delegate.close(); noInput = false; } @@ -161,7 +159,7 @@ public class MultiResourceItemReader implements ItemStreamReader { */ @Override public void open(ExecutionContext executionContext) throws ItemStreamException { - + super.open(executionContext); Assert.notNull(resources, "Resources must be set"); noInput = false; @@ -179,8 +177,8 @@ public class MultiResourceItemReader implements ItemStreamReader { Arrays.sort(resources, comparator); - if (executionContext.containsKey(executionContextUserSupport.getKey(RESOURCE_KEY))) { - currentResource = executionContext.getInt(executionContextUserSupport.getKey(RESOURCE_KEY)); + if (executionContext.containsKey(getExecutionContextKey(RESOURCE_KEY))) { + currentResource = executionContext.getInt(getExecutionContextKey(RESOURCE_KEY)); // context could have been saved before reading anything if (currentResource == -1) { @@ -200,8 +198,9 @@ public class MultiResourceItemReader implements ItemStreamReader { */ @Override public void update(ExecutionContext executionContext) throws ItemStreamException { + super.update(executionContext); if (saveState) { - executionContext.putInt(executionContextUserSupport.getKey(RESOURCE_KEY), currentResource); + executionContext.putInt(getExecutionContextKey(RESOURCE_KEY), currentResource); delegate.update(executionContext); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java index 553a6f06d..abf2a9137 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemWriter.java @@ -21,8 +21,7 @@ import java.io.IOException; import java.util.List; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStreamException; -import org.springframework.batch.item.ItemStreamWriter; -import org.springframework.batch.item.util.ExecutionContextUserSupport; +import org.springframework.batch.item.support.AbstractItemStreamItemWriter; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; import org.springframework.util.Assert; @@ -42,7 +41,7 @@ import org.springframework.util.ClassUtils; * * @author Robert Kasanicky */ -public class MultiResourceItemWriter extends ExecutionContextUserSupport implements ItemStreamWriter { +public class MultiResourceItemWriter extends AbstractItemStreamItemWriter { final static private String RESOURCE_INDEX_KEY = "resource.index"; @@ -65,7 +64,7 @@ public class MultiResourceItemWriter extends ExecutionContextUserSupport impl private boolean opened = false; public MultiResourceItemWriter() { - setName(ClassUtils.getShortName(MultiResourceItemWriter.class)); + this.setExecutionContextName(ClassUtils.getShortName(MultiResourceItemWriter.class)); } @Override @@ -128,6 +127,7 @@ public class MultiResourceItemWriter extends ExecutionContextUserSupport impl @Override public void close() throws ItemStreamException { + super.close(); resourceIndex = 1; currentResourceItemCount = 0; if (opened) { @@ -137,8 +137,9 @@ public class MultiResourceItemWriter extends ExecutionContextUserSupport impl @Override public void open(ExecutionContext executionContext) throws ItemStreamException { - resourceIndex = executionContext.getInt(getKey(RESOURCE_INDEX_KEY), 1); - currentResourceItemCount = executionContext.getInt(getKey(CURRENT_RESOURCE_ITEM_COUNT), 0); + super.open(executionContext); + resourceIndex = executionContext.getInt(getExecutionContextKey(RESOURCE_INDEX_KEY), 1); + currentResourceItemCount = executionContext.getInt(getExecutionContextKey(CURRENT_RESOURCE_ITEM_COUNT), 0); try { setResourceToDelegate(); @@ -147,7 +148,7 @@ public class MultiResourceItemWriter extends ExecutionContextUserSupport impl throw new ItemStreamException("Couldn't assign resource", e); } - if (executionContext.containsKey(getKey(CURRENT_RESOURCE_ITEM_COUNT))) { + if (executionContext.containsKey(getExecutionContextKey(CURRENT_RESOURCE_ITEM_COUNT))) { // It's a restart delegate.open(executionContext); } @@ -158,12 +159,13 @@ public class MultiResourceItemWriter extends ExecutionContextUserSupport impl @Override public void update(ExecutionContext executionContext) throws ItemStreamException { + super.update(executionContext); if (saveState) { if (opened) { delegate.update(executionContext); } - executionContext.putInt(getKey(CURRENT_RESOURCE_ITEM_COUNT), currentResourceItemCount); - executionContext.putInt(getKey(RESOURCE_INDEX_KEY), resourceIndex); + executionContext.putInt(getExecutionContextKey(CURRENT_RESOURCE_ITEM_COUNT), currentResourceItemCount); + executionContext.putInt(getExecutionContextKey(RESOURCE_INDEX_KEY), resourceIndex); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourcesItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourcesItemReader.java index ee1402009..f8dcbd7e9 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourcesItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/ResourcesItemReader.java @@ -6,8 +6,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStreamException; -import org.springframework.batch.item.ItemStreamReader; -import org.springframework.batch.item.util.ExecutionContextUserSupport; +import org.springframework.batch.item.support.AbstractItemStreamItemReader; import org.springframework.core.io.Resource; import org.springframework.core.io.support.ResourceArrayPropertyEditor; @@ -30,17 +29,17 @@ import org.springframework.core.io.support.ResourceArrayPropertyEditor; * * @since 2.1 */ -public class ResourcesItemReader extends ExecutionContextUserSupport implements ItemStreamReader { +public class ResourcesItemReader extends AbstractItemStreamItemReader { private Resource[] resources = new Resource[0]; private AtomicInteger counter = new AtomicInteger(0); - { + public ResourcesItemReader() { /* * Initialize the name for the key in the execution context. */ - setName(getClass().getName()); + this.setExecutionContextName(getClass().getName()); } /** @@ -65,18 +64,16 @@ public class ResourcesItemReader extends ExecutionContextUserSupport implements return resources[index]; } - @Override - public void close() throws ItemStreamException { - } - @Override public void open(ExecutionContext executionContext) throws ItemStreamException { - counter.set(executionContext.getInt(getKey("COUNT"), 0)); + super.open(executionContext); + counter.set(executionContext.getInt(getExecutionContextKey("COUNT"), 0)); } @Override public void update(ExecutionContext executionContext) throws ItemStreamException { - executionContext.putInt(getKey("COUNT"), counter.get()); + super.update(executionContext); + executionContext.putInt(getExecutionContextKey("COUNT"), counter.get()); } } 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 7fb019bf6..6d8b28d17 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 @@ -19,10 +19,8 @@ package org.springframework.batch.item.support; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStreamException; -import org.springframework.batch.item.ItemStreamReader; import org.springframework.batch.item.ParseException; import org.springframework.batch.item.UnexpectedInputException; -import org.springframework.batch.item.util.ExecutionContextUserSupport; import org.springframework.util.Assert; /** @@ -34,7 +32,7 @@ import org.springframework.util.Assert; * * @author Robert Kasanicky */ -public abstract class AbstractItemCountingItemStreamItemReader implements ItemStreamReader { +public abstract class AbstractItemCountingItemStreamItemReader extends AbstractItemStreamItemReader { private static final String READ_COUNT = "read.count"; @@ -44,8 +42,6 @@ public abstract class AbstractItemCountingItemStreamItemReader implements Ite private int maxItemCount = Integer.MAX_VALUE; - private ExecutionContextUserSupport ecSupport = new ExecutionContextUserSupport(); - private boolean saveState = true; /** @@ -121,6 +117,7 @@ public abstract class AbstractItemCountingItemStreamItemReader implements Ite @Override public void close() throws ItemStreamException { + super.close(); currentItemCount = 0; try { doClose(); @@ -132,7 +129,7 @@ public abstract class AbstractItemCountingItemStreamItemReader implements Ite @Override public void open(ExecutionContext executionContext) throws ItemStreamException { - + super.open(executionContext); try { doOpen(); } @@ -143,12 +140,12 @@ public abstract class AbstractItemCountingItemStreamItemReader implements Ite return; } - if (executionContext.containsKey(ecSupport.getKey(READ_COUNT_MAX))) { - maxItemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT_MAX)); + if (executionContext.containsKey(getExecutionContextKey(READ_COUNT_MAX))) { + maxItemCount = executionContext.getInt(getExecutionContextKey(READ_COUNT_MAX)); } - if (executionContext.containsKey(ecSupport.getKey(READ_COUNT))) { - int itemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT)); + if (executionContext.containsKey(getExecutionContextKey(READ_COUNT))) { + int itemCount = executionContext.getInt(getExecutionContextKey(READ_COUNT)); if (itemCount < maxItemCount) { try { @@ -166,20 +163,17 @@ public abstract class AbstractItemCountingItemStreamItemReader implements Ite @Override public void update(ExecutionContext executionContext) throws ItemStreamException { + super.update(executionContext); if (saveState) { Assert.notNull(executionContext, "ExecutionContext must not be null"); - executionContext.putInt(ecSupport.getKey(READ_COUNT), currentItemCount); + executionContext.putInt(getExecutionContextKey(READ_COUNT), currentItemCount); if (maxItemCount < Integer.MAX_VALUE) { - executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX), maxItemCount); + executionContext.putInt(getExecutionContextKey(READ_COUNT_MAX), maxItemCount); } } } - protected ExecutionContextUserSupport getExecutionContextUserSupport() { - return ecSupport; - } - /** * The name of the component which will be used as a stem for keys in the * {@link ExecutionContext}. Subclasses should provide a default value, e.g. @@ -188,7 +182,7 @@ public abstract class AbstractItemCountingItemStreamItemReader implements Ite * @param name the name for the component */ public void setName(String name) { - ecSupport.setName(name); + this.setExecutionContextName(name); } /** diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemReader.java index 606f9c73b..96132ee34 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemReader.java @@ -17,6 +17,7 @@ package org.springframework.batch.item.support; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStreamReader; import org.springframework.batch.item.ItemStreamSupport; @@ -25,6 +26,6 @@ import org.springframework.batch.item.ItemStreamSupport; * @author Dave Syer * */ -public abstract class AbstractItemStreamItemReader extends ItemStreamSupport implements ItemReader { +public abstract class AbstractItemStreamItemReader extends ItemStreamSupport implements ItemStreamReader { } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemWriter.java index 1d9e1e53d..517c60eac 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemStreamItemWriter.java @@ -17,6 +17,7 @@ package org.springframework.batch.item.support; import org.springframework.batch.item.ItemStreamSupport; +import org.springframework.batch.item.ItemStreamWriter; import org.springframework.batch.item.ItemWriter; @@ -25,6 +26,6 @@ import org.springframework.batch.item.ItemWriter; * @author Dave Syer * */ -public abstract class AbstractItemStreamItemWriter extends ItemStreamSupport implements ItemWriter { +public abstract class AbstractItemStreamItemWriter extends ItemStreamSupport implements ItemStreamWriter { } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java index 9ada94cb1..782191cbf 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java @@ -41,7 +41,7 @@ import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.WriteFailedException; import org.springframework.batch.item.file.ResourceAwareItemWriterItemStream; -import org.springframework.batch.item.util.ExecutionContextUserSupport; +import org.springframework.batch.item.support.AbstractItemStreamItemWriter; import org.springframework.batch.item.util.FileUtils; import org.springframework.batch.item.xml.stax.NoStartEndDocumentStreamWriter; import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter; @@ -69,7 +69,7 @@ import org.springframework.util.StringUtils; * @author Michael Minella * */ -public class StaxEventItemWriter extends ExecutionContextUserSupport implements +public class StaxEventItemWriter extends AbstractItemStreamItemWriter implements ResourceAwareItemWriterItemStream, InitializingBean { private static final Log log = LogFactory.getLog(StaxEventItemWriter.class); @@ -143,7 +143,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen private boolean forceSync; public StaxEventItemWriter() { - setName(ClassUtils.getShortName(StaxEventItemWriter.class)); + setExecutionContextName(ClassUtils.getShortName(StaxEventItemWriter.class)); } /** @@ -343,7 +343,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen */ @Override public void open(ExecutionContext executionContext) { - + super.open(executionContext); + Assert.notNull(resource, "The resource must be set"); long startAtPosition = 0; @@ -351,8 +352,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen // if restart data is provided, restart from provided offset // otherwise start from beginning - if (executionContext.containsKey(getKey(RESTART_DATA_NAME))) { - startAtPosition = executionContext.getLong(getKey(RESTART_DATA_NAME)); + if (executionContext.containsKey(getExecutionContextKey(RESTART_DATA_NAME))) { + startAtPosition = executionContext.getLong(getExecutionContextKey(RESTART_DATA_NAME)); restarted = true; } @@ -576,7 +577,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen */ @Override public void close() { - + super.close(); + XMLEventFactory factory = createXmlEventFactory(); try { delegateEventWriter.add(factory.createCharacters("")); @@ -665,11 +667,11 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen */ @Override public void update(ExecutionContext executionContext) { - + super.update(executionContext); if (saveState) { Assert.notNull(executionContext, "ExecutionContext must not be null"); - executionContext.putLong(getKey(RESTART_DATA_NAME), getPosition()); - executionContext.putLong(getKey(WRITE_STATISTICS_NAME), currentRecordCount); + executionContext.putLong(getExecutionContextKey(RESTART_DATA_NAME), getPosition()); + executionContext.putLong(getExecutionContextKey(WRITE_STATISTICS_NAME), currentRecordCount); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterFlatFileTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterFlatFileTests.java index 20a93ea33..d5c87c00f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterFlatFileTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemWriterFlatFileTests.java @@ -86,12 +86,12 @@ public class MultiResourceItemWriterFlatFileTests extends AbstractMultiResourceI super.setUp(delegate); tested.update(executionContext); - assertEquals(0, executionContext.getInt(tested.getKey("resource.item.count"))); - assertEquals(1, executionContext.getInt(tested.getKey("resource.index"))); + assertEquals(0, executionContext.getInt(tested.getExecutionContextKey("resource.item.count"))); + assertEquals(1, executionContext.getInt(tested.getExecutionContextKey("resource.index"))); tested.write(Arrays.asList("1", "2", "3")); tested.update(executionContext); - assertEquals(0, executionContext.getInt(tested.getKey("resource.item.count"))); - assertEquals(2, executionContext.getInt(tested.getKey("resource.index"))); + assertEquals(0, executionContext.getInt(tested.getExecutionContextKey("resource.item.count"))); + assertEquals(2, executionContext.getInt(tested.getExecutionContextKey("resource.index"))); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourcesItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourcesItemReaderTests.java index 1d934c293..a4b95067a 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourcesItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/ResourcesItemReaderTests.java @@ -34,7 +34,7 @@ public class ResourcesItemReaderTests { @Test public void testReadAfterOpen() throws Exception { ExecutionContext executionContext = new ExecutionContext(); - executionContext.putInt(reader.getKey("COUNT"), 1); + executionContext.putInt(reader.getExecutionContextKey("COUNT"), 1); reader.open(executionContext); assertNotNull(reader.read()); assertNull(reader.read()); @@ -46,7 +46,7 @@ public class ResourcesItemReaderTests { assertNotNull(reader.read()); reader.update(executionContext); - assertEquals(1, executionContext.getInt(reader.getKey("COUNT"))); + assertEquals(1, executionContext.getInt(reader.getExecutionContextKey("COUNT"))); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemStreamTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemStreamTests.java index 02d8958b4..53ad92105 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemStreamTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemStreamTests.java @@ -38,8 +38,9 @@ public class CompositeItemStreamTests extends TestCase { public void testRegisterAndOpen() { ItemStreamSupport stream = new ItemStreamSupport() { - @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { + @Override + public void open(ExecutionContext executionContext) { + super.open(executionContext); list.add("bar"); } }; @@ -51,7 +52,8 @@ public class CompositeItemStreamTests extends TestCase { public void testRegisterTwice() { ItemStreamSupport stream = new ItemStreamSupport() { @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { + public void open(ExecutionContext executionContext) { + super.open(executionContext); list.add("bar"); } }; @@ -63,8 +65,9 @@ public class CompositeItemStreamTests extends TestCase { public void testMark() { manager.register(new ItemStreamSupport() { - @Override + @Override public void update(ExecutionContext executionContext) { + super.update(executionContext); list.add("bar"); } }); @@ -74,8 +77,9 @@ public class CompositeItemStreamTests extends TestCase { public void testClose() { manager.register(new ItemStreamSupport() { - @Override - public void close() throws ItemStreamException { + @Override + public void close() { + super.close(); list.add("bar"); } }); @@ -85,8 +89,9 @@ public class CompositeItemStreamTests extends TestCase { public void testCloseDoesNotUnregister() { manager.setStreams(new ItemStream[] { new ItemStreamSupport() { - @Override - public void open(ExecutionContext executionContext) throws ItemStreamException { + @Override + public void open(ExecutionContext executionContext) { + super.open(executionContext); list.add("bar"); } } });