From 15cffeb925c4a9b2a668a3f215174677bc4eecca Mon Sep 17 00:00:00 2001 From: dsyer Date: Thu, 31 Jan 2008 09:54:38 +0000 Subject: [PATCH] IN PROGRESS - issue BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering) http://jira.springframework.org/browse/BATCH-7 Lazy-initialisation solution to the "who's in step scope" issue --- .../batch/core/domain/Step.java | 2 +- .../batch/core/domain/StepInstance.java | 4 +- .../batch/core/domain/StepSupport.java | 10 +- .../batch/core/repository/JobRepository.java | 28 +-- .../batch/core/domain/StepInstanceTests.java | 12 +- .../batch/core/domain/StepSupportTests.java | 6 +- .../repository/SimpleJobRepository.java | 8 +- .../execution/repository/dao/JdbcStepDao.java | 6 +- .../execution/repository/dao/MapStepDao.java | 2 +- .../execution/scope/SimpleStepContext.java | 91 ++++++++-- .../batch/execution/scope/StepContext.java | 17 +- .../step/simple/SimpleStepExecutor.java | 72 ++++---- .../RestartableItemOrientedTasklet.java | 82 +-------- .../repository/SimpleJobRepositoryTests.java | 16 +- .../repository/dao/AbstractStepDaoTests.java | 4 +- .../repository/dao/MapStepDaoTests.java | 6 +- .../scope/SimpleStepContextTests.java | 41 ++++- .../simple/SimpleStepConfigurationTests.java | 8 +- .../step/simple/SimpleStepExecutorTests.java | 107 ++++++------ .../tasklet/ItemOrientedTaskletTests.java | 7 +- .../RestartableItemOrientedTaskletTests.java | 160 ------------------ .../resources/simple-container-definition.xml | 2 +- .../io/cursor/HibernateCursorItemReader.java | 2 +- .../batch/io/cursor/JdbcCursorItemReader.java | 2 +- .../io/driving/DrivingQueryItemReader.java | 4 +- .../batch/io/driving/KeyGenerator.java | 2 +- ...a => ColumnMapStreamContextRowMapper.java} | 8 +- .../driving/support/IbatisKeyGenerator.java | 2 +- .../MultipleColumnJdbcKeyGenerator.java | 4 +- .../support/SingleColumnJdbcKeyGenerator.java | 2 +- .../io/file/DefaultFlatFileItemReader.java | 2 +- .../batch/io/file/FlatFileItemWriter.java | 4 +- .../io/file/SimpleFlatFileItemReader.java | 3 +- .../batch/io/xml/StaxEventItemReader.java | 4 +- .../batch/io/xml/StaxEventItemWriter.java | 4 +- .../batch/item/ItemReader.java | 2 +- .../batch/item/ItemStream.java | 20 +-- .../batch/item/StreamContextProvider.java | 32 ++++ .../batch/item/StreamException.java | 26 +++ .../batch/item/reader/AbstractItemReader.java | 3 +- .../item/reader/DelegatingItemReader.java | 20 +-- .../batch/item/reader/ItemReaderAdapter.java | 3 +- .../item/stream/CompositeItemStream.java | 158 ----------------- .../item/stream/SimpleStreamManager.java | 150 ++++++++++++++++ .../batch/item/stream/StreamManager.java | 62 +++++++ .../item/writer/CompositeItemWriter.java | 12 +- .../item/writer/DelegatingItemWriter.java | 6 +- .../driving/DrivingQueryItemReaderTests.java | 6 +- .../batch/io/driving/FooInputSource.java | 9 +- .../ColumnMapRestartDataRowMapperTests.java | 6 +- ...olumnJdbcKeyGeneratorIntegrationTests.java | 16 +- ...olumnJdbcKeyGeneratorIntegrationTests.java | 10 +- .../file/DefaultFlatFileItemReaderTests.java | 4 +- .../io/file/FlatFileItemWriterTests.java | 8 +- ...bstractJdbcItemReaderIntegrationTests.java | 4 +- ...tDataSourceItemReaderIntegrationTests.java | 6 +- .../xml/StaxEventReaderItemReaderTests.java | 2 +- .../xml/StaxEventWriterItemWriterTests.java | 2 +- .../reader/DelegatingItemReaderTests.java | 6 +- .../item/writer/CompositeItemWriterTests.java | 81 --------- .../writer/ItemWriterItemProcessorTests.java | 10 +- .../src/main/resources/jobs/footballJob.xml | 3 - .../src/main/resources/jobs/parallelJob.xml | 2 +- .../src/main/resources/jobs/restartSample.xml | 4 +- .../resources/simple-container-definition.xml | 5 +- .../batch/sample/RestartFunctionalTests.java | 2 +- 66 files changed, 653 insertions(+), 761 deletions(-) delete mode 100644 spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTaskletTests.java rename spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/{ColumnMapRestartDataRowMapper.java => ColumnMapStreamContextRowMapper.java} (92%) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContextProvider.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java delete mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/CompositeItemStream.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Step.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Step.java index 79a0723e2..90966c0b1 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Step.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Step.java @@ -43,7 +43,7 @@ public interface Step { * Flag to indicate if restart data needs to be saved for this step. * @return true if restart data should be saved */ - boolean isSaveRestartData(); + boolean isSaveStreamContext(); /** * @return the number of times a job can be started with the same diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepInstance.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepInstance.java index cb57b37cc..c01e09c9e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepInstance.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepInstance.java @@ -87,11 +87,11 @@ public class StepInstance extends Entity { this.stepExecutionCount = stepExecutionCount; } - public StreamContext getRestartData() { + public StreamContext getStreamContext() { return streamContext; } - public void setRestartData(StreamContext streamContext) { + public void setStreamContext(StreamContext streamContext) { this.streamContext = streamContext; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepSupport.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepSupport.java index d8920ff25..8c869f37a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepSupport.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepSupport.java @@ -34,7 +34,7 @@ public class StepSupport implements Step, BeanNameAware { private boolean allowStartIfComplete; - private boolean saveRestartData = false; + private boolean saveStreamContext = false; /** * Default constructor for {@link StepSupport}. @@ -121,12 +121,12 @@ public class StepSupport implements Step, BeanNameAware { this.allowStartIfComplete = allowStartIfComplete; } - public void setSaveRestartData(boolean saveRestartData) { - this.saveRestartData = saveRestartData; + public void setSaveStreamContext(boolean saveStreamContext) { + this.saveStreamContext = saveStreamContext; } - public boolean isSaveRestartData() { - return saveRestartData; + public boolean isSaveStreamContext() { + return saveStreamContext; } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java index edf60ab55..02f127bce 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java @@ -22,14 +22,16 @@ import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobParameters; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; +import org.springframework.batch.item.StreamContext; /** *

* Repository for storing batch jobs and steps. Before using any methods, a Job * must first be obtained using the findOrCreateJob method. Once a Job and it's * related steps are obtained, they can be updated. It should be noted that any - * reconstituted steps are expected to contain restart data if the - * RestartPolicy associated with the step returns true, and RestartData exists. + * reconstituted steps are expected to contain restart data if the step + * says it wants to be restored after a restart, and {@link StreamContext} + * exists. *

* * Once a Job/Steps has been created, Job and Step executions can be created and @@ -47,23 +49,21 @@ import org.springframework.batch.core.domain.StepInstance; public interface JobRepository { /** - * Find or create a job execution for a given {@link JobIdentifier} and configuration. - * If the job that is uniquely identified by {@link JobIdentifier} already - * exists, its persisted values (including ID) will be returned in a new - * {@link JobInstance} object. If no previous run is found, a new job will - * be created and returned. + * Find or create a job execution for a given {@link JobIdentifier} and + * configuration. If the job that is uniquely identified by + * {@link JobIdentifier} already exists, its persisted values (including ID) + * will be returned in a new {@link JobInstance} object. If no previous run + * is found, a new job will be created and returned. * @param jobParameters the runtime parameters for the job - * @param jobConfiguration - * describes the configuration for this job + * @param jobConfiguration describes the configuration for this job * * @return a valid job execution for the identifier provided - * @throws JobExecutionAlreadyRunningException - * if there is a {@link JobExecution} alrady running for the - * job instance that would otherwise be returned + * @throws JobExecutionAlreadyRunningException if there is a + * {@link JobExecution} alrady running for the job instance that would + * otherwise be returned * */ - public JobExecution createJobExecution(Job job, - JobParameters jobParameters) + public JobExecution createJobExecution(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException; /** diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepInstanceTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepInstanceTests.java index 09d48bbf4..58dbdf8ee 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepInstanceTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepInstanceTests.java @@ -46,15 +46,15 @@ public class StepInstanceTests extends TestCase { } /** - * Test method for {@link org.springframework.batch.core.domain.StepInstance#getRestartData()}. + * Test method for {@link org.springframework.batch.core.domain.StepInstance#getStreamContext()}. */ - public void testGetRestartData() { - assertNotNull(instance.getRestartData()); - assertTrue(instance.getRestartData().getProperties().isEmpty()); - instance.setRestartData(new GenericStreamContext(new Properties() {{ + public void testGetStreamContext() { + assertNotNull(instance.getStreamContext()); + assertTrue(instance.getStreamContext().getProperties().isEmpty()); + instance.setStreamContext(new GenericStreamContext(new Properties() {{ setProperty("foo", "bar"); }})); - assertEquals("bar", instance.getRestartData().getProperties().getProperty("foo")); + assertEquals("bar", instance.getStreamContext().getProperties().getProperty("foo")); } /** diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepSupportTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepSupportTests.java index def1d5d28..aab413690 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepSupportTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepSupportTests.java @@ -64,9 +64,9 @@ public class StepSupportTests extends TestCase { } public void testSaveRestartFlag() throws Exception { - assertEquals(false, configuration.isSaveRestartData()); - configuration.setSaveRestartData(true); - assertEquals(true, configuration.isSaveRestartData()); + assertEquals(false, configuration.isSaveStreamContext()); + configuration.setSaveStreamContext(true); + assertEquals(true, configuration.isSaveStreamContext()); } /** diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java index 29e12bbff..b0ed04ad0 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java @@ -303,8 +303,8 @@ public class SimpleJobRepository implements JobRepository { Step step = (Step) i.next(); StepInstance stepInstance = stepDao.createStep(job, step.getName()); // Ensure valid restart data is being returned. - if (stepInstance.getRestartData() == null || stepInstance.getRestartData().getProperties() == null) { - stepInstance.setRestartData(new GenericStreamContext(new Properties())); + if (stepInstance.getStreamContext() == null || stepInstance.getStreamContext().getProperties() == null) { + stepInstance.setStreamContext(new GenericStreamContext(new Properties())); } stepInstances.add(stepInstance); } @@ -326,8 +326,8 @@ public class SimpleJobRepository implements JobRepository { step.setStepExecutionCount(stepDao.getStepExecutionCount(step)); // Ensure valid restart data is being returned. - if (step.getRestartData() == null || step.getRestartData().getProperties() == null) { - step.setRestartData(new GenericStreamContext(new Properties())); + if (step.getStreamContext() == null || step.getStreamContext().getProperties() == null) { + step.setStreamContext(new GenericStreamContext(new Properties())); } stepInstances.add(step); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java index 35aaaf447..12a1a7cc6 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java @@ -161,7 +161,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { StepInstance step = new StepInstance(new Long(rs.getLong(1))); step.setStatus(BatchStatus.getStatus(rs.getString(2))); - step.setRestartData(new GenericStreamContext(PropertiesConverter.stringToProperties(rs.getString(3)))); + step.setStreamContext(new GenericStreamContext(PropertiesConverter.stringToProperties(rs.getString(3)))); return step; } @@ -243,7 +243,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { StepInstance step = new StepInstance(job, rs.getString(2), new Long(rs.getLong(1))); String status = rs.getString(3); step.setStatus(BatchStatus.getStatus(status)); - step.setRestartData(new GenericStreamContext(PropertiesConverter.stringToProperties(rs.getString(3)))); + step.setStreamContext(new GenericStreamContext(PropertiesConverter.stringToProperties(rs.getString(3)))); return step; } }; @@ -422,7 +422,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { Assert.notNull(step.getId(), "Step Id cannot be null."); Properties restartProps = null; - StreamContext streamContext = step.getRestartData(); + StreamContext streamContext = step.getStreamContext(); if (streamContext != null) { restartProps = streamContext.getProperties(); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java index ab363a75e..caffaf5d5 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java @@ -80,7 +80,7 @@ public class MapStepDao implements StepDao { return new ArrayList(steps); } - public StreamContext getRestartData(Long stepId) { + public StreamContext getStreamContext(Long stepId) { return (StreamContext) restartsById.get(stepId); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java index ba8b4b16e..7998cb12a 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java @@ -25,6 +25,10 @@ import java.util.Properties; import java.util.Set; import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.io.exception.BatchCriticalException; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.context.SynchronizedAttributeAccessor; import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.batch.statistics.StatisticsService; @@ -35,53 +39,72 @@ import org.springframework.batch.statistics.StatisticsService; * @author Dave Syer * */ -public class SimpleStepContext extends SynchronizedAttributeAccessor implements - StepContext, StatisticsProvider { +public class SimpleStepContext extends SynchronizedAttributeAccessor implements StepContext { private Map callbacks = new HashMap(); + private StepContext parent; + private StepExecution stepExecution; + private StatisticsService statisticsService; + private StreamManager streamManager; + + private StreamContext streamContext; + /** * Default constructor. */ public SimpleStepContext(StepExecution stepExecution) { - this(stepExecution, null, null); + this(stepExecution, null, null, null); } /** * Default constructor. */ public SimpleStepContext(StepExecution stepExecution, StepContext parent) { - this(stepExecution, parent, null); + this(stepExecution, parent, null, null); } /** * @param object */ - public SimpleStepContext(StepExecution stepExecution, StepContext parent, StatisticsService statisticsService) { + public SimpleStepContext(StepExecution stepExecution, StepContext parent, StatisticsService statisticsService, + StreamManager streamManager) { super(); this.parent = parent; this.statisticsService = statisticsService; + this.streamManager = streamManager; this.stepExecution = stepExecution; } - - /* (non-Javadoc) - * @see org.springframework.batch.repeat.context.SynchronizedAttributeAccessor#setAttribute(java.lang.String, java.lang.Object) + + /* + * (non-Javadoc) + * @see org.springframework.batch.repeat.context.SynchronizedAttributeAccessor#setAttribute(java.lang.String, + * java.lang.Object) */ public void setAttribute(String name, Object value) { super.setAttribute(name, value); - if (statisticsService!=null && (value instanceof StatisticsProvider)) { + if (statisticsService != null && (value instanceof StatisticsProvider)) { statisticsService.register(this, (StatisticsProvider) value); } + if (streamManager != null && (value instanceof ItemStream)) { + ItemStream stream = (ItemStream) value; + streamManager.register(this, stream); + stream.open(); + if (streamContext != null) { + stream.restoreFrom(streamContext); + } + } } - - /* (non-Javadoc) + + /* + * (non-Javadoc) * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics() */ public Properties getStatistics() { - if (statisticsService==null) { + if (statisticsService == null) { return new Properties(); } return statisticsService.getStatistics(this); @@ -100,13 +123,13 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements * (non-Javadoc) * * @see org.springframework.batch.repeat.RepeatContext#registerDestructionCallback(java.lang.String, - * java.lang.Runnable) + * java.lang.Runnable) */ /* * (non-Javadoc) * * @see org.springframework.batch.execution.scope.StepContext#registerDestructionCallback(java.lang.String, - * java.lang.Runnable) + * java.lang.Runnable) */ public void registerDestructionCallback(String name, Runnable callback) { synchronized (callbacks) { @@ -119,13 +142,23 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements } } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.batch.execution.scope.StepContext#close() */ public void close() { List errors = new ArrayList(); + try { + if (streamManager != null) { + streamManager.close(this); + } + } + catch (Exception t) { + errors.add(t); + } + Set copy; synchronized (callbacks) { @@ -152,7 +185,8 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements */ try { callback.run(); - } catch (RuntimeException t) { + } + catch (RuntimeException t) { errors.add(t); } } @@ -163,7 +197,14 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements return; } - throw (RuntimeException) errors.get(0); + Exception error = (Exception) errors.get(0); + if (error instanceof RuntimeException) { + throw (RuntimeException) error; + } + else { + throw new BatchCriticalException("Could not close step context, rethrowing first of " + errors.size() + + " execptions.", error); + } } /* @@ -175,4 +216,20 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements return stepExecution; } + /* + * (non-Javadoc) + * @see org.springframework.batch.item.ItemStream#getStreamContext() + */ + public StreamContext getStreamContext() { + return streamManager.getStreamContext(this); + } + + /* + * (non-Javadoc) + * @see org.springframework.batch.execution.scope.StepContext#setInitialStreamContext(org.springframework.batch.item.StreamContext) + */ + public void setInitialStreamContext(StreamContext streamContext) { + this.streamContext = streamContext; + } + } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java index 0df4f248f..a4a8f3ee0 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java @@ -16,6 +16,9 @@ package org.springframework.batch.execution.scope; import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamContextProvider; import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.core.AttributeAccessor; @@ -25,7 +28,7 @@ import org.springframework.core.AttributeAccessor; * @author Dave Syer * */ -public interface StepContext extends AttributeAccessor, StatisticsProvider { +public interface StepContext extends AttributeAccessor, StreamContextProvider, StatisticsProvider { /** * Accessor for the {@link StepExecution} associated with the currently @@ -34,7 +37,7 @@ public interface StepContext extends AttributeAccessor, StatisticsProvider { * @return the {@link StepExecution} associated with the current step */ StepExecution getStepExecution(); - + /** * Accessor for the parent context. * @@ -51,4 +54,14 @@ public interface StepContext extends AttributeAccessor, StatisticsProvider { * Clean up any resources held during the context of the step. */ void close(); + + /** + * Provide the stream context that will be needed to restore + * {@link ItemStream} instances. If this is not set the streams will simply + * not be initialised and repositioned for restart (which is sometimes + * desirable). + * + * @param streamContext + */ + void setInitialStreamContext(StreamContext streamContext); } \ No newline at end of file diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index 94098b40d..f6e1d0a7a 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -34,8 +34,8 @@ import org.springframework.batch.execution.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.io.Skippable; import org.springframework.batch.io.exception.BatchCriticalException; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.stream.SimpleStreamManager; +import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; @@ -97,6 +97,8 @@ public class SimpleStepExecutor { private AbstractStep step; + private StreamManager streamManager = new SimpleStreamManager(); + /** * Package private constructor so the factory can create a the executor. */ @@ -118,6 +120,20 @@ public class SimpleStepExecutor { this.statisticsService = statisticsService; } + /** + * Public setter for the {@link StreamManager}. This will be used to create + * the {@link StepContext}, and hence any component that is a + * {@link StatisticsProvider} and in step scope will be registered with the + * service. The {@link StepContext} is then a source of aggregate statistics + * for the step. + * + * @param streamManager the {@link StreamManager} to set. Default is a + * {@link SimpleStreamManager}. + */ + public void setStreamManager(StreamManager streamManager) { + this.streamManager = streamManager; + } + /** * Injected strategy for transaction management * @param transactionManager @@ -173,8 +189,7 @@ public class SimpleStepExecutor { * execution * @see StepExecutor#execute(StepExecution) */ - public void execute(final StepExecution stepExecution) throws BatchCriticalException, - StepInterruptedException { + public void execute(final StepExecution stepExecution) throws BatchCriticalException, StepInterruptedException { final StepInstance stepInstance = stepExecution.getStep(); boolean isRestart = stepInstance.getStepExecutionCount() > 0 ? true : false; @@ -182,24 +197,25 @@ public class SimpleStepExecutor { ExitStatus status = ExitStatus.FAILED; - StepContext parentStepScopeContext = StepSynchronizationManager.getContext(); - final StepContext stepScopeContext = new SimpleStepContext(stepExecution, parentStepScopeContext, - statisticsService); - StepSynchronizationManager.register(stepScopeContext); + StepContext parentStepContext = StepSynchronizationManager.getContext(); + final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, statisticsService, + streamManager); + StepSynchronizationManager.register(stepContext); // Add the job identifier so that it can be used to identify // the conversation in StepScope - stepScopeContext.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution().getId()); + stepContext.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution().getId()); + + final boolean saveRestartData = step.isSaveStreamContext(); + + if (saveRestartData && isRestart) { + stepContext.setInitialStreamContext(stepInstance.getStreamContext()); + } try { + stepExecution.setStartTime(new Date(System.currentTimeMillis())); updateStatus(stepExecution, BatchStatus.STARTED); - final boolean saveRestartData = step.isSaveRestartData(); - - if (saveRestartData && isRestart) { - restoreFromRestartData(tasklet, stepInstance.getRestartData()); - } - status = stepOperations.iterate(new RepeatCallback() { public ExitStatus doInIteration(final RepeatContext context) throws Exception { @@ -229,8 +245,8 @@ public class SimpleStepExecutor { // TODO: check that stepExecution can // aggregate these contributions if they - // come in asnchronously. - Properties statistics = stepScopeContext.getStatistics(); + // come in asynchronously. + Properties statistics = stepContext.getStatistics(); contribution.setStatistics(statistics); contribution.incrementCommitCount(); // Apply the contribution to the step @@ -238,7 +254,7 @@ public class SimpleStepExecutor { stepExecution.apply(contribution); if (saveRestartData) { - stepInstance.setRestartData(getRestartData(tasklet)); + stepInstance.setStreamContext(stepContext.getStreamContext()); jobRepository.update(stepInstance); } jobRepository.saveOrUpdate(stepExecution); @@ -388,26 +404,6 @@ public class SimpleStepExecutor { return exitStatus; } - /** - * @param tasklet - * @return restart data from the {@link Tasklet} if it is - * {@link ItemStream} - */ - private StreamContext getRestartData(Tasklet tasklet) { - if (tasklet instanceof ItemStream) { - return ((ItemStream) tasklet).getRestartData(); - } - else { - return null; - } - } - - private void restoreFromRestartData(Tasklet tasklet, StreamContext streamContext) { - if (tasklet instanceof ItemStream && streamContext != null) { - ((ItemStream) tasklet).restoreFrom(streamContext); - } - } - /** * Setter for the {@link StepInterruptionPolicy}. The policy is used to * check whether an external request has been made to interrupt the job diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTasklet.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTasklet.java index 345c6c131..0ae4e17e5 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTasklet.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTasklet.java @@ -16,14 +16,11 @@ package org.springframework.batch.execution.tasklet; -import java.util.Properties; - import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.StreamContext; -import org.springframework.batch.item.stream.GenericStreamContext; -import org.springframework.batch.support.PropertiesConverter; +import org.springframework.batch.item.StreamException; /** * An extension of {@link ItemOrientedTasklet} that delegates calls to @@ -40,95 +37,30 @@ import org.springframework.batch.support.PropertiesConverter; public class RestartableItemOrientedTasklet extends ItemOrientedTasklet implements ItemStream { /** - * @see ItemStream#getRestartData() + * @see ItemStream#getStreamContext() */ - public StreamContext getRestartData() { - - StreamContext itemProviderRestartData = null; - StreamContext itemProcessorRestartData = null; - - if (itemProvider instanceof ItemStream) { - itemProviderRestartData = ((ItemStream) itemProvider).getRestartData(); - } - - if (itemWriter instanceof ItemStream) { - itemProcessorRestartData = ((ItemStream) itemWriter).getRestartData(); - } - - RestartableItemOrientedTaskletRestartData restartData = new RestartableItemOrientedTaskletRestartData(itemProviderRestartData, itemProcessorRestartData); - - return restartData; + public StreamContext getStreamContext() { + throw new UnsupportedOperationException("This class is not used"); } /** * @see ItemStream#restoreFrom(StreamContext) */ public void restoreFrom(StreamContext data) { - if (data == null || data.getProperties() == null) - return; - - RestartableItemOrientedTaskletRestartData moduleRestartData; - - if (data instanceof RestartableItemOrientedTaskletRestartData) { - moduleRestartData = (RestartableItemOrientedTaskletRestartData) data; - } - else { - moduleRestartData = new RestartableItemOrientedTaskletRestartData(data.getProperties()); - } - - if (itemProvider instanceof ItemStream) { - ((ItemStream) itemProvider).restoreFrom(moduleRestartData.readerData); - } - if (itemWriter instanceof ItemStream) { - ((ItemStream) itemWriter).restoreFrom(moduleRestartData.writerData); - } - } - - private class RestartableItemOrientedTaskletRestartData implements StreamContext { - - private static final String READER_KEY = "DATA_PROVIDER"; - - private static final String WRITER_KEY = "DATA_PROCESSOR"; - - private StreamContext readerData; - - private StreamContext writerData; - - public RestartableItemOrientedTaskletRestartData(StreamContext providerData, StreamContext writerData) { - this.readerData = providerData; - this.writerData = writerData; - } - - public RestartableItemOrientedTaskletRestartData(Properties data) { - readerData = new GenericStreamContext(PropertiesConverter - .stringToProperties(data.getProperty(READER_KEY))); - writerData = new GenericStreamContext(PropertiesConverter.stringToProperties(data - .getProperty(WRITER_KEY))); - } - - public Properties getProperties() { - Properties props = new Properties(); - if (readerData != null) { - props.setProperty(READER_KEY, PropertiesConverter.propertiesToString(readerData.getProperties())); - } - if (writerData != null) { - props.setProperty(WRITER_KEY, PropertiesConverter.propertiesToString(writerData.getProperties())); - } - return props; - } + throw new UnsupportedOperationException("This class is not used"); } /* (non-Javadoc) * @see org.springframework.batch.item.ItemStream#open() */ - public void open() throws Exception { + public void open() throws StreamException { throw new UnsupportedOperationException("Not implemented."); } /* (non-Javadoc) * @see org.springframework.batch.item.ItemStream#close() */ - public void close() throws Exception { + public void close() throws StreamException { throw new UnsupportedOperationException("Not implemented."); } } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java index 84e08f8d9..42c2c89bc 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java @@ -405,10 +405,10 @@ public class SimpleJobRepositoryTests extends TestCase { jobDao.createJobInstance(jobConfiguration.getName(), jobParameters); jobDaoControl.setReturnValue(databaseJob); stepDao.createStep(databaseJob, "TestStep1"); - databaseStep1.setRestartData(null); + databaseStep1.setStreamContext(null); stepDaoControl.setReturnValue(databaseStep1); stepDao.createStep(databaseJob, "TestStep2"); - databaseStep2.setRestartData(new GenericStreamContext(null)); + databaseStep2.setStreamContext(new GenericStreamContext(null)); stepDaoControl.setReturnValue(databaseStep2); jobDao.save(new JobExecution(databaseJob)); jobDaoControl.setMatcher(new ArgumentsMatcher(){ @@ -426,10 +426,10 @@ public class SimpleJobRepositoryTests extends TestCase { Iterator it = jobSteps.iterator(); StepInstance step = (StepInstance) it.next(); assertTrue(step.equals(databaseStep1)); - assertTrue(step.getRestartData().getProperties().isEmpty()); + assertTrue(step.getStreamContext().getProperties().isEmpty()); step = (StepInstance) it.next(); assertTrue(step.equals(databaseStep2)); - assertTrue(step.getRestartData().getProperties().isEmpty()); + assertTrue(step.getStreamContext().getProperties().isEmpty()); } public void testFindStepsFixesInvalidRestartData() throws Exception{ @@ -438,12 +438,12 @@ public class SimpleJobRepositoryTests extends TestCase { jobs.add(databaseJob); jobDaoControl.setReturnValue(jobs); stepDao.findStep(databaseJob, "TestStep1"); - databaseStep1.setRestartData(null); + databaseStep1.setStreamContext(null); stepDaoControl.setReturnValue(databaseStep1); stepDao.getStepExecutionCount(databaseStep1); stepDaoControl.setReturnValue(1); stepDao.findStep(databaseJob, "TestStep2"); - databaseStep2.setRestartData(new GenericStreamContext(null)); + databaseStep2.setStreamContext(new GenericStreamContext(null)); stepDaoControl.setReturnValue(databaseStep2); stepDao.getStepExecutionCount(databaseStep2); stepDaoControl.setReturnValue(1); @@ -470,9 +470,9 @@ public class SimpleJobRepositoryTests extends TestCase { Iterator it = jobSteps.iterator(); StepInstance step = (StepInstance) it.next(); assertTrue(step.equals(databaseStep1)); - assertTrue(step.getRestartData().getProperties().isEmpty()); + assertTrue(step.getStreamContext().getProperties().isEmpty()); step = (StepInstance) it.next(); - assertTrue(step.getRestartData().getProperties().isEmpty()); + assertTrue(step.getStreamContext().getProperties().isEmpty()); assertTrue(step.equals(databaseStep2)); } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java index 4cc9a58cf..5ceb44ed7 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java @@ -153,11 +153,11 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour Properties data = new Properties(); data.setProperty("restart.key1", "restartData"); StreamContext streamContext = new GenericStreamContext(data); - step1.setRestartData(streamContext); + step1.setStreamContext(streamContext); stepDao.update(step1); StepInstance tempStep = stepDao.findStep(jobInstance, step1.getName()); assertEquals(tempStep, step1); - assertEquals(tempStep.getRestartData().getProperties().toString(), + assertEquals(tempStep.getStreamContext().getProperties().toString(), streamContext.getProperties().toString()); } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java index c5fac91ae..276e8cc09 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java @@ -107,16 +107,16 @@ public class MapStepDaoTests extends TestCase { } public void testSaveRestartData() throws Exception { - assertEquals(null, dao.getRestartData(step.getId())); + assertEquals(null, dao.getStreamContext(step.getId())); step.setStatus(BatchStatus.COMPLETED); Properties data = new Properties(); data.setProperty("restart.key1", "restartData"); StreamContext streamContext = new GenericStreamContext(data); - step.setRestartData(streamContext); + step.setStreamContext(streamContext); dao.update(step); StepInstance tempStep = dao.findStep(job, step.getName()); assertEquals(tempStep, step); - assertEquals(tempStep.getRestartData().getProperties().toString(), + assertEquals(tempStep.getStreamContext().getProperties().toString(), streamContext.getProperties().toString()); } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java index c8e6eb0e0..80afe4825 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java @@ -24,6 +24,10 @@ import java.util.Properties; import junit.framework.TestCase; import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.stream.GenericStreamContext; +import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.batch.statistics.StatisticsService; import org.springframework.batch.support.PropertiesConverter; @@ -131,21 +135,21 @@ public class SimpleStepContextTests extends TestCase { assertTrue(list.contains("bar")); assertTrue(list.contains("spam")); } - + public void testStatisticsWithNullService() throws Exception { assertEquals(0, context.getStatistics().size()); } public void testStatisticsWithNotNullService() throws Exception { Map map = new HashMap(); - context = new SimpleStepContext(null, null, new StubStatisticsService(map)); + context = new SimpleStepContext(null, null, new StubStatisticsService(map), new StubStreamManager(map)); assertEquals(1, context.getStatistics().size()); assertEquals("bar", context.getStatistics().getProperty("foo")); } public void testStatisticsServiceRegistration() throws Exception { Map map = new HashMap(); - context = new SimpleStepContext(null, null, new StubStatisticsService(map)); + context = new SimpleStepContext(null, null, new StubStatisticsService(map), new StubStreamManager(map)); StubStatisticsProvider provider = new StubStatisticsProvider(); context.setAttribute("foo", provider); assertEquals(1, map.size()); @@ -155,7 +159,7 @@ public class SimpleStepContextTests extends TestCase { /** * @author Dave Syer - * + * */ private class StubStatisticsService implements StatisticsService { private final Map map; @@ -173,6 +177,35 @@ public class SimpleStepContextTests extends TestCase { } } + /** + * @author Dave Syer + * + */ + private class StubStreamManager implements StreamManager { + private final Map map; + + private StubStreamManager(Map map) { + this.map = map; + } + + public void close(Object key) { + } + + public StreamContext getStreamContext(Object key) { + return new GenericStreamContext(PropertiesConverter.stringToProperties("foo=bar")); + } + + public void open(Object key) { + } + + public void register(Object key, ItemStream stream) { + map.put(key, stream); + } + + public void restoreFrom(Object key, StreamContext data) { + } + } + /** * @author Dave Syer * diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepConfigurationTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepConfigurationTests.java index 576089530..a896f3921 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepConfigurationTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepConfigurationTests.java @@ -99,12 +99,12 @@ public class SimpleStepConfigurationTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.execution.step.simple.AbstractStep#isSaveRestartData()}. + * {@link org.springframework.batch.execution.step.simple.AbstractStep#isSaveStreamContext()}. */ public void testIsSaveRestartData() { - assertEquals(false, configuration.isSaveRestartData()); - configuration.setSaveRestartData(true); - assertEquals(true, configuration.isSaveRestartData()); + assertEquals(false, configuration.isSaveStreamContext()); + configuration.setSaveStreamContext(true); + assertEquals(true, configuration.isSaveStreamContext()); } } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java index 66e7b4ee2..e555b9e08 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java @@ -25,10 +25,10 @@ import java.util.Properties; import junit.framework.TestCase; -import org.springframework.batch.core.domain.JobSupport; import org.springframework.batch.core.domain.JobExecution; import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobParameters; +import org.springframework.batch.core.domain.JobSupport; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; @@ -43,7 +43,9 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamException; import org.springframework.batch.item.reader.ListItemReader; +import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatContext; @@ -72,7 +74,7 @@ public class SimpleStepExecutorTests extends TestCase { private AbstractStep stepConfiguration; private RepeatTemplate template; - + private JobInstance jobInstance; private ItemReader getReader(String[] args) { @@ -111,7 +113,7 @@ public class SimpleStepExecutorTests extends TestCase { template = new RepeatTemplate(); template.setCompletionPolicy(new SimpleCompletionPolicy(1)); stepExecutor.setChunkOperations(template); - + jobInstance = new JobInstance(new Long(0), new JobParameters()); jobInstance.setJob(new JobSupport("FOO")); } @@ -120,8 +122,7 @@ public class SimpleStepExecutorTests extends TestCase { StepInstance step = new StepInstance(new Long(9)); JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, - jobExecutionContext); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); stepExecutor.execute(stepExecution); assertEquals(1, processed.size()); @@ -158,8 +159,7 @@ public class SimpleStepExecutorTests extends TestCase { final StepInstance step = new StepInstance(new Long(1)); final JobExecution jobExecution = new JobExecution(jobInstance); - final StepExecution stepExecution = new StepExecution(step, - jobExecution); + final StepExecution stepExecution = new StepExecution(step, jobExecution); stepConfiguration.setTasklet(new Tasklet() { public ExitStatus execute() throws Exception { @@ -186,8 +186,7 @@ public class SimpleStepExecutorTests extends TestCase { final StepInstance step = new StepInstance(new Long(1)); final JobExecution jobExecution = new JobExecution(jobInstance); jobExecution.setId(new Long(1)); - final StepExecution stepExecution = new StepExecution(step, - jobExecution); + final StepExecution stepExecution = new StepExecution(step, jobExecution); template.setInterceptor(new RepeatInterceptorAdapter() { public void open(RepeatContext context) { @@ -210,8 +209,7 @@ public class SimpleStepExecutorTests extends TestCase { StepInstance step = new StepInstance(new Long(1)); JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, - jobExecutionContext); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); stepExecutor.execute(stepExecution); assertEquals(1, processed.size()); @@ -237,8 +235,7 @@ public class SimpleStepExecutorTests extends TestCase { StepInstance step = new StepInstance(new Long(1)); stepConfiguration.setTasklet(tasklet); JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, - jobExecutionContext); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); try { stepExecutor.execute(stepExecution); @@ -269,8 +266,7 @@ public class SimpleStepExecutorTests extends TestCase { StepInstance step = new StepInstance(new Long(1)); stepConfiguration.setTasklet(tasklet); JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, - jobExecutionContext); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); try { stepExecutor.execute(stepExecution); @@ -285,21 +281,15 @@ public class SimpleStepExecutorTests extends TestCase { * make sure a job that has never been executed before, but does have * saveRestartData = true, doesn't have restoreFrom called on it. */ - public void testNonRestartedJob() { + public void testNonRestartedJob() throws Exception { StepInstance step = new StepInstance(new Long(1)); MockRestartableTasklet tasklet = new MockRestartableTasklet(); stepExecutor.setTasklet(tasklet); - stepConfiguration.setSaveRestartData(true); + stepConfiguration.setSaveStreamContext(true); JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, - jobExecutionContext); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - try { - stepExecutor.execute(stepExecution); - } - catch (Throwable t) { - fail(); - } + stepExecutor.execute(stepExecution); assertFalse(tasklet.isRestoreFromCalled()); assertTrue(tasklet.isGetRestartDataCalled()); @@ -309,24 +299,21 @@ public class SimpleStepExecutorTests extends TestCase { * make sure a job that has been executed before, and is therefore being * restarted, is restored. */ - public void testRestartedJob() { + public void testRestartedJob() throws Exception { StepInstance step = new StepInstance(new Long(1)); step.setStepExecutionCount(1); MockRestartableTasklet tasklet = new MockRestartableTasklet(); stepExecutor.setTasklet(tasklet); - stepConfiguration.setSaveRestartData(true); + stepConfiguration.setSaveStreamContext(true); JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, - jobExecutionContext); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + stepExecution.getStep().setStreamContext( + new GenericStreamContext(PropertiesConverter.stringToProperties("foo=bar"))); - try { - stepExecutor.execute(stepExecution); - } - catch (Throwable t) { - fail(); - } + stepExecutor.execute(stepExecution); assertTrue(tasklet.isRestoreFromCalled()); + assertTrue(tasklet.isRestoreFromCalledWithSomeContext()); assertTrue(tasklet.isGetRestartDataCalled()); } @@ -339,10 +326,9 @@ public class SimpleStepExecutorTests extends TestCase { step.setStepExecutionCount(1); MockRestartableTasklet tasklet = new MockRestartableTasklet(); stepConfiguration.setTasklet(tasklet); - stepConfiguration.setSaveRestartData(false); + stepConfiguration.setSaveStreamContext(false); JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, - jobExecutionContext); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); try { stepExecutor.execute(stepExecution); @@ -359,7 +345,7 @@ public class SimpleStepExecutorTests extends TestCase { * Even though the job is restarted, and saveRestartData is true, nothing * will be restored because the Tasklet does not implement Restartable. */ - public void testRestartJobOnNonRestartableTasklet() { + public void testRestartJobOnNonRestartableTasklet() throws Exception { StepInstance step = new StepInstance(new Long(1)); step.setStepExecutionCount(1); stepConfiguration.setTasklet(new Tasklet() { @@ -367,16 +353,11 @@ public class SimpleStepExecutorTests extends TestCase { return ExitStatus.FINISHED; } }); - stepConfiguration.setSaveRestartData(true); + stepConfiguration.setSaveStreamContext(true); JobExecution jobExecution = new JobExecution(jobInstance); StepExecution stepExecution = new StepExecution(step, jobExecution); - try { - stepExecutor.execute(stepExecution); - } - catch (Throwable t) { - fail(); - } + stepExecutor.execute(stepExecution); } public void testApplyConfigurationWithExceptionHandler() throws Exception { @@ -426,10 +407,10 @@ public class SimpleStepExecutorTests extends TestCase { return ExitStatus.FINISHED; } }); - stepConfiguration.setSaveRestartData(true); + stepConfiguration.setSaveStreamContext(true); JobExecution jobExecution = new JobExecution(jobInstance); StepExecution stepExecution = new StepExecution(step, jobExecution); - + assertEquals(null, stepExecution.getStatistics().getProperty("foo")); final Map map = new HashMap(); @@ -437,6 +418,7 @@ public class SimpleStepExecutorTests extends TestCase { public Properties getStatistics(Object key) { return PropertiesConverter.stringToProperties("foo=bar"); } + public void register(Object key, StatisticsProvider provider) { map.put(key, provider); } @@ -448,8 +430,9 @@ public class SimpleStepExecutorTests extends TestCase { catch (Throwable t) { fail(); } - - // At least once in that process the statistics service was asked for statistics... + + // At least once in that process the statistics service was asked for + // statistics... assertEquals("bar", stepExecution.getStatistics().getProperty("foo")); // ...but nothing was registered because nothing with step scoped. assertEquals(0, map.size()); @@ -461,17 +444,25 @@ public class SimpleStepExecutorTests extends TestCase { private boolean restoreFromCalled = false; + private boolean restoreFromCalledWithSomeContext = false; + public ExitStatus execute() throws Exception { + StepSynchronizationManager.getContext().setAttribute("TASKLET_TEST", this); return ExitStatus.FINISHED; } - public StreamContext getRestartData() { + public boolean isRestoreFromCalledWithSomeContext() { + return restoreFromCalledWithSomeContext; + } + + public StreamContext getStreamContext() { getRestartDataCalled = true; - return null; + return new GenericStreamContext(PropertiesConverter.stringToProperties("spam=bucket")); } public void restoreFrom(StreamContext data) { restoreFromCalled = true; + restoreFromCalledWithSomeContext = data.getProperties().size() > 0; } public boolean isGetRestartDataCalled() { @@ -481,15 +472,13 @@ public class SimpleStepExecutorTests extends TestCase { public boolean isRestoreFromCalled() { return restoreFromCalled; } - - public void open() throws Exception { - throw new UnsupportedOperationException("Not implemented."); + + public void open() throws StreamException { } - - public void close() throws Exception { - throw new UnsupportedOperationException("Not implemented."); + + public void close() throws StreamException { } - + } /* diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java index 6304ccb34..1ac0bbeb1 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java @@ -28,6 +28,7 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.KeyedItemReader; +import org.springframework.batch.item.StreamException; import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.context.RepeatContextSupport; @@ -60,7 +61,7 @@ public class ItemOrientedTaskletTests extends TestCase { return null; } - public void close() throws Exception { + public void close() throws StreamException { } }; @@ -182,7 +183,7 @@ public class ItemOrientedTaskletTests extends TestCase { return "bar"; } - public void close() throws Exception { + public void close() throws StreamException { } }); @@ -287,7 +288,7 @@ public class ItemOrientedTaskletTests extends TestCase { public Properties getStatistics() { return PropertiesConverter.stringToProperties("foo=bar"); } - public void close() throws Exception { + public void close() throws StreamException { } } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTaskletTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTaskletTests.java deleted file mode 100644 index 0dcd6c904..000000000 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTaskletTests.java +++ /dev/null @@ -1,160 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.batch.execution.tasklet; - -import java.util.Properties; - -import junit.framework.TestCase; - -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.StreamContext; -import org.springframework.batch.item.stream.GenericStreamContext; -import org.springframework.batch.support.PropertiesConverter; - -/** - * @author Peter Zozom - */ -public class RestartableItemOrientedTaskletTests extends TestCase { - - private static class MockProvider implements ItemReader, ItemStream { - - StreamContext data = new StreamContext() { - - public Properties getProperties() { - return PropertiesConverter.stringToProperties("a=b"); - } - - }; - - public Object read() { - return null; - } - - public StreamContext getRestartData() { - return data; - } - - public void restoreFrom(StreamContext data) { - // restart data should be same as returned by getRestartData - assertEquals(this.data.getProperties(), data.getProperties()); - } - - public boolean recover(Object data, Throwable cause) { - return false; - } - - public void open() throws Exception { - throw new UnsupportedOperationException("Not implemented."); - } - - public void close() throws Exception { - throw new UnsupportedOperationException("Not implemented."); - } - - } - - private static class MockWriter implements ItemWriter, ItemStream { - - StreamContext data = new StreamContext() { - public Properties getProperties() { - return PropertiesConverter.stringToProperties("x=y"); - } - }; - - public void write(Object data) { - } - - public StreamContext getRestartData() { - return data; - } - - public void restoreFrom(StreamContext data) { - // restart data should be same as returned by getRestartData - assertEquals(this.data.getProperties(), data.getProperties()); - } - - public void open() throws Exception { - throw new UnsupportedOperationException("Not implemented."); - } - - public void close() throws Exception { - throw new UnsupportedOperationException("Not implemented."); - } - - } - - private ItemReader itemProvider; - - private ItemWriter itemWriter; - - private RestartableItemOrientedTasklet module; - - public void testRestart() { - - // create data provider and data processor - itemProvider = new MockProvider(); - itemWriter = new MockWriter(); - - // create and set up module - module = new RestartableItemOrientedTasklet(); - module.setItemReader(itemProvider); - module.setItemWriter(itemWriter); - - // get restart data - StreamContext data = module.getRestartData(); - assertNotNull(data); - // restore from restart data (see asserts in mock classes) - module.restoreFrom(data); - } - - public void testRestartFromGenericData() { - - // create data provider and data processor - itemProvider = new MockProvider(); - itemWriter = new MockWriter(); - - // create and set up module - module = new RestartableItemOrientedTasklet(); - module.setItemReader(itemProvider); - module.setItemWriter(itemWriter); - - // get restart data - StreamContext data = module.getRestartData(); - assertNotNull(data); - data = new GenericStreamContext(data.getProperties()); - // restore from restart data (see asserts in mock classes) - module.restoreFrom(data); - } - - public void testRestartFromNotRestartable() { - - // create and set up module - module = new RestartableItemOrientedTasklet(); - module.setItemReader(null); - module.setItemWriter(null); - - // get restart data - StreamContext data = module.getRestartData(); - assertNotNull(data); - // restore from restart data (see asserts in mock classes) - module.restoreFrom(data); - //System.err.println(data.getProperties()); - } - -} diff --git a/spring-batch-execution/src/test/resources/simple-container-definition.xml b/spring-batch-execution/src/test/resources/simple-container-definition.xml index a82fbcf89..ce0054edc 100644 --- a/spring-batch-execution/src/test/resources/simple-container-definition.xml +++ b/spring-batch-execution/src/test/resources/simple-container-definition.xml @@ -27,7 +27,7 @@ class="org.springframework.batch.execution.step.simple.SimpleStep" abstract="true"> - + diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java index ba376bc79..ec0fd1a61 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java @@ -186,7 +186,7 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite /** * @return the current row number wrapped as RestartData */ - public StreamContext getRestartData() { + public StreamContext getStreamContext() { Properties props = new Properties(); props.setProperty(RESTART_DATA_ROW_NUMBER_KEY, ""+currentProcessedRow); String skipped = skippedRows.toString(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java index 23d089520..b94272c8a 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java @@ -405,7 +405,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * * @see org.springframework.batch.restart.Restartable#getRestartData() */ - public StreamContext getRestartData() { + public StreamContext getStreamContext() { String skipped = skippedRows.toString(); Properties statistics = getStatistics(); statistics.setProperty(SKIPPED_ROWS, skipped.substring(1,skipped.length()-1)); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java index 8ac6ac8a6..72ed88c66 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java @@ -191,8 +191,8 @@ public class DrivingQueryItemReader extends AbstractTransactionalIoSource } } - public StreamContext getRestartData() { - return keyGenerator.getKeyAsRestartData(getCurrentKey()); + public StreamContext getStreamContext() { + return keyGenerator.getKeyAsStreamContext(getCurrentKey()); } public void afterPropertiesSet() throws Exception { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java index 2b3c9a8e5..3c35c66bb 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/KeyGenerator.java @@ -34,5 +34,5 @@ public interface KeyGenerator { * @throws IllegalArgumentException if key is null. * @throws IllegalArgumentException if key is an incompatible type. */ - StreamContext getKeyAsRestartData(Object key); + StreamContext getKeyAsStreamContext(Object key); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapStreamContextRowMapper.java similarity index 92% rename from spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapper.java rename to spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapStreamContextRowMapper.java index 2b5b1ecaf..428e480e3 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapper.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapStreamContextRowMapper.java @@ -23,8 +23,8 @@ import org.springframework.util.Assert; import org.springframework.util.ClassUtils; /** - *

Extension of the ColumnMapRowMapper that converts a column map to RestartData and allows - * RestartData to be converted back as a PreparedStatementSetter. This is useful in a restart + *

Extension of the ColumnMapRowMapper that converts a column map to {@link StreamContext} and allows + * {@link StreamContext} to be converted back as a PreparedStatementSetter. This is useful in a restart * scenario, as it allows for the standard functionality of the ColumnMapRowMapper to be used to * create a map representing the columns returned by a query. It should be noted that this column ordering * is preserved in the map using a link list version of Map. @@ -34,9 +34,9 @@ import org.springframework.util.ClassUtils; * @author Dave Syer * @see RestartDataRowMapper */ -public class ColumnMapRestartDataRowMapper extends ColumnMapRowMapper implements RestartDataRowMapper{ +public class ColumnMapStreamContextRowMapper extends ColumnMapRowMapper implements RestartDataRowMapper{ - static final String KEY = ClassUtils.getQualifiedName(ColumnMapRestartDataRowMapper.class) + ".KEY."; + static final String KEY = ClassUtils.getQualifiedName(ColumnMapStreamContextRowMapper.class) + ".KEY."; public PreparedStatementSetter createSetter(StreamContext streamContext) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java index 1afc2a660..48cbcaffd 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/IbatisKeyGenerator.java @@ -44,7 +44,7 @@ public class IbatisKeyGenerator implements KeyGenerator { * * @see org.springframework.batch.restart.Restartable#getRestartData() */ - public StreamContext getKeyAsRestartData(Object key) { + public StreamContext getKeyAsStreamContext(Object key) { Properties props = new Properties(); props.setProperty(RESTART_KEY, key.toString()); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java index 468abc733..c720b4cb3 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java @@ -43,7 +43,7 @@ public class MultipleColumnJdbcKeyGenerator implements private JdbcTemplate jdbcTemplate; - private RestartDataRowMapper keyMapper = new ColumnMapRestartDataRowMapper(); + private RestartDataRowMapper keyMapper = new ColumnMapStreamContextRowMapper(); private String sql; @@ -96,7 +96,7 @@ public class MultipleColumnJdbcKeyGenerator implements /* (non-Javadoc) * @see org.springframework.batch.restart.Restartable#getRestartData() */ - public StreamContext getKeyAsRestartData(Object key) { + public StreamContext getKeyAsStreamContext(Object key) { Assert.state(keyMapper != null, "RestartDataConverter must not be null."); return keyMapper.createRestartData(key); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java index fef500d17..dd25d15ee 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGenerator.java @@ -94,7 +94,7 @@ public class SingleColumnJdbcKeyGenerator implements KeyGenerator { * @see KeyGenerator#getKeyAsRestartData() * @throws IllegalArgumentException if key is null. */ - public StreamContext getKeyAsRestartData(Object key) { + public StreamContext getKeyAsStreamContext(Object key) { Assert.notNull(key, "The key must not be null."); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java index 09f47c036..c9267c50b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java @@ -104,7 +104,7 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen * current Line Count which can be used to re initialise the batch job in * case of restart. */ - public StreamContext getRestartData() { + public StreamContext getStreamContext() { return new GenericStreamContext(getStatistics()); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java index 4a708b0fc..17f46dfe4 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java @@ -257,9 +257,9 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements } /** - * @see ItemStream#getRestartData() + * @see ItemStream#getStreamContext() */ - public StreamContext getRestartData() { + public StreamContext getStreamContext() { final OutputState os = getOutputState(); streamContext.getProperties().setProperty(RESTART_DATA_NAME, String.valueOf(os.position())); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java index e2a185dec..9f0ec041e 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java @@ -31,6 +31,7 @@ import org.springframework.batch.io.file.transform.AbstractLineTokenizer; import org.springframework.batch.io.file.transform.DelimitedLineTokenizer; import org.springframework.batch.io.file.transform.LineTokenizer; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.StreamException; import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; @@ -175,7 +176,7 @@ public class SimpleFlatFileItemReader extends AbstractItemReader implements Item * * @see ResourceLifecycle */ - public void close() throws Exception { + public void close() throws StreamException { try { if (reader != null) { log.debug("Closing flat file for reading: "+resource); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java index e20ef4048..1710cecd7 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java @@ -191,9 +191,9 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade /** * @return wrapped count of records read so far. - * @see ItemStream#getRestartData() + * @see ItemStream#getStreamContext() */ - public StreamContext getRestartData() { + public StreamContext getStreamContext() { Properties restartData = new Properties(); restartData.setProperty(RESTART_DATA_NAME, String.valueOf(currentRecordCount)); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java index 27768f639..b19abb083 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java @@ -379,9 +379,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Get the restart data. * @return the restart data - * @see org.springframework.batch.item.ItemStream#getRestartData() + * @see org.springframework.batch.item.ItemStream#getStreamContext() */ - public StreamContext getRestartData() { + public StreamContext getStreamContext() { Properties properties = new Properties(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java index a36561d88..f11b57fcf 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java @@ -58,5 +58,5 @@ public interface ItemReader { * TODO: this is only used in sandbox? * */ - void close() throws Exception; + void close() throws StreamException; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java index 30a79e9c9..fd9cb0fb5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java @@ -26,7 +26,7 @@ package org.springframework.batch.item; * The state that is stored is represented as {@link StreamContext} which * enforces a requirement that any restart data can be represented by a * Properties object. In general, the contract is that {@link StreamContext} - * that is returned via the {@link #getRestartData()} method will be given back + * that is returned via the {@link #getStreamContext()} method will be given back * to the {@link #restoreFrom(StreamContext)} method, exactly as it was * provided. *

@@ -34,32 +34,24 @@ package org.springframework.batch.item; * @author Lucas Ward * */ -public interface ItemStream { - - /** - * Get {@link StreamContext} representing this object's current state. - * Should not return null even if there is no state. - * - * @return {@link StreamContext} representing current state. - */ - StreamContext getRestartData(); +public interface ItemStream extends StreamContextProvider { /** * Restart state given the provided {@link StreamContext}. * - * @param data + * @param context */ - void restoreFrom(StreamContext data); + void restoreFrom(StreamContext context); /** * If any resources are needed for the stream to operate they need to be * initialised here. */ - void open() throws Exception; + void open() throws StreamException; /** * If any resources are needed for the stream to operate they need to be * destroyed here. */ - void close() throws Exception; + void close() throws StreamException; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContextProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContextProvider.java new file mode 100644 index 000000000..271f830b6 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContextProvider.java @@ -0,0 +1,32 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.item; + +/** + * @author Dave Syer + * + */ +public interface StreamContextProvider { + + /** + * Get {@link StreamContext} representing this object's current state. + * Should not return null even if there is no state. + * + * @return {@link StreamContext} representing current state. + */ + StreamContext getStreamContext(); + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java new file mode 100644 index 000000000..8396a1f84 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java @@ -0,0 +1,26 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.item; + +import org.springframework.batch.io.exception.BatchCriticalException; + +/** + * @author Dave Syer + * + */ +public class StreamException extends BatchCriticalException { + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java index c9f33f7ca..13c9818cb 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java @@ -17,6 +17,7 @@ package org.springframework.batch.item.reader; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.StreamException; /** * Base class for {@link ItemReader} implementations. @@ -29,7 +30,7 @@ public abstract class AbstractItemReader implements ItemReader { * Do nothing. * @see org.springframework.batch.item.ItemReader#close() */ - public void close() throws Exception { + public void close() throws StreamException { } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java index 2a75d64f3..ca7f5b3da 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java @@ -20,6 +20,7 @@ import org.springframework.batch.io.Skippable; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamException; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; @@ -47,15 +48,14 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl } /** - * @see ItemStream#getRestartData() + * @see ItemStream#getStreamContext() * @throws IllegalStateException if the parent template is not itself * {@link ItemStream}. */ - public StreamContext getRestartData() { - if (!(inputSource instanceof ItemStream)) { - throw new IllegalStateException("Input Template is not Restartable"); - } - return ((ItemStream) inputSource).getRestartData(); + public StreamContext getStreamContext() { + // TODO: this is not necessary... + Assert.state(inputSource instanceof ItemStream, "Input source is not ItemStream"); + return ((ItemStream) inputSource).getStreamContext(); } /** @@ -64,9 +64,7 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl * {@link ItemStream}. */ public void restoreFrom(StreamContext data) { - if (!(inputSource instanceof ItemStream)) { - throw new IllegalStateException("Input Template is not Restartable"); - } + Assert.state(inputSource instanceof ItemStream, "Input source is not ItemStream"); ((ItemStream) inputSource).restoreFrom(data); } @@ -91,7 +89,7 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl /* (non-Javadoc) * @see org.springframework.batch.item.ItemStream#open() */ - public void open() throws Exception { + public void open() throws StreamException { if (inputSource instanceof ItemStream) { ((ItemStream) inputSource).open(); } @@ -100,7 +98,7 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl /* (non-Javadoc) * @see org.springframework.batch.item.ItemStream#open() */ - public void close() throws Exception { + public void close() throws StreamException { if (inputSource instanceof ItemStream) { ((ItemStream) inputSource).close(); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java index f9ffcebcf..42ab28106 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java @@ -17,6 +17,7 @@ package org.springframework.batch.item.reader; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.StreamException; import org.springframework.batch.support.AbstractMethodInvokingDelegator; /** @@ -38,7 +39,7 @@ public class ItemReaderAdapter extends AbstractMethodInvokingDelegator implement * * @see org.springframework.batch.item.ItemReader#close() */ - public void close() throws Exception { + public void close() throws StreamException { } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/CompositeItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/CompositeItemStream.java deleted file mode 100644 index 17ae317bc..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/CompositeItemStream.java +++ /dev/null @@ -1,158 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.batch.item.stream; - -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.Properties; - -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.StreamContext; - -/** - * @author Dave Syer - * - */ -public class CompositeItemStream implements ItemStream { - - private static final String SEPARATOR = "#"; - - private List delegates; - - public void setDelegates(List delegates) { - this.delegates = delegates; - } - - /** - * Compound restart data of all injected (Restartable) ItemProcessors, - * property keys are prefixed with list index of the ItemProcessor. - */ - public StreamContext getRestartData() { - Properties props = createCompoundProperties(new PropertiesExtractor() { - public Properties extractProperties(Object o) { - if (o instanceof ItemStream) { - return ((ItemStream) o).getRestartData().getProperties(); - } - else { - return null; - } - } - }); - return new GenericStreamContext(props); - } - - /** - * @param data contains values of restart data, property keys are expected - * to be prefixed with list index of the ItemProcessor. - */ - public void restoreFrom(StreamContext data) { - if (data == null || data.getProperties() == null) { - // do nothing - return; - } - - List restartDataList = parseProperties(data.getProperties()); - - // iterators would make the loop below less readable - for (int i = 0; i < delegates.size(); i++) { - if (delegates.get(i) instanceof ItemStream) { - ((ItemStream) delegates.get(i)).restoreFrom((StreamContext) restartDataList.get(i)); - } - } - - } - - /** - * Parses compound properties into a list of RestartData. - */ - private List parseProperties(Properties props) { - List restartDataList = new ArrayList(delegates.size()); - for (int i = 0; i < delegates.size(); i++) { - restartDataList.add(new GenericStreamContext(new Properties())); - } - - for (Iterator iterator = props.entrySet().iterator(); iterator.hasNext();) { - Map.Entry entry = (Map.Entry) iterator.next(); - String key = (String) entry.getKey(); - String value = (String) entry.getValue(); - int separatorIndex = key.indexOf(SEPARATOR); - int i = Integer.valueOf(key.substring(0, separatorIndex)).intValue(); - ((StreamContext) restartDataList.get(i)).getProperties().setProperty(key.substring(separatorIndex + 1), - value); - } - return restartDataList; - } - - /** - * @param extractor used to extract Properties from {@link ItemReader}s - * @return compound Properties containing all the Properties from injected - * delegates with property keys prefixed by list index. - */ - private Properties createCompoundProperties(PropertiesExtractor extractor) { - Properties stats = new Properties(); - int index = 0; - for (Iterator iterator = delegates.listIterator(); iterator.hasNext();) { - Properties writerStats = extractor.extractProperties(iterator.next()); - if (writerStats != null) { - for (Iterator iterator2 = writerStats.entrySet().iterator(); iterator2.hasNext();) { - Map.Entry entry = (Map.Entry) iterator2.next(); - stats.setProperty("" + index + SEPARATOR + entry.getKey(), (String) entry.getValue()); - } - } - index++; - } - return stats; - } - - /** - * Extracts information from given object in the form of {@link Properties}. - * If the information is not available (e.g. unexpected object class) return - * null. - */ - private interface PropertiesExtractor { - Properties extractProperties(Object o); - } - - public void close() throws Exception { - for (Iterator iterator = delegates.listIterator(); iterator.hasNext();) { - Object delegate = iterator.next(); - if (delegate instanceof ItemStream) { - ((ItemStream) delegate).close(); - } - } - } - - public void open() throws Exception { - for (Iterator iterator = delegates.listIterator(); iterator.hasNext();) { - Object delegate = iterator.next(); - if (delegate instanceof ItemStream) { - ((ItemStream) delegate).open(); - } - } - } - - /** - * Public getter for the list of delegates. - * @return the delegates - */ - public List getDelegates() { - return delegates; - } - -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java new file mode 100644 index 000000000..6297e939d --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java @@ -0,0 +1,150 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.item.stream; + +import java.util.Collection; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Properties; +import java.util.Set; + +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamException; +import org.springframework.batch.statistics.StatisticsProvider; +import org.springframework.batch.support.PropertiesConverter; + +/** + * Simple {@link StreamManager} that makes no attempt to aggregate or resolve + * conflicts between key names. All the contributions registered are simply + * polled and added "as is" to the aggregate properties. + * + * @author Dave Syer + * + */ +public class SimpleStreamManager implements StreamManager { + + private Map registry = new HashMap(); + + /** + * Simple aggregate statistics provider for the contributions registered + * under the given key. + * + * @see org.springframework.batch.statistics.StatisticsService#getStatistics(java.lang.Object) + */ + public StreamContext getStreamContext(Object key) { + Set set = new LinkedHashSet(); + synchronized (registry) { + Collection collection = (Collection) registry.get(key); + if (collection != null) { + set = new LinkedHashSet(collection); + } + } + return new GenericStreamContext(aggregate(set)); + } + + /** + * @param list a list of {@link StatisticsProvider}s + * @return aggregated statistics + */ + private Properties aggregate(Collection list) { + Properties result = new Properties(); + for (Iterator iterator = list.iterator(); iterator.hasNext();) { + ItemStream provider = (ItemStream) iterator.next(); + Properties properties = provider.getStreamContext().getProperties(); + if (properties != null) { + result.putAll(properties); + } + } + return result; + } + + /** + * Register a {@link ItemStream} as one of the interesting providers under + * the provided key. + * + * @see org.springframework.batch.statistics.StreamManager#register(java.lang.Object, + * org.springframework.batch.statistics.StatisticsProvider) + */ + public void register(Object key, ItemStream provider) { + synchronized (registry) { + Set set = (Set) registry.get(key); + if (set == null) { + set = new LinkedHashSet(); + registry.put(key, set); + } + set.add(provider); + } + } + + /** + * Broadcast the call to close from this {@link StreamContext}. + * @throws Exception + * + * @see StreamManager#restoreFrom(Object, StreamContext) + */ + public void close(Object key) throws StreamException { + Set set = new LinkedHashSet(); + synchronized (registry) { + Collection collection = (Collection) registry.get(key); + if (collection != null) { + set.addAll(collection); + } + } + for (Iterator iterator = set.iterator(); iterator.hasNext();) { + ItemStream stream = (ItemStream) iterator.next(); + stream.close(); + } + } + + // TODO: integrate this + private class SimpleStreamManagerStreamContext implements StreamContext { + + private static final String READER_KEY = "DATA_PROVIDER"; + + private static final String WRITER_KEY = "DATA_PROCESSOR"; + + private StreamContext readerData; + + private StreamContext writerData; + + public SimpleStreamManagerStreamContext(StreamContext providerData, StreamContext writerData) { + this.readerData = providerData; + this.writerData = writerData; + } + + public SimpleStreamManagerStreamContext(Properties data) { + readerData = new GenericStreamContext(PropertiesConverter + .stringToProperties(data.getProperty(READER_KEY))); + writerData = new GenericStreamContext(PropertiesConverter.stringToProperties(data + .getProperty(WRITER_KEY))); + } + + public Properties getProperties() { + Properties props = new Properties(); + if (readerData != null) { + props.setProperty(READER_KEY, PropertiesConverter.propertiesToString(readerData.getProperties())); + } + if (writerData != null) { + props.setProperty(WRITER_KEY, PropertiesConverter.propertiesToString(writerData.getProperties())); + } + return props; + } + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java new file mode 100644 index 000000000..eea390488 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java @@ -0,0 +1,62 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.item.stream; + +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamException; + +/** + * Generalised stream management broadcast strategy. Clients register + * {@link ItemStream} instances under a well-known key, and then when they ask + * for {@link ItemStream} operations by that key, we call each one in turn that + * is registered under the key. + * + * @author Dave Syer + * + */ +public interface StreamManager { + + /** + * Register the {@link ItemStream} instance as one of possibly several that + * are associated with the given key. + * + * @param key the key under which to add the provider + * @param stream an {@link ItemStream} + */ + void register(Object key, ItemStream stream); + + /** + * Extract and aggregate the {@link StreamContext} from all streams under + * this key. + * + * @param key the key under which {@link ItemStream} instances might have + * been registered. + * @return {@link StreamContext} aggregating the contexts of all providers + * registered under this key, or empty otherwise. + */ + StreamContext getStreamContext(Object key); + + /** + * If any resources are needed for the stream to operate they need to be + * destroyed here. + * + * @param key the key under which {@link ItemStream} instances might have + * been registered. + */ + void close(Object key) throws StreamException; + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/CompositeItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/CompositeItemWriter.java index 7e8e8f942..5cd849607 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/CompositeItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/CompositeItemWriter.java @@ -1,22 +1,28 @@ package org.springframework.batch.item.writer; import java.util.Iterator; +import java.util.List; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.stream.CompositeItemStream; /** * Runs a collection of ItemProcessors in fixed-order sequence. * * @author Robert Kasanicky */ -public class CompositeItemWriter extends CompositeItemStream implements ItemWriter { +public class CompositeItemWriter extends AbstractItemWriter implements ItemWriter { + + private List delegates; + + public void setDelegates(List delegates) { + this.delegates = delegates; + } /** * Calls injected ItemProcessors in order. */ public void write(Object data) throws Exception { - for (Iterator iterator = getDelegates().listIterator(); iterator.hasNext();) { + for (Iterator iterator = delegates.listIterator(); iterator.hasNext();) { ((ItemWriter) iterator.next()).write(data); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java index bbd07e810..2a128f70d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java @@ -50,14 +50,14 @@ public class DelegatingItemWriter implements ItemWriter, Skippable, Initializing } /** - * @see ItemStream#getRestartData() + * @see ItemStream#getStreamContext() */ - public StreamContext getRestartData() { + public StreamContext getStreamContext() { Assert.state(writer != null, "Source must not be null."); if (writer instanceof ItemStream) { - return ((ItemStream) writer).getRestartData(); + return ((ItemStream) writer).getStreamContext(); } else { return new GenericStreamContext(new Properties()); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/DrivingQueryItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/DrivingQueryItemReaderTests.java index 214b4a311..f72f97668 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/DrivingQueryItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/DrivingQueryItemReaderTests.java @@ -76,7 +76,7 @@ public class DrivingQueryItemReaderTests extends TestCase { Foo foo2 = (Foo) source.read(); assertEquals(2, foo2.getValue()); - StreamContext streamContext = getAsRestartable(source).getRestartData(); + StreamContext streamContext = getAsRestartable(source).getStreamContext(); // create new input source source = createItemReader(); @@ -98,7 +98,7 @@ public class DrivingQueryItemReaderTests extends TestCase { Foo foo2 = (Foo) source.read(); assertEquals(2, foo2.getValue()); - StreamContext streamContext = getAsRestartable(source).getRestartData(); + StreamContext streamContext = getAsRestartable(source).getStreamContext(); // create new input source source = createItemReader(); @@ -198,7 +198,7 @@ public class DrivingQueryItemReaderTests extends TestCase { restartKeys.add(new Foo(5, "5", 5)); } - public StreamContext getKeyAsRestartData(Object key) { + public StreamContext getKeyAsStreamContext(Object key) { return streamContext; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java index 135f1cfd9..7f7a5da21 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/FooInputSource.java @@ -27,8 +27,8 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader } } - public StreamContext getRestartData() { - return inputSource.getRestartData(); + public StreamContext getStreamContext() { + return inputSource.getStreamContext(); } public void restoreFrom(StreamContext data) { @@ -46,6 +46,9 @@ class FooItemReader extends AbstractItemReader implements ItemStream, ItemReader public void afterPropertiesSet() throws Exception { } - public void open() throws Exception { + public void open() { + }; + + public void close() { }; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapperTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapperTests.java index e6c56e1c9..4fe509c0d 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapperTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapperTests.java @@ -22,9 +22,9 @@ import org.springframework.util.ClassUtils; */ public class ColumnMapRestartDataRowMapperTests extends TestCase { - private static final String KEY = ClassUtils.getQualifiedName(ColumnMapRestartDataRowMapper.class) + ".KEY."; + private static final String KEY = ClassUtils.getQualifiedName(ColumnMapStreamContextRowMapper.class) + ".KEY."; - ColumnMapRestartDataRowMapper mapper; + ColumnMapStreamContextRowMapper mapper; Map key; @@ -34,7 +34,7 @@ public class ColumnMapRestartDataRowMapperTests extends TestCase { protected void setUp() throws Exception { super.setUp(); - mapper = new ColumnMapRestartDataRowMapper(); + mapper = new ColumnMapStreamContextRowMapper(); key = CollectionFactory.createLinkedCaseInsensitiveMapIfPossible(2); key.put("1", new Integer(1)); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java index 2be86b0ac..64354d3e4 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java @@ -47,8 +47,8 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran public void testRestoreKeys(){ Properties props = new Properties(); - props.setProperty(ColumnMapRestartDataRowMapper.KEY + "0", "3"); - props.setProperty(ColumnMapRestartDataRowMapper.KEY + "1", "3"); + props.setProperty(ColumnMapStreamContextRowMapper.KEY + "0", "3"); + props.setProperty(ColumnMapStreamContextRowMapper.KEY + "1", "3"); StreamContext streamContext = new GenericStreamContext(props); List keys = keyStrategy.restoreKeys(streamContext); @@ -68,18 +68,18 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran key.put("ID", new Long(3)); key.put("VALUE", new Integer(3)); - StreamContext streamContext = keyStrategy.getKeyAsRestartData(key); + StreamContext streamContext = keyStrategy.getKeyAsStreamContext(key); Properties props = streamContext.getProperties(); assertEquals(2, props.size()); - assertEquals("3", props.get(ColumnMapRestartDataRowMapper.KEY + "0")); - assertEquals("3", props.get(ColumnMapRestartDataRowMapper.KEY + "1")); + assertEquals("3", props.get(ColumnMapStreamContextRowMapper.KEY + "0")); + assertEquals("3", props.get(ColumnMapStreamContextRowMapper.KEY + "1")); } - public void testGetNullKeyAsRestartData(){ + public void testGetNullKeyAsStreamContext(){ try{ - keyStrategy.getKeyAsRestartData(null); + keyStrategy.getKeyAsStreamContext(null); fail(); }catch(IllegalArgumentException ex){ //expected @@ -89,7 +89,7 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran public void testRestoreKeysFromNull(){ try{ - keyStrategy.getKeyAsRestartData(null); + keyStrategy.getKeyAsStreamContext(null); }catch(IllegalArgumentException ex){ //expected } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGeneratorIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGeneratorIntegrationTests.java index 8338eef9c..f54493d80 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGeneratorIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/SingleColumnJdbcKeyGeneratorIntegrationTests.java @@ -53,19 +53,19 @@ public class SingleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTransa assertEquals(new Long(5), keys.get(1)); } - public void testGetKeyAsRestartData(){ + public void testGetKeyAsStreamContext(){ - StreamContext streamContext = keyStrategy.getKeyAsRestartData(new Long(3)); + StreamContext streamContext = keyStrategy.getKeyAsStreamContext(new Long(3)); Properties props = streamContext.getProperties(); assertEquals(1, props.size()); assertEquals("3", props.get(SingleColumnJdbcKeyGenerator.RESTART_KEY)); } - public void testGetNullKeyAsRestartData(){ + public void testGetNullKeyAsStreamContext(){ try{ - keyStrategy.getKeyAsRestartData(null); + keyStrategy.getKeyAsStreamContext(null); fail(); }catch(IllegalArgumentException ex){ //expected @@ -75,7 +75,7 @@ public class SingleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTransa public void testRestoreKeysFromNull(){ try{ - keyStrategy.getKeyAsRestartData(null); + keyStrategy.getKeyAsStreamContext(null); }catch(IllegalArgumentException ex){ //expected } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java index 76ac290ac..a172f2af6 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java @@ -184,7 +184,7 @@ public class DefaultFlatFileItemReaderTests extends TestCase { inputSource.setResource(getInputResource(TEST_STRING)); inputSource.setFieldSetMapper(fieldSetMapper); // do not open the template... - inputSource.restoreFrom(inputSource.getRestartData()); + inputSource.restoreFrom(inputSource.getStreamContext()); assertEquals("[FlatFileInputTemplate-TestData]", inputSource.read().toString()); } @@ -204,7 +204,7 @@ public class DefaultFlatFileItemReaderTests extends TestCase { inputSource.read(); // get restart data - StreamContext streamContext = inputSource.getRestartData(); + StreamContext streamContext = inputSource.getStreamContext(); assertEquals("4", (String) streamContext.getProperties().getProperty( DefaultFlatFileItemReader.READ_STATISTICS_NAME)); // close input diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java index 4748575cb..616736c32 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java @@ -320,7 +320,7 @@ public class FlatFileItemWriterTests extends TestCase { commit(); // get restart data - StreamContext restartData = inputSource.getRestartData(); + StreamContext streamContext = inputSource.getStreamContext(); // close template inputSource.close(); @@ -338,7 +338,7 @@ public class FlatFileItemWriterTests extends TestCase { } // init with correct data - inputSource.restoreFrom(restartData); + inputSource.restoreFrom(streamContext); // write more lines inputSource.write("testLine6"); @@ -371,9 +371,9 @@ public class FlatFileItemWriterTests extends TestCase { } } - public void testDefaultRestartData() throws Exception { + public void testDefaultStreamContext() throws Exception { inputSource = new FlatFileItemWriter(); - StreamContext streamContext = inputSource.getRestartData(); + StreamContext streamContext = inputSource.getStreamContext(); assertNotNull(streamContext); assertEquals(1, streamContext.getProperties().size()); assertEquals("0", streamContext.getProperties().getProperty(FlatFileItemWriter.RESTART_DATA_NAME)); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java index 6bb890b6e..dd56f2bc7 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java @@ -85,7 +85,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra Foo foo2 = (Foo) source.read(); assertEquals(2, foo2.getValue()); - StreamContext streamContext = getAsRestartable(source).getRestartData(); + StreamContext streamContext = getAsRestartable(source).getStreamContext(); // create new input source source = createItemReader(); @@ -107,7 +107,7 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra Foo foo2 = (Foo) source.read(); assertEquals(2, foo2.getValue()); - StreamContext streamContext = getAsRestartable(source).getRestartData(); + StreamContext streamContext = getAsRestartable(source).getStreamContext(); // create new input source source = createItemReader(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java index 08f0b828e..e99ef2c55 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java @@ -92,7 +92,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr Foo foo2 = (Foo) source.read(); assertEquals(2, foo2.getValue()); - StreamContext streamContext = getAsRestartable(source).getRestartData(); + StreamContext streamContext = getAsRestartable(source).getStreamContext(); // create new input source source = createItemReader(); @@ -114,7 +114,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr Foo foo2 = (Foo) source.read(); assertEquals(2, foo2.getValue()); - StreamContext streamContext = getAsRestartable(source).getRestartData(); + StreamContext streamContext = getAsRestartable(source).getStreamContext(); // create new input source source = createItemReader(); @@ -217,7 +217,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr rollback(); - StreamContext streamContext = getAsRestartable(source).getRestartData(); + StreamContext streamContext = getAsRestartable(source).getStreamContext(); // create new input source source = createItemReader(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventReaderItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventReaderItemReaderTests.java index dfcd72c36..aac2f64c1 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventReaderItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventReaderItemReaderTests.java @@ -117,7 +117,7 @@ public class StaxEventReaderItemReaderTests extends TestCase { */ public void testRestart() { source.read(); - StreamContext streamContext = source.getRestartData(); + StreamContext streamContext = source.getStreamContext(); assertEquals("1", streamContext.getProperties(). getProperty("StaxEventReaderItemReader.recordcount")); List expectedAfterRestart = (List) source.read(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java index 0a364c8c6..9096e15e6 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java @@ -92,7 +92,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { // write records writer.write(record); writer.getSynchronization().afterCompletion(TransactionSynchronization.STATUS_COMMITTED); - StreamContext streamContext = writer.getRestartData(); + StreamContext streamContext = writer.getStreamContext(); // create new writer from saved restart data and continue writing writer = createItemWriter(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java index fb5bae51b..e795dfa3a 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java @@ -75,8 +75,8 @@ public class DelegatingItemReaderTests extends TestCase { /** * Gets restart data from the input template */ - public void testGetRestartData() { - Properties props = itemProvider.getRestartData().getProperties(); + public void testGetStreamContext() { + Properties props = itemProvider.getStreamContext().getProperties(); assertEquals("foo", props.getProperty("value")); } @@ -102,7 +102,7 @@ public class DelegatingItemReaderTests extends TestCase { return PropertiesConverter.stringToProperties("a=b"); } - public StreamContext getRestartData() { + public StreamContext getStreamContext() { return new GenericStreamContext(PropertiesConverter.stringToProperties("value=foo")); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/CompositeItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/CompositeItemWriterTests.java index 2b38984ba..bce2b4807 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/CompositeItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/CompositeItemWriterTests.java @@ -3,17 +3,11 @@ package org.springframework.batch.item.writer; import java.util.ArrayList; import java.util.Iterator; import java.util.List; -import java.util.Properties; import junit.framework.TestCase; import org.easymock.MockControl; -import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.StreamContext; -import org.springframework.batch.item.stream.GenericStreamContext; -import org.springframework.batch.item.writer.CompositeItemWriter; -import org.springframework.batch.statistics.StatisticsProvider; /** * Tests for {@link CompositeItemWriter} @@ -57,37 +51,7 @@ public class CompositeItemWriterTests extends TestCase { control.verify(); } } - - /** - * All Restartable processors should be restarted, not-Restartable processors should be ignored. - */ - public void testRestart() { - //this mock with undefined behaviour makes sure not-Restartable processor is ignored - MockControl p1c = MockControl.createStrictControl(ItemWriter.class); - final ItemWriter p1 = (ItemWriter) p1c.getMock(); - final ItemWriter p2 = new StubItemWriter(); - final ItemWriter p3 = new StubItemWriter(); - List itemProcessors = new ArrayList(){{ - add(p1); - add(p2); - add(p3); - }}; - itemProcessor.setDelegates(itemProcessors); - - StreamContext rd = itemProcessor.getRestartData(); - itemProcessor.restoreFrom(rd); - - for (Iterator iterator = itemProcessors.iterator(); iterator.hasNext();) { - ItemWriter processor = (ItemWriter) iterator.next(); - if (processor instanceof StubItemWriter) { - assertTrue("Injected processors are restarted", - ((StubItemWriter)processor).restarted); - } - } - - } - public void testClose() throws Exception { final int NUMBER_OF_PROCESSORS = 10; @@ -115,49 +79,4 @@ public class CompositeItemWriterTests extends TestCase { } - /** - * Stub for testing restart. Checks the restart data received is the same that was returned by - * getRestartData() - */ - private static class StubItemWriter implements ItemWriter, ItemStream, StatisticsProvider { - - private static final String RESTART_KEY = "restartData"; - private static final String STATS_KEY = "stats"; - - private boolean restarted = false; - - private final int hashCode = this.hashCode(); - - - public StreamContext getRestartData() { - Properties props = new Properties(){{ - setProperty(RESTART_KEY, String.valueOf(hashCode)); - }}; - return new GenericStreamContext(props); - } - - public void restoreFrom(StreamContext data) { - if (Integer.valueOf(data.getProperties().getProperty(RESTART_KEY)).intValue() != hashCode()) { - fail("received restart data is not the same which was saved"); - } - restarted = true; - } - - public void write(Object data) throws Exception { - // do nothing - } - - public Properties getStatistics() { - return new Properties() {{ - setProperty(STATS_KEY, String.valueOf(hashCode)); - }}; - } - - public void close() throws Exception { - } - - public void open() throws Exception { - } - } - } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java index 0bd9e43c2..07402aa7f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java @@ -59,8 +59,8 @@ public class ItemWriterItemProcessorTests extends TestCase { /** * Gets restart data from the input template */ - public void testGetRestartData() { - Properties props = processor.getRestartData().getProperties(); + public void testGetStreamContext() { + Properties props = processor.getStreamContext().getProperties(); assertEquals("foo", props.getProperty("value")); } @@ -78,10 +78,10 @@ public class ItemWriterItemProcessorTests extends TestCase { * Forward restart data to input template * @throws Exception */ - public void testGetRestartDataWithoutRestartable() throws Exception { + public void testGetStreamContextWithoutItemStream() throws Exception { processor.setDelegate(null); try { - processor.getRestartData(); + processor.getStreamContext(); fail("Expected IllegalStateException"); } catch (IllegalStateException e) { @@ -152,7 +152,7 @@ public class ItemWriterItemProcessorTests extends TestCase { return PropertiesConverter.stringToProperties("a=b"); } - public StreamContext getRestartData() { + public StreamContext getStreamContext() { return new GenericStreamContext(PropertiesConverter.stringToProperties("value=foo")); } diff --git a/spring-batch-samples/src/main/resources/jobs/footballJob.xml b/spring-batch-samples/src/main/resources/jobs/footballJob.xml index accca5cb3..479b77a83 100644 --- a/spring-batch-samples/src/main/resources/jobs/footballJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/footballJob.xml @@ -19,7 +19,6 @@ - - @@ -63,7 +61,6 @@ - diff --git a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml index 551cb823b..c3e22fa92 100644 --- a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml @@ -48,7 +48,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/restartSample.xml b/spring-batch-samples/src/main/resources/jobs/restartSample.xml index 97cfd80d1..006321b50 100644 --- a/spring-batch-samples/src/main/resources/jobs/restartSample.xml +++ b/spring-batch-samples/src/main/resources/jobs/restartSample.xml @@ -18,7 +18,7 @@ + class="org.springframework.batch.item.reader.ValidatingItemReader" scope="step"> @@ -31,7 +31,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/simple-container-definition.xml b/spring-batch-samples/src/main/resources/simple-container-definition.xml index 16d42cc51..629bfd2e5 100644 --- a/spring-batch-samples/src/main/resources/simple-container-definition.xml +++ b/spring-batch-samples/src/main/resources/simple-container-definition.xml @@ -87,7 +87,7 @@ - + + - + diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RestartFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RestartFunctionalTests.java index 38e0a4e76..075651dca 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RestartFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RestartFunctionalTests.java @@ -75,7 +75,7 @@ public class RestartFunctionalTests extends AbstractBatchLauncherTests { } int medium = jdbcTemplate.queryForInt("SELECT COUNT(*) FROM TRADE"); - // assert based on commit inyerval = 2 + // assert based on commit interval = 2 assertEquals(before+2, medium); runJob();