Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering)

Browse Source 
http://jira.springframework.org/browse/BATCH-7

Rename Restartable to ItemStream
This commit is contained in:
dsyer
2008-01-30 17:56:56 +00:00
parent 02f1ae5c0c
commit a2117e4f22
50 changed files with 398 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-batch-execution/src/main/java/org/springframework/batch/execution/job/simple/SimpleJob.java
View File

@@ -129,9 +129,8 @@ public class SimpleJob extends JobSupport {
private boolean shouldStart(StepInstance stepInstance, Step step) {
if (stepInstance.getStatus() == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) {
// step is complete, false should be returned, indicated that the
// step should
// not be started
// step is complete, false should be returned, indicating that the
// step should not be started
return false;
}

6
spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java
View File

@@ -34,7 +34,7 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.execution.repository.dao.JobDao;
import org.springframework.batch.execution.repository.dao.StepDao;
import org.springframework.batch.restart.GenericRestartData;
import org.springframework.batch.stream.GenericStreamContext;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.util.Assert;
@@ -304,7 +304,7 @@ public class SimpleJobRepository implements JobRepository {
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 GenericRestartData(new Properties()));
stepInstance.setRestartData(new GenericStreamContext(new Properties()));
}
stepInstances.add(stepInstance);
}
@@ -327,7 +327,7 @@ 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 GenericRestartData(new Properties()));
step.setRestartData(new GenericStreamContext(new Properties()));
}
stepInstances.add(step);
}

14
spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java
View File

@@ -31,8 +31,8 @@ import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.execution.repository.dao.JdbcJobDao.JobExecutionRowMapper;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.restart.GenericRestartData;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.stream.GenericStreamContext;
import org.springframework.batch.stream.StreamContext;
import org.springframework.batch.support.PropertiesConverter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.IncorrectResultSizeDataAccessException;
@@ -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 GenericRestartData(PropertiesConverter.stringToProperties(rs.getString(3))));
step.setRestartData(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 GenericRestartData(PropertiesConverter.stringToProperties(rs.getString(3))));
step.setRestartData(new GenericStreamContext(PropertiesConverter.stringToProperties(rs.getString(3))));
return step;
}
};
@@ -422,9 +422,9 @@ public class JdbcStepDao implements StepDao, InitializingBean {
Assert.notNull(step.getId(), "Step Id cannot be null.");
Properties restartProps = null;
RestartData restartData = step.getRestartData();
if (restartData != null) {
restartProps = restartData.getProperties();
StreamContext streamContext = step.getRestartData();
if (streamContext != null) {
restartProps = streamContext.getProperties();
}
Object[] parameters = new Object[] { step.getStatus().toString(),

10
spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java
View File

@@ -25,7 +25,7 @@ import java.util.Set;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.stream.StreamContext;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
public class MapStepDao implements StepDao {
@@ -80,8 +80,8 @@ public class MapStepDao implements StepDao {
return new ArrayList(steps);
}
public RestartData getRestartData(Long stepId) {
return (RestartData) restartsById.get(stepId);
public StreamContext getRestartData(Long stepId) {
return (StreamContext) restartsById.get(stepId);
}
public int getStepExecutionCount(StepInstance stepInstance) {
@@ -99,8 +99,8 @@ public class MapStepDao implements StepDao {
executions.add(stepExecution);
}
public void saveRestartData(Long stepId, RestartData restartData) {
restartsById.put(stepId, restartData);
public void saveRestartData(Long stepId, StreamContext streamContext) {
restartsById.put(stepId, streamContext);
}
public List findStepExecutions(StepInstance step) {

18
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
View File

@@ -43,11 +43,11 @@ import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHa
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.restart.Restartable;
import org.springframework.batch.statistics.SimpleStatisticsService;
import org.springframework.batch.statistics.StatisticsProvider;
import org.springframework.batch.statistics.StatisticsService;
import org.springframework.batch.stream.ItemStream;
import org.springframework.batch.stream.StreamContext;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
@@ -391,20 +391,20 @@ public class SimpleStepExecutor {
/**
* @param tasklet
* @return restart data from the {@link Tasklet} if it is
* {@link Restartable}
* {@link ItemStream}
*/
private RestartData getRestartData(Tasklet tasklet) {
if (tasklet instanceof Restartable) {
return ((Restartable) tasklet).getRestartData();
private StreamContext getRestartData(Tasklet tasklet) {
if (tasklet instanceof ItemStream) {
return ((ItemStream) tasklet).getRestartData();
}
else {
return null;
}
}
private void restoreFromRestartData(Tasklet tasklet, RestartData restartData) {
if (tasklet instanceof Restartable && restartData != null) {
((Restartable) tasklet).restoreFrom(restartData);
private void restoreFromRestartData(Tasklet tasklet, StreamContext streamContext) {
if (tasklet instanceof ItemStream && streamContext != null) {
((ItemStream) tasklet).restoreFrom(streamContext);
}
}

52
spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTasklet.java
View File

@@ -20,39 +20,39 @@ import java.util.Properties;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.restart.GenericRestartData;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.restart.Restartable;
import org.springframework.batch.stream.GenericStreamContext;
import org.springframework.batch.stream.ItemStream;
import org.springframework.batch.stream.StreamContext;
import org.springframework.batch.support.PropertiesConverter;
/**
* An extension of {@link ItemOrientedTasklet} that delegates calls to
* {@link Restartable} to the reader and writer.
* {@link ItemStream} to the reader and writer.
*
* @see ItemReader
* @see ItemWriter
* @see Restartable
* @see ItemStream
*
* @author Lucas Ward
* @author Dave Syer
*
*/
public class RestartableItemOrientedTasklet extends ItemOrientedTasklet implements Restartable {
public class RestartableItemOrientedTasklet extends ItemOrientedTasklet implements ItemStream {
/**
* @see Restartable#getRestartData()
* @see ItemStream#getRestartData()
*/
public RestartData getRestartData() {
public StreamContext getRestartData() {
RestartData itemProviderRestartData = null;
RestartData itemProcessorRestartData = null;
StreamContext itemProviderRestartData = null;
StreamContext itemProcessorRestartData = null;
if (itemProvider instanceof Restartable) {
itemProviderRestartData = ((Restartable) itemProvider).getRestartData();
if (itemProvider instanceof ItemStream) {
itemProviderRestartData = ((ItemStream) itemProvider).getRestartData();
}
if (itemWriter instanceof Restartable) {
itemProcessorRestartData = ((Restartable) itemWriter).getRestartData();
if (itemWriter instanceof ItemStream) {
itemProcessorRestartData = ((ItemStream) itemWriter).getRestartData();
}
RestartableItemOrientedTaskletRestartData restartData = new RestartableItemOrientedTaskletRestartData(itemProviderRestartData, itemProcessorRestartData);
@@ -61,9 +61,9 @@ public class RestartableItemOrientedTasklet extends ItemOrientedTasklet implemen
}
/**
* @see Restartable#restoreFrom(RestartData)
* @see ItemStream#restoreFrom(StreamContext)
*/
public void restoreFrom(RestartData data) {
public void restoreFrom(StreamContext data) {
if (data == null || data.getProperties() == null)
return;
@@ -76,33 +76,33 @@ public class RestartableItemOrientedTasklet extends ItemOrientedTasklet implemen
moduleRestartData = new RestartableItemOrientedTaskletRestartData(data.getProperties());
}
if (itemProvider instanceof Restartable) {
((Restartable) itemProvider).restoreFrom(moduleRestartData.readerData);
if (itemProvider instanceof ItemStream) {
((ItemStream) itemProvider).restoreFrom(moduleRestartData.readerData);
}
if (itemWriter instanceof Restartable) {
((Restartable) itemWriter).restoreFrom(moduleRestartData.writerData);
if (itemWriter instanceof ItemStream) {
((ItemStream) itemWriter).restoreFrom(moduleRestartData.writerData);
}
}
private class RestartableItemOrientedTaskletRestartData implements RestartData {
private class RestartableItemOrientedTaskletRestartData implements StreamContext {
private static final String READER_KEY = "DATA_PROVIDER";
private static final String WRITER_KEY = "DATA_PROCESSOR";
private RestartData readerData;
private StreamContext readerData;
private RestartData writerData;
private StreamContext writerData;
public RestartableItemOrientedTaskletRestartData(RestartData providerData, RestartData writerData) {
public RestartableItemOrientedTaskletRestartData(StreamContext providerData, StreamContext writerData) {
this.readerData = providerData;
this.writerData = writerData;
}
public RestartableItemOrientedTaskletRestartData(Properties data) {
readerData = new GenericRestartData(PropertiesConverter
readerData = new GenericStreamContext(PropertiesConverter
.stringToProperties(data.getProperty(READER_KEY)));
writerData = new GenericRestartData(PropertiesConverter.stringToProperties(data
writerData = new GenericStreamContext(PropertiesConverter.stringToProperties(data
.getProperty(WRITER_KEY)));
}

6
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/MockStepDao.java
View File

@@ -22,7 +22,7 @@ import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.execution.repository.dao.StepDao;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.stream.StreamContext;
public class MockStepDao implements StepDao {
@@ -46,7 +46,7 @@ public class MockStepDao implements StepDao {
return newSteps;
}
public RestartData getRestartData(Long stepId) {
public StreamContext getRestartData(Long stepId) {
return null;
}
@@ -57,7 +57,7 @@ public class MockStepDao implements StepDao {
public void save(StepExecution stepExecution) {
}
public void saveRestartData(Long stepId, RestartData restartData) {
public void saveRestartData(Long stepId, StreamContext streamContext) {
}
public void update(StepInstance step) {

6
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java
View File

@@ -37,7 +37,7 @@ import org.springframework.batch.core.domain.StepSupport;
import org.springframework.batch.core.repository.BatchRestartException;
import org.springframework.batch.execution.repository.dao.JobDao;
import org.springframework.batch.execution.repository.dao.StepDao;
import org.springframework.batch.restart.GenericRestartData;
import org.springframework.batch.stream.GenericStreamContext;
/*
* Test SimpleJobRepository. The majority of test cases are tested using EasyMock,
@@ -408,7 +408,7 @@ public class SimpleJobRepositoryTests extends TestCase {
databaseStep1.setRestartData(null);
stepDaoControl.setReturnValue(databaseStep1);
stepDao.createStep(databaseJob, "TestStep2");
databaseStep2.setRestartData(new GenericRestartData(null));
databaseStep2.setRestartData(new GenericStreamContext(null));
stepDaoControl.setReturnValue(databaseStep2);
jobDao.save(new JobExecution(databaseJob));
jobDaoControl.setMatcher(new ArgumentsMatcher(){
@@ -443,7 +443,7 @@ public class SimpleJobRepositoryTests extends TestCase {
stepDao.getStepExecutionCount(databaseStep1);
stepDaoControl.setReturnValue(1);
stepDao.findStep(databaseJob, "TestStep2");
databaseStep2.setRestartData(new GenericRestartData(null));
databaseStep2.setRestartData(new GenericStreamContext(null));
stepDaoControl.setReturnValue(databaseStep2);
stepDao.getStepExecutionCount(databaseStep2);
stepDaoControl.setReturnValue(1);

10
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java
View File

@@ -30,8 +30,8 @@ import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.core.runtime.ExitCodeExceptionClassifier;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.restart.GenericRestartData;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.stream.GenericStreamContext;
import org.springframework.batch.stream.StreamContext;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
import org.springframework.util.ClassUtils;
@@ -152,13 +152,13 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
step1.setStatus(BatchStatus.COMPLETED);
Properties data = new Properties();
data.setProperty("restart.key1", "restartData");
RestartData restartData = new GenericRestartData(data);
step1.setRestartData(restartData);
StreamContext streamContext = new GenericStreamContext(data);
step1.setRestartData(streamContext);
stepDao.update(step1);
StepInstance tempStep = stepDao.findStep(jobInstance, step1.getName());
assertEquals(tempStep, step1);
assertEquals(tempStep.getRestartData().getProperties().toString(),
restartData.getProperties().toString());
streamContext.getProperties().toString());
}
public void testSaveStepExecution(){

10
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java
View File

@@ -27,8 +27,8 @@ import org.springframework.batch.core.domain.JobParameters;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.execution.repository.dao.MapStepDao;
import org.springframework.batch.restart.GenericRestartData;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.stream.GenericStreamContext;
import org.springframework.batch.stream.StreamContext;
public class MapStepDaoTests extends TestCase {
@@ -111,13 +111,13 @@ public class MapStepDaoTests extends TestCase {
step.setStatus(BatchStatus.COMPLETED);
Properties data = new Properties();
data.setProperty("restart.key1", "restartData");
RestartData restartData = new GenericRestartData(data);
step.setRestartData(restartData);
StreamContext streamContext = new GenericStreamContext(data);
step.setRestartData(streamContext);
dao.update(step);
StepInstance tempStep = dao.findStep(job, step.getName());
assertEquals(tempStep, step);
assertEquals(tempStep.getRestartData().getProperties().toString(),
restartData.getProperties().toString());
streamContext.getProperties().toString());
}
}

10
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java
View File

@@ -50,10 +50,10 @@ import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
import org.springframework.batch.repeat.interceptor.RepeatInterceptorAdapter;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.restart.Restartable;
import org.springframework.batch.statistics.StatisticsProvider;
import org.springframework.batch.statistics.StatisticsService;
import org.springframework.batch.stream.ItemStream;
import org.springframework.batch.stream.StreamContext;
import org.springframework.batch.support.PropertiesConverter;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
@@ -455,7 +455,7 @@ public class SimpleStepExecutorTests extends TestCase {
assertEquals(0, map.size());
}
private class MockRestartableTasklet implements Tasklet, Restartable {
private class MockRestartableTasklet implements Tasklet, ItemStream {
private boolean getRestartDataCalled = false;
@@ -465,12 +465,12 @@ public class SimpleStepExecutorTests extends TestCase {
return ExitStatus.FINISHED;
}
public RestartData getRestartData() {
public StreamContext getRestartData() {
getRestartDataCalled = true;
return null;
}
public void restoreFrom(RestartData data) {
public void restoreFrom(StreamContext data) {
restoreFromCalled = true;
}

30
spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/RestartableItemOrientedTaskletTests.java
View File

@@ -22,9 +22,9 @@ import junit.framework.TestCase;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.restart.GenericRestartData;
import org.springframework.batch.restart.RestartData;
import org.springframework.batch.restart.Restartable;
import org.springframework.batch.stream.GenericStreamContext;
import org.springframework.batch.stream.ItemStream;
import org.springframework.batch.stream.StreamContext;
import org.springframework.batch.support.PropertiesConverter;
/**
@@ -32,9 +32,9 @@ import org.springframework.batch.support.PropertiesConverter;
*/
public class RestartableItemOrientedTaskletTests extends TestCase {
private static class MockProvider implements ItemReader, Restartable {
private static class MockProvider implements ItemReader, ItemStream {
RestartData data = new RestartData() {
StreamContext data = new StreamContext() {
public Properties getProperties() {
return PropertiesConverter.stringToProperties("a=b");
@@ -46,11 +46,11 @@ public class RestartableItemOrientedTaskletTests extends TestCase {
return null;
}
public RestartData getRestartData() {
public StreamContext getRestartData() {
return data;
}
public void restoreFrom(RestartData data) {
public void restoreFrom(StreamContext data) {
// restart data should be same as returned by getRestartData
assertEquals(this.data.getProperties(), data.getProperties());
}
@@ -64,9 +64,9 @@ public class RestartableItemOrientedTaskletTests extends TestCase {
}
private static class MockWriter implements ItemWriter, Restartable {
private static class MockWriter implements ItemWriter, ItemStream {
RestartData data = new RestartData() {
StreamContext data = new StreamContext() {
public Properties getProperties() {
return PropertiesConverter.stringToProperties("x=y");
}
@@ -75,11 +75,11 @@ public class RestartableItemOrientedTaskletTests extends TestCase {
public void write(Object data) {
}
public RestartData getRestartData() {
public StreamContext getRestartData() {
return data;
}
public void restoreFrom(RestartData data) {
public void restoreFrom(StreamContext data) {
// restart data should be same as returned by getRestartData
assertEquals(this.data.getProperties(), data.getProperties());
}
@@ -107,7 +107,7 @@ public class RestartableItemOrientedTaskletTests extends TestCase {
module.setItemWriter(itemWriter);
// get restart data
RestartData data = module.getRestartData();
StreamContext data = module.getRestartData();
assertNotNull(data);
// restore from restart data (see asserts in mock classes)
module.restoreFrom(data);
@@ -125,9 +125,9 @@ public class RestartableItemOrientedTaskletTests extends TestCase {
module.setItemWriter(itemWriter);
// get restart data
RestartData data = module.getRestartData();
StreamContext data = module.getRestartData();
assertNotNull(data);
data = new GenericRestartData(data.getProperties());
data = new GenericStreamContext(data.getProperties());
// restore from restart data (see asserts in mock classes)
module.restoreFrom(data);
}
@@ -140,7 +140,7 @@ public class RestartableItemOrientedTaskletTests extends TestCase {
module.setItemWriter(null);
// get restart data
RestartData data = module.getRestartData();
StreamContext data = module.getRestartData();
assertNotNull(data);
// restore from restart data (see asserts in mock classes)
module.restoreFrom(data);