* 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">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 @@