diff --git a/infrastructure/src/main/java/org/springframework/batch/item/provider/AggregateItemProvider.java b/infrastructure/src/main/java/org/springframework/batch/item/provider/AggregateItemProvider.java
new file mode 100644
index 000000000..43a4134bf
--- /dev/null
+++ b/infrastructure/src/main/java/org/springframework/batch/item/provider/AggregateItemProvider.java
@@ -0,0 +1,115 @@
+/*
+ * 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.provider;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.file.FieldSetMapper;
+import org.springframework.batch.item.ItemProvider;
+
+/**
+ * An {@link ItemProvider} that delivers a list as its item, storing up objects
+ * from the injected {@link InputSource} until they are ready to be packed out
+ * as a collection. The {@link InputSource} should mark the beginning and end of
+ * records with the constant values in {@link FieldSetMapper} ({@link FieldSetMapper#BEGIN_RECORD}
+ * and {@link FieldSetMapper#END_RECORD}).<br/>
+ * 
+ * This class is thread safe (it can be used concurrently by multiple threads)
+ * as long as the {@link InputSource} is also thread safe.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class AggregateItemProvider extends AbstractItemProvider {
+
+	private static final Log log = LogFactory
+			.getLog(AggregateItemProvider.class);
+
+	private InputSource inputSource;
+
+	/**
+	 * Get the next list of records.
+	 * 
+	 * @see org.springframework.batch.item.ItemProvider#next()
+	 */
+	public Object next() {
+		ResultHolder holder = new ResultHolder();
+
+		while (process(inputSource.read(), holder)) {
+			continue;
+		}
+
+		if (!holder.exhausted) {
+			return holder.records;
+		} else {
+			return null;
+		}
+	}
+
+	private boolean process(Object value, ResultHolder holder) {
+		// finish processing if we hit the end of file
+		if (value == null) {
+			log.debug("Exhausted InputSource");
+			holder.exhausted = true;
+			return false;
+		}
+
+		// start a new collection
+		if (value == FieldSetMapper.BEGIN_RECORD) {
+			log.debug("Start of new record detected");
+			return true;
+		}
+
+		// mark we are finished with current collection
+		if (value == FieldSetMapper.END_RECORD) {
+			log.debug("End of record detected");
+			return false;
+		}
+
+		// add a simple record to the current collection
+		log.debug("Mapping: " + value);
+		holder.records.add(value);
+		return true;
+	}
+
+	/**
+	 * Injection setter for {@link InputSource}.
+	 * 
+	 * @param inputSource
+	 *            an {@link InputSource}.
+	 */
+	public void setInputSource(InputSource inputSource) {
+		this.inputSource = inputSource;
+	}
+
+	/**
+	 * Private class for temporary state management while item is being
+	 * collected.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 */
+	private static class ResultHolder {
+		Collection records = new ArrayList();
+		boolean exhausted = false;
+	}
+
+}
diff --git a/infrastructure/src/main/java/org/springframework/batch/item/provider/AggregateItemProviderTests.java b/infrastructure/src/main/java/org/springframework/batch/item/provider/AggregateItemProviderTests.java
new file mode 100644
index 000000000..462f14c1d
--- /dev/null
+++ b/infrastructure/src/main/java/org/springframework/batch/item/provider/AggregateItemProviderTests.java
@@ -0,0 +1,60 @@
+package org.springframework.batch.item.provider;
+
+import java.util.Collection;
+import java.util.Iterator;
+
+import junit.framework.TestCase;
+
+import org.easymock.MockControl;
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.file.FieldSetMapper;
+
+public class AggregateItemProviderTests extends TestCase {
+
+	private MockControl inputControl;
+	private InputSource input;
+	private AggregateItemProvider provider;
+
+	public void setUp() {
+
+		//create mock for input
+		inputControl = MockControl.createControl(InputSource.class);
+		input = (InputSource) inputControl.getMock();
+
+		//create provider
+		provider = new AggregateItemProvider();
+		provider.setInputSource(input);
+	}
+
+	public void testNext() {
+
+		//set-up mock input
+		input.read();
+		inputControl.setReturnValue(FieldSetMapper.BEGIN_RECORD);
+		input.read();
+		inputControl.setReturnValue("line",3);
+		input.read();
+		inputControl.setReturnValue(FieldSetMapper.END_RECORD);
+		input.read();
+		inputControl.setReturnValue(null);
+		inputControl.replay();
+
+		//read object
+		Object result = provider.next();
+
+		//it should be collection of 3 strings "line"
+		assertTrue(result instanceof Collection);
+		Collection lines = (Collection)result;
+		assertEquals(3, lines.size());
+
+		for (Iterator i = lines.iterator(); i.hasNext();) {
+			assertEquals("line", i.next());
+		}
+
+		//read object again - it should return null
+		assertNull(provider.next());
+
+		//verify method calls
+		inputControl.verify();
+	}
+}