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"); } } });