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

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

Remove Statistics* - use StreamContext instead.

spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java

@@ -21,7 +21,6 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 import java.util.Set;
 
 import org.springframework.batch.core.domain.StepExecution;
@@ -30,8 +29,6 @@ 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;
 
 /**
  * Simple implementation of {@link StepContext}.
@@ -47,8 +44,6 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements
 
 	private StepExecution stepExecution;
 
-	private StatisticsService statisticsService;
-
 	private StreamManager streamManager;
 
 	private StreamContext streamContext;
@@ -57,24 +52,22 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements
 	 * Default constructor.
 	 */
 	public SimpleStepContext(StepExecution stepExecution) {
-		this(stepExecution, null, null, null);
+		this(stepExecution, null, null);
 	}
 
 	/**
 	 * Default constructor.
 	 */
 	public SimpleStepContext(StepExecution stepExecution, StepContext parent) {
-		this(stepExecution, parent, null, null);
+		this(stepExecution, parent, null);
 	}
 
 	/**
 	 * @param object
 	 */
-	public SimpleStepContext(StepExecution stepExecution, StepContext parent, StatisticsService statisticsService,
-			StreamManager streamManager) {
+	public SimpleStepContext(StepExecution stepExecution, StepContext parent, StreamManager streamManager) {
 		super();
 		this.parent = parent;
-		this.statisticsService = statisticsService;
 		this.streamManager = streamManager;
 		this.stepExecution = stepExecution;
 	}
@@ -86,9 +79,6 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements
 	 */
 	public void setAttribute(String name, Object value) {
 		super.setAttribute(name, value);
-		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);
@@ -99,17 +89,6 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements
 		}
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics()
-	 */
-	public Properties getStatistics() {
-		if (statisticsService == null) {
-			return new Properties();
-		}
-		return statisticsService.getStatistics(this);
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * 

spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java

@@ -19,7 +19,6 @@ 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;
 
 /**
@@ -28,7 +27,7 @@ import org.springframework.core.AttributeAccessor;
  * @author Dave Syer
  * 
  */
-public interface StepContext extends AttributeAccessor, StreamContextProvider, StatisticsProvider {
+public interface StepContext extends AttributeAccessor, StreamContextProvider {
 
 	/**
 	 * Accessor for the {@link StepExecution} associated with the currently

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

@@ -17,7 +17,6 @@
 package org.springframework.batch.execution.step.simple;
 
 import java.util.Date;
-import java.util.Properties;
 
 import org.springframework.batch.core.domain.BatchStatus;
 import org.springframework.batch.core.domain.Step;
@@ -34,6 +33,7 @@ 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.StreamContext;
 import org.springframework.batch.item.stream.SimpleStreamManager;
 import org.springframework.batch.item.stream.StreamManager;
 import org.springframework.batch.repeat.ExitStatus;
@@ -45,9 +45,6 @@ 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.statistics.SimpleStatisticsService;
-import org.springframework.batch.statistics.StatisticsProvider;
-import org.springframework.batch.statistics.StatisticsService;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionStatus;
@@ -91,8 +88,6 @@ public class SimpleStepExecutor {
 	// Not for production use...
 	protected PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
 
-	private StatisticsService statisticsService = new SimpleStatisticsService();
-
 	private Tasklet tasklet;
 
 	private AbstractStep step;
@@ -106,20 +101,6 @@ public class SimpleStepExecutor {
 		this.step = abstractStep;
 	}
 
-	/**
-	 * Public setter for the {@link StatisticsService}. 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 statisticsService the {@link StatisticsService} to set. Default is
-	 * a {@link SimpleStatisticsService}.
-	 */
-	public void setStatisticsService(StatisticsService statisticsService) {
-		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
@@ -198,8 +179,7 @@ public class SimpleStepExecutor {
 		ExitStatus status = ExitStatus.FAILED;
 
 		StepContext parentStepContext = StepSynchronizationManager.getContext();
-		final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, statisticsService,
-				streamManager);
+		final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, streamManager);
 		StepSynchronizationManager.register(stepContext);
 		// Add the job identifier so that it can be used to identify
 		// the conversation in StepScope
@@ -246,8 +226,8 @@ public class SimpleStepExecutor {
 										// TODO: check that stepExecution can
 										// aggregate these contributions if they
 										// come in asynchronously.
-										Properties statistics = stepContext.getStatistics();
-										contribution.setStatistics(statistics);
+										StreamContext statistics = stepContext.getStreamContext();
+										contribution.setStatistics(statistics.getProperties());
 										contribution.incrementCommitCount();
 										// Apply the contribution to the step
 										// only if chunk was successful

spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java

@@ -19,7 +19,6 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 
 import junit.framework.TestCase;
 
@@ -27,9 +26,8 @@ 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.ItemStreamAdapter;
 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;
 
 /**
@@ -136,47 +134,23 @@ public class SimpleStepContextTests extends TestCase {
 		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), new StubStreamManager(map));
-		assertEquals(1, context.getStatistics().size());
-		assertEquals("bar", context.getStatistics().getProperty("foo"));
+		context = new SimpleStepContext(null, null, new StubStreamManager(map));
+		assertEquals(1, context.getStreamContext().getProperties().size());
+		assertEquals("bar", context.getStreamContext().getProperties().getProperty("foo"));
 	}
 
 	public void testStatisticsServiceRegistration() throws Exception {
 		Map map = new HashMap();
-		context = new SimpleStepContext(null, null, new StubStatisticsService(map), new StubStreamManager(map));
-		StubStatisticsProvider provider = new StubStatisticsProvider();
+		context = new SimpleStepContext(null, null, new StubStreamManager(map));
+		ItemStreamAdapter provider = new ItemStreamAdapter();
 		context.setAttribute("foo", provider);
 		assertEquals(1, map.size());
 		assertEquals(context, map.keySet().iterator().next());
 		assertEquals(provider, map.values().iterator().next());
 	}
 
-	/**
-	 * @author Dave Syer
-	 * 
-	 */
-	private class StubStatisticsService implements StatisticsService {
-		private final Map map;
-
-		private StubStatisticsService(Map map) {
-			this.map = map;
-		}
-
-		public Properties getStatistics(Object key) {
-			return PropertiesConverter.stringToProperties("foo=bar");
-		}
-
-		public void register(Object key, StatisticsProvider provider) {
-			map.put(key, provider);
-		}
-	}
-
 	/**
 	 * @author Dave Syer
 	 * 
@@ -206,14 +180,4 @@ public class SimpleStepContextTests extends TestCase {
 		}
 	}
 
-	/**
-	 * @author Dave Syer
-	 * 
-	 */
-	private class StubStatisticsProvider implements StatisticsProvider {
-		public Properties getStatistics() {
-			return null;
-		}
-	}
-
 }

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

@@ -21,7 +21,6 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 
 import junit.framework.TestCase;
 
@@ -46,6 +45,7 @@ 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.stream.SimpleStreamManager;
 import org.springframework.batch.item.writer.AbstractItemWriter;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatContext;
@@ -54,8 +54,6 @@ 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.statistics.StatisticsProvider;
-import org.springframework.batch.statistics.StatisticsService;
 import org.springframework.batch.support.PropertiesConverter;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 
@@ -399,7 +397,7 @@ public class SimpleStepExecutorTests extends TestCase {
 		assertEquals(1, list.size());
 	}
 
-	public void testStatisticsService() throws Exception {
+	public void testStreamManager() throws Exception {
 		StepInstance step = new StepInstance(new Long(1));
 		step.setStepExecutionCount(1);
 		stepConfiguration.setTasklet(new Tasklet() {
@@ -414,22 +412,14 @@ public class SimpleStepExecutorTests extends TestCase {
 		assertEquals(null, stepExecution.getStatistics().getProperty("foo"));
 
 		final Map map = new HashMap();
-		stepExecutor.setStatisticsService(new StatisticsService() {
-			public Properties getStatistics(Object key) {
-				return PropertiesConverter.stringToProperties("foo=bar");
-			}
-
-			public void register(Object key, StatisticsProvider provider) {
-				map.put(key, provider);
+		stepExecutor.setStreamManager(new SimpleStreamManager() {
+			public StreamContext getStreamContext(Object key) {
+				// TODO Auto-generated method stub
+				return new GenericStreamContext(PropertiesConverter.stringToProperties("foo=bar"));
 			}
 		});
 
-		try {
-			stepExecutor.execute(stepExecution);
-		}
-		catch (Throwable t) {
-			fail();
-		}
+		stepExecutor.execute(stepExecution);
 
 		// At least once in that process the statistics service was asked for
 		// statistics...

spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java

@@ -34,7 +34,6 @@ import org.springframework.batch.item.writer.AbstractItemWriter;
 import org.springframework.batch.repeat.context.RepeatContextSupport;
 import org.springframework.batch.repeat.synch.RepeatSynchronizationManager;
 import org.springframework.batch.retry.policy.SimpleRetryPolicy;
-import org.springframework.batch.statistics.StatisticsProvider;
 import org.springframework.batch.support.PropertiesConverter;
 
 /**
@@ -135,7 +134,8 @@ public class ItemOrientedTaskletTests extends TestCase {
 		try {
 			module.execute();
 			fail("RuntimeException was expected");
-		} catch (RuntimeException bce) {
+		}
+		catch (RuntimeException bce) {
 			// expected
 			assertEquals("foo", bce.getMessage());
 		}
@@ -144,7 +144,8 @@ public class ItemOrientedTaskletTests extends TestCase {
 	public void testNotSkippable() throws Exception {
 		try {
 			module.skip();
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 			// Unexpected
 			throw e;
 		}
@@ -183,7 +184,7 @@ public class ItemOrientedTaskletTests extends TestCase {
 				return "bar";
 			}
 
-			public void close() throws StreamException {				
+			public void close() throws StreamException {
 			}
 		});
 
@@ -198,7 +199,8 @@ public class ItemOrientedTaskletTests extends TestCase {
 		try {
 			module.execute();
 			fail("Expected RuntimeException");
-		} catch (RuntimeException e) {
+		}
+		catch (RuntimeException e) {
 			assertEquals("FOO", e.getMessage());
 		}
 
@@ -206,8 +208,7 @@ public class ItemOrientedTaskletTests extends TestCase {
 
 		// verify method calls
 		assertEquals(1, list.size());
-		assertEquals("The item was not passed in to recover method", "bar",
-				list.get(0));
+		assertEquals("The item was not passed in to recover method", "bar", list.get(0));
 	}
 
 	public void testRetryPolicy() throws Exception {
@@ -215,7 +216,7 @@ public class ItemOrientedTaskletTests extends TestCase {
 		module.setItemRecoverer(new ItemRecoverer() {
 			public boolean recover(Object item, Throwable cause) {
 				assertEquals("FOO", cause.getMessage());
-				list.add(item+"_recovered");
+				list.add(item + "_recovered");
 				return true;
 			}
 		});
@@ -232,7 +233,8 @@ public class ItemOrientedTaskletTests extends TestCase {
 		try {
 			module.execute();
 			fail("Expected RuntimeException");
-		} catch (RuntimeException e) {
+		}
+		catch (RuntimeException e) {
 			assertEquals("FOO", e.getMessage());
 		}
 
@@ -243,15 +245,15 @@ public class ItemOrientedTaskletTests extends TestCase {
 
 		// verify method calls
 		assertEquals(1, list.size());
-		assertEquals("The item was not passed in to recover method",
-				"foo_recovered", list.get(0));
+		assertEquals("The item was not passed in to recover method", "foo_recovered", list.get(0));
 	}
 
 	public void testInitialisationWithNullProvider() throws Exception {
 		module.setItemReader(null);
 		try {
 			module.afterPropertiesSet();
-		} catch (IllegalArgumentException e) {
+		}
+		catch (IllegalArgumentException e) {
 			assertTrue(e.getMessage().toLowerCase().indexOf("reader") >= 0);
 		}
 	}
@@ -260,8 +262,10 @@ public class ItemOrientedTaskletTests extends TestCase {
 		module.setItemWriter(null);
 		try {
 			module.afterPropertiesSet();
-		} catch (IllegalArgumentException e) {
-			assertTrue("Message did not contain writer: "+e.getMessage(), e.getMessage().toLowerCase().indexOf("writer") >= 0);
+		}
+		catch (IllegalArgumentException e) {
+			assertTrue("Message did not contain writer: " + e.getMessage(), e.getMessage().toLowerCase().indexOf(
+					"writer") >= 0);
 		}
 	}
 
@@ -269,31 +273,34 @@ public class ItemOrientedTaskletTests extends TestCase {
 		public Object read() throws Exception {
 			return "foo";
 		}
+
 		public Object getKey(Object item) {
 			return item;
 		}
 	}
 
-	private class SkippableItemReader implements KeyedItemReader,
-			Skippable, StatisticsProvider {
+	private class SkippableItemReader implements KeyedItemReader, Skippable {
 		public Object read() throws Exception {
 			return itemProvider.read();
 		}
+
 		public Object getKey(Object item) {
 			return item;
 		}
+
 		public void skip() {
 			list.add("provider");
 		}
+
 		public Properties getStatistics() {
 			return PropertiesConverter.stringToProperties("foo=bar");
 		}
-		public void close() throws StreamException {			
+
+		public void close() throws StreamException {
 		}
 	}
 
-	private class SkippableItemWriter implements ItemWriter, Skippable,
-			StatisticsProvider {
+	private class SkippableItemWriter implements ItemWriter, Skippable {
 		String props = "foo=bar";
 
 		public SkippableItemWriter() {