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

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

Lazy-initialisation solution to the "who's in step scope" issue

spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/HibernateCursorItemReader.java

@@ -186,7 +186,7 @@ public class HibernateCursorItemReader extends AbstractItemReader implements Ite
 	/**
 	 * @return the current row number wrapped as <code>RestartData</code>
 	 */
-	public StreamContext getRestartData() {
+	public StreamContext getStreamContext() {
 		Properties props = new Properties();
 		props.setProperty(RESTART_DATA_ROW_NUMBER_KEY, ""+currentProcessedRow);
 		String skipped = skippedRows.toString();

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));

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 {

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);
 }

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;
 
 /**
- * </p>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 
+ * </p>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) {
 		

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());
 

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);
 	}

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.");
 		

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());
 	}
 

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()));

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);

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));

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();
 

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;
 }

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.
  * </p>
@@ -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;
 }

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();
+
+}

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 {
+
+}

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 {		
 	}
 
 }

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();
 		}

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 {
 		
 	}
 

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/CompositeItemStream.java

@@ -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;
-	}
-
-}

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;
+		}
+	}
+	
+}

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;
+
+}

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);
 		}
 	}

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());

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;
 		}
 

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() {
 	};
 }

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));

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
 		}

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
 		}

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

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));

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();

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();

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();

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();

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"));
 		}
 

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
-	 * <code>getRestartData()</code>
-	 */
-	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 {
-		}
-	}
-
 }

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"));
 		}