REOPENED - issue BATCH-222: StreamManager for reader/writer abstractions uses StreamContext to position at the right place for restart

http://jira.springframework.org/browse/BATCH-222

ALign some dependencies - SimpleStepExecutor does not need a transaction manager

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AbstractStep.java

@@ -22,13 +22,14 @@ import org.springframework.batch.core.domain.StepSupport;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.tasklet.Tasklet;
 import org.springframework.batch.io.exception.BatchCriticalException;
+import org.springframework.batch.item.stream.SimpleStreamManager;
+import org.springframework.batch.item.stream.StreamManager;
 import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.util.Assert;
 
 /**
- * A {@link Step} implementation that provides common behaviour to
- * subclasses.
+ * A {@link Step} implementation that provides common behaviour to subclasses.
  * 
  * @author Dave Syer
  * 
@@ -45,6 +46,8 @@ public abstract class AbstractStep extends StepSupport {
 
 	private Tasklet tasklet;
 
+	private StreamManager streamManager;
+
 	/**
 	 * Default constructor.
 	 */
@@ -79,8 +82,7 @@ public abstract class AbstractStep extends StepSupport {
 	/**
 	 * Public setter for {@link JobRepository}.
 	 * 
-	 * @param jobRepository
-	 *            is a mandatory dependence (no default).
+	 * @param jobRepository is a mandatory dependence (no default).
 	 */
 	public void setJobRepository(JobRepository jobRepository) {
 		this.jobRepository = jobRepository;
@@ -93,7 +95,16 @@ public abstract class AbstractStep extends StepSupport {
 	public void setTransactionManager(PlatformTransactionManager transactionManager) {
 		this.transactionManager = transactionManager;
 	}
-	
+
+	/**
+	 * Public setter for the {@link StreamManager}. Set either this or the
+	 * transaction manager, but not both.
+	 * @param streamManager the {@link StreamManager} to set.
+	 */
+	public void setStreamManager(StreamManager streamManager) {
+		this.streamManager = streamManager;
+	}
+
 	/**
 	 * Assert that all mandatory properties are set (the {@link JobRepository}).
 	 * 
@@ -105,10 +116,14 @@ public abstract class AbstractStep extends StepSupport {
 
 	protected void assertMandatoryProperties() {
 		Assert.notNull(jobRepository, "JobRepository is mandatory");
-		Assert.notNull(transactionManager, "TransactionManager is mandatory");
+		Assert.state(transactionManager != null || streamManager != null,
+				"Either StreamManager or TransactionManager must be set");
+		Assert.state(transactionManager == null || streamManager == null,
+				"Only one of StreamManager or TransactionManager must be set");
 	}
-	
-	/* (non-Javadoc)
+
+	/*
+	 * (non-Javadoc)
 	 * @see org.springframework.batch.core.domain.StepSupport#process(org.springframework.batch.core.domain.StepExecution)
 	 */
 	public void execute(StepExecution stepExecution) throws StepInterruptedException, BatchCriticalException {
@@ -126,15 +141,20 @@ public abstract class AbstractStep extends StepSupport {
 	}
 
 	/**
-	 * @return
+	 * @return a {@link SimpleStepExecutor} that can be used to launch the job.
 	 */
 	protected SimpleStepExecutor createStepExecutor() {
 		assertMandatoryProperties();
-		SimpleStepExecutor executor = new SimpleStepExecutor(this);
+		// Do not set the streamManager field if it is null, otherwise
+		// the mandatory properties check will fail.
+		StreamManager manager = streamManager;
+		if (streamManager == null) {
+			manager = new SimpleStreamManager(transactionManager);
+		}
+		SimpleStepExecutor executor = new SimpleStepExecutor(manager, this);
 		executor.setRepository(jobRepository);
 		executor.applyConfiguration(this);
 		executor.setTasklet(tasklet);
-		executor.setTransactionManager(transactionManager);
 		return executor;
 	}
 

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

@@ -45,8 +45,6 @@ import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
 import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler;
 import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
 import org.springframework.batch.repeat.support.RepeatTemplate;
-import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
-import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.util.Assert;
 
@@ -83,9 +81,6 @@ public class SimpleStepExecutor {
 	// default to checking current thread for interruption.
 	private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy();
 
-	// Not for production use...
-	protected PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
-
 	private Tasklet tasklet;
 
 	private AbstractStep step;
@@ -93,10 +88,11 @@ public class SimpleStepExecutor {
 	private StreamManager streamManager;
 
 	/**
-	 * Package private constructor so the factory can create a the executor.
+	 * Package private constructor so the step can create a the executor.
 	 */
-	SimpleStepExecutor(AbstractStep abstractStep) {
+	SimpleStepExecutor(StreamManager streamManager, AbstractStep abstractStep) {
 		this.step = abstractStep;
+		this.streamManager = streamManager;
 	}
 
 	/**
@@ -113,14 +109,6 @@ public class SimpleStepExecutor {
 		this.streamManager = streamManager;
 	}
 
-	/**
-	 * Injected strategy for transaction management
-	 * @param transactionManager
-	 */
-	public void setTransactionManager(PlatformTransactionManager transactionManager) {
-		this.transactionManager = transactionManager;
-	}
-
 	/**
 	 * Injected strategy for storage and retrieval of persistent step
 	 * information. Mandatory property.
@@ -176,10 +164,6 @@ public class SimpleStepExecutor {
 
 		ExitStatus status = ExitStatus.FAILED;
 
-		if (streamManager == null) {
-			streamManager = new SimpleStreamManager(transactionManager);
-		}
-
 		StepContext parentStepContext = StepSynchronizationManager.getContext();
 		final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, streamManager);
 		StepSynchronizationManager.register(stepContext);

spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepTests.java

@@ -25,6 +25,7 @@ import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
 import org.springframework.batch.core.domain.StepExecution;
 import org.springframework.batch.core.domain.StepInstance;
+import org.springframework.batch.item.stream.SimpleStreamManager;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
@@ -48,23 +49,22 @@ public class SimpleStepTests extends TestCase {
 		configuration.setTransactionManager(new ResourcelessTransactionManager());
 		final List list = new ArrayList();
 		configuration.setExceptionHandler(new ExceptionHandler() {
-			public void handleException(RepeatContext context,
-					Throwable throwable) throws RuntimeException {
+			public void handleException(RepeatContext context, Throwable throwable) throws RuntimeException {
 				list.add(throwable);
 				throw new RuntimeException("Oops");
 			}
 		});
-		SimpleStepExecutor executor = (SimpleStepExecutor) configuration
-				.createStepExecutor();
-		StepExecution stepExecution = new StepExecution(new StepInstance(
-				new Long(11)), new JobExecution(new JobInstance(new Long(0L), new JobParameters()),
-				new Long(12)));
+		SimpleStepExecutor executor = (SimpleStepExecutor) configuration.createStepExecutor();
+		StepExecution stepExecution = new StepExecution(new StepInstance(new Long(11)), new JobExecution(
+				new JobInstance(new Long(0L), new JobParameters()), new Long(12)));
 		try {
 			executor.execute(stepExecution);
 			fail("Expected RuntimeException");
-		} catch (NullPointerException e) {
+		}
+		catch (NullPointerException e) {
 			throw e;
-		}catch (RuntimeException e) {
+		}
+		catch (RuntimeException e) {
 			assertEquals("Oops", e.getMessage());
 		}
 		assertEquals(1, list.size());
@@ -74,11 +74,11 @@ public class SimpleStepTests extends TestCase {
 		try {
 			new SimpleStep().createStepExecutor();
 			fail("Expected IllegalArgumentException");
-		} catch (IllegalArgumentException e) {
+		}
+		catch (IllegalArgumentException e) {
 			// expected
-			assertTrue("Error message does not contain JobRepository: "
-					+ e.getMessage(),
-					e.getMessage().indexOf("JobRepository") >= 0);
+			assertTrue("Error message does not contain JobRepository: " + e.getMessage(), e.getMessage().indexOf(
+					"JobRepository") >= 0);
 		}
 	}
 
@@ -86,9 +86,45 @@ public class SimpleStepTests extends TestCase {
 		try {
 			new SimpleStep().afterPropertiesSet();
 			fail("Expected IllegalArgumentException");
-		} catch (IllegalArgumentException e) {
+		}
+		catch (IllegalArgumentException e) {
 			// expected
 		}
 	}
 
+	public void testMandatoryPropertiesNoTransactionManagerOrStreamManager() throws Exception {
+		try {
+			SimpleStep configuration = new SimpleStep("foo");
+			configuration.setJobRepository(new JobRepositorySupport());
+			configuration.assertMandatoryProperties();
+			fail("Experetscted IllegalStateException");
+		}
+		catch (IllegalStateException e) {
+			// expected
+		}
+	}
+
+	public void testMandatoryPropertiesTransactionManagerAndStreamManager() throws Exception {
+		try {
+			SimpleStep configuration = new SimpleStep("foo");
+			configuration.setJobRepository(new JobRepositorySupport());
+			configuration.setTransactionManager(new ResourcelessTransactionManager());
+			configuration.setStreamManager(new SimpleStreamManager());
+			configuration.assertMandatoryProperties();
+			fail("Expected IllegalStateException");
+		}
+		catch (IllegalStateException e) {
+			// expected
+		}
+	}
+
+	public void testMandatoryPropertiesAfterExecution() throws Exception {
+		SimpleStep step = new SimpleStep();
+		step.setJobRepository(new JobRepositorySupport());
+		step.setTransactionManager(new ResourcelessTransactionManager());
+		assertNotNull(step.createStepExecutor());
+		// If we do that again, we don't expect a different result (e.g.
+		// mandatory properties test failing).
+		assertNotNull(step.createStepExecutor());
+	}
 }