Merge pull request 123 from mminella/BATCH-1728

* BATCH-1728:
  BATCH-1728: Added some documentation in the AbstractPaginatedDataItemReader and fixed the message in the RepositoryItemWriter
  BATCH-1728: Initial commit of Spring Data ItemReader/ItemWriter implementations

spring-batch-infrastructure/pom.xml

@@ -39,14 +39,6 @@
 			<groupId>junit</groupId>
 			<artifactId>junit</artifactId>
 		</dependency>
-<!-- 		<dependency> -->
-<!-- 			<groupId>org.easymock</groupId> -->
-<!-- 			<artifactId>easymock</artifactId> -->
-<!-- 		</dependency> -->
-<!--         <dependency> -->
-<!--             <groupId>org.easymock</groupId> -->
-<!--             <artifactId>easymockclassextension</artifactId> -->
-<!--         </dependency> -->
 		<dependency>
 			<groupId>org.aspectj</groupId>
 			<artifactId>aspectjrt</artifactId>
@@ -123,6 +115,11 @@
 			<artifactId>hibernate-annotations</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+		  <groupId>org.hibernate</groupId>
+		  <artifactId>hibernate-validator</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.geronimo.specs</groupId>
 			<artifactId>geronimo-jta_1.1_spec</artifactId>
@@ -191,6 +188,21 @@
 			<artifactId>spring-tx</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			 <groupId>org.springframework.data</groupId>
+			 <artifactId>spring-data-commons-core</artifactId>
+			 <optional>true</optional>
+		</dependency>
+ 		<dependency>
+			<groupId>org.springframework.data</groupId>
+			<artifactId>spring-data-mongodb</artifactId>
+			 <optional>true</optional>
+		</dependency>
+ 		<dependency>
+			<groupId>org.springframework.data</groupId>
+			<artifactId>spring-data-neo4j</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.codehaus.woodstox</groupId>
 			<artifactId>woodstox-core-asl</artifactId>

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/AbstractPaginatedDataItemReader.java

@@ -0,0 +1,98 @@
+package org.springframework.batch.item.data;
+
+import java.util.Iterator;
+
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStreamReader;
+import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
+
+/**
+ * A base class that handles basic reading logic based on the paginated
+ * semantics of Spring Data's paginated facilities.  It also handles the
+ * semantics required for restartability based on those facilities.
+ * 
+ * @author Michael Minella
+ * @since 2.2
+ * @param <T> Type of item to be read
+ */
+public abstract class AbstractPaginatedDataItemReader<T> extends
+AbstractItemCountingItemStreamItemReader<T> {
+
+	protected volatile int page = 0;
+
+	protected int pageSize = 10;
+
+	protected Iterator<T> results;
+
+	private Object lock = new Object();
+
+	/**
+	 * The number of items to be read with each page.
+	 * 
+	 * @param pageSize the number of items
+	 */
+	public void setPageSize(int pageSize) {
+		this.pageSize = pageSize;
+	}
+
+	@Override
+	protected T doRead() throws Exception {
+
+		synchronized (lock) {
+			if(results == null || !results.hasNext()) {
+
+				results = doPageRead();
+
+				page ++;
+
+				if(results == null || !results.hasNext()) {
+					return null;
+				}
+			}
+
+
+			if(results.hasNext()) {
+				return results.next();
+			}
+			else {
+				return null;
+			}
+		}
+	}
+
+	/**
+	 * Method this {@link ItemStreamReader} delegates to
+	 * for the actual work of reading a page.  Each time
+	 * this method is called, the resulting {@link Iterator}
+	 * should contain the items read within the next page.
+	 * <br/><br/>
+	 * If the {@link Iterator} is empty or null when it is
+	 * returned, this {@link ItemReader} will assume that the
+	 * input has been exhausted.
+	 * 
+	 * @return an {@link Iterator} containing the items within a page.
+	 */
+	protected abstract Iterator<T> doPageRead();
+
+	@Override
+	protected void doOpen() throws Exception {
+	}
+
+	@Override
+	protected void doClose() throws Exception {
+	}
+
+	@Override
+	protected void jumpToItem(int itemLastIndex) throws Exception {
+		synchronized (lock) {
+			page = itemLastIndex / pageSize;
+			int current = itemLastIndex % pageSize;
+
+			Iterator<T> initialPage = doPageRead();
+
+			for(; current >= 0; current--) {
+				initialPage.next();
+			}
+		}
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemReader.java

@@ -0,0 +1,221 @@
+/*
+ * Copyright 2012 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.data;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.data.mongodb.core.query.BasicQuery;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.StringUtils;
+
+import com.mongodb.util.JSON;
+
+/**
+ * <p>
+ * Restartable {@link ItemReader} that reads documents from MongoDB
+ * via a paging technique.
+ * </p>
+ *
+ * <p>
+ * It executes the JSON {@link #setQuery(String)} to retrieve the requested
+ * documents.  The query is executed using paged requests specified in the
+ * {@link #setPageSize(int)}.  Additional pages are requested as needed to
+ * provide data when the {@link #read()} method is called.
+ * </p>
+ *
+ * <p>
+ * The JSON query provided supports parameter substitution via ?&lt;index&gt;
+ * placeholders where the &lt;index&gt; indicates the index of the
+ * parameterValue to substitute.
+ * </p>
+ *
+ * <p>
+ * The implementation is thread-safe between calls to
+ * {@link #open(ExecutionContext)}, but remember to use <code>saveState=false</code>
+ * if used in a multi-threaded client (no restart available.
+ * </p>
+ *
+ *
+ * @author Michael Minella
+ */
+public class MongoItemReader<T> extends AbstractPaginatedDataItemReader<T> implements InitializingBean {
+
+	private static final Pattern PLACEHOLDER = Pattern.compile("\\?(\\d+)");
+	private MongoOperations template;
+	private String query;
+	private Class<? extends T> type;
+	private Sort sort;
+	private String hint;
+	private String fields;
+	private List<Object> parameterValues;
+
+	public MongoItemReader() {
+		super();
+		setName(ClassUtils.getShortName(MongoItemReader.class));
+	}
+
+	/**
+	 * Used to perform operations against the MongoDB instance.  Also
+	 * handles the mapping of documents to objects.
+	 *
+	 * @param template the MongoOperations instance to use
+	 * @see MongoOperations
+	 */
+	public void setTemplate(MongoOperations template) {
+		this.template = template;
+	}
+
+	/**
+	 * A JSON formatted MongoDB query.  Parameterization of the provided query is allowed
+	 * via ?&lt;index&gt; placeholders where the &lt;index&gt; indicates the index of the
+	 * parameterValue to substitute.
+	 *
+	 * @param query JSON formatted Mongo query
+	 */
+	public void setQuery(String query) {
+		this.query = query;
+	}
+
+	/**
+	 * The type of object to be returned for each {@link #read()} call.
+	 *
+	 * @param type the type of object to return
+	 */
+	public void setTargetType(Class<? extends T> type) {
+		this.type = type;
+	}
+
+	/**
+	 * {@link List} of values to be substituted in for each of the
+	 * parameters in the query.
+	 *
+	 * @param parameterValues
+	 */
+	public void setParameterValues(List<Object> parameterValues) {
+		this.parameterValues = parameterValues;
+	}
+
+	/**
+	 * JSON defining the fields to be returned from the matching documents
+	 * by MongoDB.
+	 *
+	 * @param fields JSON string that identifies the fields to sorty by.
+	 */
+	public void setFields(String fields) {
+		this.fields = fields;
+	}
+
+	/**
+	 * {@link Map} of property names/{@link org.springframework.data.domain.Sort.Direction} values to
+	 * sort the input by.
+	 *
+	 * @param sorts map of properties and direction to sort each.
+	 */
+	public void setSort(Map<String, Sort.Direction> sorts) {
+		this.sort = convertToSort(sorts);
+	}
+
+	/**
+	 * JSON String telling MongoDB what index to use.
+	 *
+	 * @param hint string indicating what index to use.
+	 */
+	public void setHint(String hint) {
+		this.hint = hint;
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	protected Iterator<T> doPageRead() {
+
+		Pageable pageRequest = new PageRequest(page, pageSize, sort);
+
+		String populatedQuery = replacePlaceholders(query, parameterValues);
+
+		Query mongoQuery = null;
+
+		if(StringUtils.hasText(fields)) {
+			mongoQuery = new BasicQuery(populatedQuery, fields);
+		}
+		else {
+			mongoQuery = new BasicQuery(populatedQuery);
+		}
+
+		mongoQuery.with(pageRequest);
+
+		if(StringUtils.hasText(hint)) {
+			mongoQuery.withHint(hint);
+		}
+
+		return (Iterator<T>) template.find(mongoQuery, type).iterator();
+	}
+
+	/**
+	 * Checks mandatory properties
+	 *
+	 * @see InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.state(template != null, "An implementation of MongoOperations is required.");
+		Assert.state(type != null, "A type to convert the input into is required.");
+		Assert.state(query != null, "A query is required.");
+		Assert.state(sort != null, "A sort is required.");
+	}
+
+	// Copied from StringBasedMongoQuery...is there a place where this type of logic is already exposed?
+	private String replacePlaceholders(String input, List<Object> values) {
+		Matcher matcher = PLACEHOLDER.matcher(input);
+		String result = input;
+
+		while (matcher.find()) {
+			String group = matcher.group();
+			int index = Integer.parseInt(matcher.group(1));
+			result = result.replace(group, getParameterWithIndex(values, index));
+		}
+
+		return result;
+	}
+
+	// Copied from StringBasedMongoQuery...is there a place where this type of logic is already exposed?
+	private String getParameterWithIndex(List<Object> values, int index) {
+		return JSON.serialize(values.get(index));
+	}
+
+	private Sort convertToSort(Map<String, Sort.Direction> sorts) {
+		List<Sort.Order> sortValues = new ArrayList<Sort.Order>();
+
+		for (Map.Entry<String, Sort.Direction> curSort : sorts.entrySet()) {
+			sortValues.add(new Sort.Order(curSort.getValue(), curSort.getKey()));
+		}
+
+		return new Sort(sortValues);
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemWriter.java

@@ -0,0 +1,175 @@
+/*
+ * Copyright 2012 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.data;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * <p>
+ * A {@link ItemWriter} implementation that writes to a MongoDB store using an implementation of Spring Data's
+ * {@link MongoOperations}.  Since MongoDB is not a transactional store, a best effort is made to persist
+ * written data at the last moment, yet still honor job status contracts.  No attempt to roll back is made
+ * if an error occurs during writing.
+ * </p>
+ *
+ * <p>
+ * This writer is thread safe once all properties are set (normal singleton behavior) so it can be used in multiple
+ * concurrent transactions.
+ * </p>
+ *
+ * @author Michael Minella
+ *
+ */
+public class MongoItemWriter<T> implements ItemWriter<T>, InitializingBean {
+
+	private static final String BUFFER_KEY_PREFIX = MongoItemWriter.class.getName() + ".BUFFER_KEY";
+	private MongoOperations template;
+	private final String bufferKey;
+	private String collection;
+	private boolean delete = false;
+
+	public MongoItemWriter() {
+		super();
+		this.bufferKey = BUFFER_KEY_PREFIX + "." + hashCode();
+	}
+
+	/**
+	 * Indicates if the items being passed to the writer are to be saved or
+	 * removed from the data store.  If set to false (default), the items will
+	 * be saved.  If set to true, the items will be removed.
+	 *
+	 * @param delete removal indicator
+	 */
+	public void setDelete(boolean delete) {
+		this.delete = delete;
+	}
+
+	/**
+	 * Set the {@link MongoOperations} to be used to save items to be written.
+	 *
+	 * @param template the template implementation to be used.
+	 */
+	public void setTemplate(MongoOperations template) {
+		this.template = template;
+	}
+
+	/**
+	 * Set the name of the Mongo collection to be written to.
+	 *
+	 * @param collection the name of the collection.
+	 */
+	public void setCollection(String collection) {
+		this.collection = collection;
+	}
+
+	/**
+	 * If a transaction is active, buffer items to be written just before commit.
+	 * Otherwise write items using the provided template.
+	 *
+	 * @see org.springframework.batch.item.ItemWriter#write(List)
+	 */
+	public void write(List<? extends T> items) throws Exception {
+		if(!transactionActive()) {
+			doWrite(items);
+			return;
+		}
+
+		List bufferedItems = getCurrentBuffer();
+		bufferedItems.addAll(items);
+	}
+
+	/**
+	 * Performs the actual write to the store via the template.
+	 * This can be overridden by a subclass if necessary.
+	 *
+	 * @param items the list of items to be persisted.
+	 */
+	protected void doWrite(List<? extends T> items) {
+		if(! CollectionUtils.isEmpty(items)) {
+			if(delete) {
+				if(StringUtils.hasText(collection)) {
+					for (Object object : items) {
+						template.remove(object, collection);
+					}
+				}
+				else {
+					for (Object object : items) {
+						template.remove(object);
+					}
+				}
+			}
+			else {
+				if(StringUtils.hasText(collection)) {
+					for (Object object : items) {
+						template.save(object, collection);
+					}
+				}
+				else {
+					for (Object object : items) {
+						template.save(object);
+					}
+				}
+			}
+		}
+	}
+
+	private boolean transactionActive() {
+		return TransactionSynchronizationManager.isActualTransactionActive();
+	}
+
+	private List<? extends T> getCurrentBuffer() {
+		if(!TransactionSynchronizationManager.hasResource(bufferKey)) {
+			TransactionSynchronizationManager.bindResource(bufferKey, new ArrayList());
+
+			TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
+				@Override
+				public void beforeCommit(boolean readOnly) {
+					List items = (List) TransactionSynchronizationManager.getResource(bufferKey);
+
+					if(!CollectionUtils.isEmpty(items)) {
+						if(!readOnly) {
+							doWrite(items);
+						}
+					}
+				}
+
+				@Override
+				public void afterCompletion(int status) {
+					if(TransactionSynchronizationManager.hasResource(bufferKey)) {
+						TransactionSynchronizationManager.unbindResource(bufferKey);
+					}
+				}
+			});
+		}
+
+		return (List) TransactionSynchronizationManager.getResource(bufferKey);
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.state(template != null, "A MongoOperations implementation is required.");
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/Neo4jItemReader.java

@@ -0,0 +1,226 @@
+/*
+ * Copyright 2012 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.data;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.data.neo4j.conversion.DefaultConverter;
+import org.springframework.data.neo4j.conversion.Result;
+import org.springframework.data.neo4j.conversion.ResultConverter;
+import org.springframework.data.neo4j.template.Neo4jOperations;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * <p>
+ * Restartable {@link ItemReader} that reads objects from the graph database Neo4j
+ * via a paging technique.
+ * </p>
+ *
+ * <p>
+ * It executes cypher queries built from the statement fragments provided to
+ * retrieve the requested data.  The query is executed using paged requests of
+ * a size specified in {@link #setPageSize(int)}.  Additional pages are requested
+ * as needed when the {@link #read()} method is called.  On restart, the reader
+ * will begin again at the same number item it left off at.
+ * </p>
+ *
+ * <p>
+ * Performance is dependent on your Neo4J configuration (embedded or remote) as
+ * well as page size.  Setting a fairly large page size and using a commit
+ * interval that matches the page size should provide better performance.
+ * </p>
+ *
+ * <p>
+ * This implementation is thread-safe between calls to
+ * {@link #open(org.springframework.batch.item.ExecutionContext)}, however you
+ * should set <code>saveState=false</code> if used in a multi-threaded
+ * environment (no restart available).
+ * </p>
+ *
+ * @author Michael Minella
+ *
+ */
+public class Neo4jItemReader<T> extends AbstractPaginatedDataItemReader<T> implements
+InitializingBean {
+
+	protected Log logger = LogFactory.getLog(getClass());
+
+	private Neo4jOperations template;
+
+	private String startStatement;
+	private String returnStatement;
+	private String matchStatement;
+	private String whereStatement;
+	private String orderByStatement;
+
+	private Class targetType;
+
+	private Map<String, Object> parameterValues;
+
+	private ResultConverter resultConverter;
+
+	public Neo4jItemReader() {
+		setName(ClassUtils.getShortName(Neo4jItemReader.class));
+	}
+
+	/**
+	 * The start segment of the cypher query.  START is prepended
+	 * to the statement provided and should <em>not</em> be
+	 * included.
+	 *
+	 * @param startStatement the start fragment of the cypher query.
+	 */
+	public void setStartStatement(String startStatement) {
+		this.startStatement = startStatement;
+	}
+
+	/**
+	 * The return statement of the cypher query.  RETURN is prepended
+	 * to the statement provided and should <em>not</em> be
+	 * included
+	 *
+	 * @param returnStatement the return fragment of the cypher query.
+	 */
+	public void setReturnStatement(String returnStatement) {
+		this.returnStatement = returnStatement;
+	}
+
+	/**
+	 * An optional match fragment of the cypher query.  MATCH is
+	 * prepended to the statement provided and should <em>not</em>
+	 * be included.
+	 *
+	 * @param matchStatement the match fragment of the cypher query
+	 */
+	public void setMatchStatement(String matchStatement) {
+		this.matchStatement = matchStatement;
+	}
+
+	/**
+	 * An optional where fragement of the cypher query.  WHERE is
+	 * prepended to the statement provided and should <em>not</em>
+	 * be included.
+	 *
+	 * @param whereStatement where fragment of the cypher query
+	 */
+	public void setWhereStatement(String whereStatement) {
+		this.whereStatement = whereStatement;
+	}
+
+	/**
+	 * A list of properties to order the results by.  This is
+	 * required so that subsequent page requests pull back the
+	 * segment of results correctly.  ORDER BY is prepended to
+	 * the statement provided and should <em>not</em> be included.
+	 *
+	 * @param orderByStatement order by fragment of the cypher query.
+	 */
+	public void setOrderByStatement(String orderByStatement) {
+		this.orderByStatement = orderByStatement;
+	}
+
+	/**
+	 * Used to perform operations against the Neo4J database.
+	 *
+	 * @param template the Neo4jOperations instance to use
+	 * @see Neo4jOperations
+	 */
+	public void setTemplate(Neo4jOperations template) {
+		this.template = template;
+	}
+
+	/**
+	 * The object type to be returned from each call to {@link #read()}
+	 *
+	 * @param targetType the type of object to return.
+	 */
+	public void setTargetType(Class targetType) {
+		this.targetType = targetType;
+	}
+
+	/**
+	 * Set the converter used to convert node to the targetType.  By
+	 * default, {@link DefaultConverter} is used.
+	 *
+	 * @param resultConverter the converter to use.
+	 */
+	public void setResultConverter(ResultConverter resultConverter) {
+		this.resultConverter = resultConverter;
+	}
+
+	@Override
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	protected Iterator<T> doPageRead() {
+		Result<Map<String, Object>> queryResults = template.query(
+				generateLimitCypherQuery(), parameterValues);
+
+		if(queryResults != null) {
+			if (resultConverter != null) {
+				return queryResults.to(targetType, resultConverter).iterator();
+			}
+			else {
+				return queryResults.to(targetType).iterator();
+			}
+		}
+		else {
+			return new ArrayList().iterator();
+		}
+	}
+
+	private String generateLimitCypherQuery() {
+		StringBuilder query = new StringBuilder();
+
+		query.append("START ").append(startStatement);
+		query.append(matchStatement != null ? " MATCH " + matchStatement : "");
+		query.append(whereStatement != null ? " WHERE " + whereStatement : "");
+		query.append(" RETURN ").append(returnStatement);
+		query.append(" ORDER BY ").append(orderByStatement);
+		query.append(" SKIP " + (pageSize * page));
+		query.append(" LIMIT " + pageSize);
+
+		String resultingQuery = query.toString();
+		System.out.println("resulting query = " + resultingQuery);
+
+		if (logger.isDebugEnabled()) {
+			logger.debug(resultingQuery);
+		}
+
+		return resultingQuery;
+	}
+
+	/**
+	 * Checks mandatory properties
+	 *
+	 * @see InitializingBean#afterPropertiesSet()
+	 */
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.state(template != null, "A Neo4JOperations implementation is required");
+		Assert.state(targetType != null, "The type to be returned is required");
+		Assert.state(StringUtils.hasText(startStatement), "A START statement is required");
+		Assert.state(StringUtils.hasText(returnStatement), "A RETURN statement is required");
+		Assert.state(StringUtils.hasText(orderByStatement), "A ORDER BY statement is required");
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/Neo4jItemWriter.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2012 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.data;
+
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.data.neo4j.template.Neo4jOperations;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * <p>
+ * A {@link ItemWriter} implementation that writes to a Neo4j database using an
+ * implementation of Spring Data's {@link Neo4jOperations}.
+ * </p>
+ *
+ * <p>
+ * This writer is thread safe once all properties are set (normal singleton
+ * behavior) so it can be used in multiple concurrent transactions.
+ * </p>
+ *
+ * @author Michael Minella
+ *
+ */
+public class Neo4jItemWriter<T> implements ItemWriter<T>, InitializingBean {
+
+	protected static final Log logger = LogFactory
+			.getLog(Neo4jItemWriter.class);
+
+	private boolean delete = false;
+
+	private Neo4jOperations template;
+
+	public void setDelete(boolean delete) {
+		this.delete = delete;
+	}
+
+	/**
+	 * Set the {@link Neo4jOperations} to be used to save items
+	 *
+	 * @param template the template implementation to be used
+	 */
+	public void setTemplate(Neo4jOperations template) {
+		this.template = template;
+	}
+
+	/**
+	 * Checks mandatory properties
+	 *
+	 * @see InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.state(template != null, "A Neo4JOperations implementation is required");
+	}
+
+	/**
+	 * Write all items to the data store.
+	 *
+	 * @see org.springframework.batch.item.ItemWriter#write(java.util.List)
+	 */
+	public void write(List<? extends T> items) throws Exception {
+		if(!CollectionUtils.isEmpty(items)) {
+			doWrite(items);
+		}
+	}
+
+	/**
+	 * Performs the actual write using the template.  This can be overriden by
+	 * a subclass if necessary.
+	 *
+	 * @param items the list of items to be persisted.
+	 */
+	protected void doWrite(List<? extends T> items) {
+		if(delete) {
+			for (T t : items) {
+				template.delete(t);
+			}
+		}
+		else {
+			for (T t : items) {
+				template.save(t);
+			}
+		}
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemReader.java

@@ -0,0 +1,263 @@
+/*
+ * Copyright 2012 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.data;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.adapter.AbstractMethodInvokingDelegator.InvocationTargetThrowableWrapper;
+import org.springframework.batch.item.adapter.DynamicMethodInvocationException;
+import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.repository.PagingAndSortingRepository;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.MethodInvoker;
+
+/**
+ * <p>
+ * A {@link org.springframework.batch.item.ItemReader} that reads records utilizing
+ * a {@link org.springframework.data.repository.PagingAndSortingRepository}.
+ * </p>
+ *
+ * <p>
+ * Performance of the reader is dependent on the repository implementation, however
+ * setting a reasonably large page size and matching that to the commit interval should
+ * yield better performance.
+ * </p>
+ *
+ * <p>
+ * The reader must be configured with a {@link org.springframework.data.repository.PagingAndSortingRepository},
+ * a {@link org.springframework.data.domain.Sort}, and a pageSize greater than 0.
+ * </p>
+ *
+ * <p>
+ * This implementation is thread safe between calls to {@link #open(ExecutionContext)}, but remember to use
+ * <code>saveState=false</code> if used in a multi-threaded client (no restart available).
+ * </p>
+ *
+ * @author Michael Minella
+ * @since 2.2
+ */
+@SuppressWarnings("rawtypes")
+public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements InitializingBean {
+
+	protected Log logger = LogFactory.getLog(getClass());
+
+	private PagingAndSortingRepository repository;
+
+	private Sort sort;
+
+	private volatile int page = 0;
+
+	private int pageSize = 10;
+
+	private volatile int current = 0;
+
+	private List arguments;
+
+	private volatile List<T> results;
+
+	private Object lock = new Object();
+
+	private String methodName;
+
+	public RepositoryItemReader() {
+		setName(ClassUtils.getShortName(RepositoryItemReader.class));
+	}
+
+	/**
+	 * Arguments to be passed to the data providing method.
+	 *
+	 * @param arguments list of method arguments to be passed to the repository
+	 */
+	public void setArguments(List arguments) {
+		this.arguments = arguments;
+	}
+
+	/**
+	 * Provides ordering of the results so that order is maintained between paged queries
+	 *
+	 * @param sorts the fields to sort by and the directions
+	 */
+	public void setSort(Map<String, Sort.Direction> sorts) {
+		this.sort = convertToSort(sorts);
+	}
+
+	/**
+	 * @param pageSize The number of items to retrieve per page.
+	 */
+	public void setPageSize(int pageSize) {
+		this.pageSize = pageSize;
+	}
+
+	/**
+	 * The {@link org.springframework.data.repository.PagingAndSortingRepository}
+	 * implementation used to read input from.
+	 *
+	 * @param repository underlying repository for input to be read from.
+	 */
+	public void setRepository(PagingAndSortingRepository repository) {
+		this.repository = repository;
+	}
+
+	/**
+	 * Specifies what method on the repository to call.  This method must take
+	 * {@link org.springframework.data.domain.Pageable} as the <em>last</em> argument.
+	 *
+	 * @param methodName
+	 */
+	public void setMethodName(String methodName) {
+		this.methodName = methodName;
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.state(repository != null, "A PagingAndSortingRepository is required");
+		Assert.state(pageSize > 0, "Page size must be greater than 0");
+		Assert.state(sort != null, "A sort is required");
+	}
+
+	@Override
+	protected T doRead() throws Exception {
+
+		synchronized (lock) {
+			if(results == null || current >= results.size()) {
+
+				if (logger.isDebugEnabled()) {
+					logger.debug("Reading page " + page);
+				}
+
+				results = doPageRead();
+
+				current = 0;
+				page ++;
+
+				if(results.size() <= 0) {
+					return null;
+				}
+			}
+
+			if(current < results.size()) {
+				T curLine = results.get(current);
+				current++;
+				return curLine;
+			}
+			else {
+				return null;
+			}
+		}
+	}
+
+	@Override
+	protected void jumpToItem(int itemLastIndex) throws Exception {
+		synchronized (lock) {
+			page = itemLastIndex / pageSize;
+			current = itemLastIndex % pageSize;
+
+			results = doPageRead();
+		}
+	}
+
+	/**
+	 * Performs the actual reading of a page via the repository.
+	 * Available for overriding as needed.
+	 *
+	 * @return the list of items that make up the page
+	 * @throws Exception
+	 */
+	@SuppressWarnings("unchecked")
+	protected List<T> doPageRead() throws Exception {
+		Pageable pageRequest = new PageRequest(page, pageSize, sort);
+
+		MethodInvoker invoker = createMethodInvoker(repository, methodName);
+
+		List parameters = new ArrayList();
+
+		if(arguments != null && arguments.size() > 0) {
+			parameters.addAll(arguments);
+		}
+
+		parameters.add(pageRequest);
+
+		invoker.setArguments(parameters.toArray());
+
+		Page curPage = (Page) doInvoke(invoker);
+
+		return curPage.getContent();
+	}
+
+	@Override
+	protected void doOpen() throws Exception {
+	}
+
+	@Override
+	protected void doClose() throws Exception {
+	}
+
+	private Sort convertToSort(Map<String, Sort.Direction> sorts) {
+		List<Sort.Order> sortValues = new ArrayList<Sort.Order>();
+
+		for (Map.Entry<String, Sort.Direction> curSort : sorts.entrySet()) {
+			sortValues.add(new Sort.Order(curSort.getValue(), curSort.getKey()));
+		}
+
+		return new Sort(sortValues);
+	}
+
+	private Object doInvoke(MethodInvoker invoker) throws Exception{
+		try {
+			invoker.prepare();
+		}
+		catch (ClassNotFoundException e) {
+			throw new DynamicMethodInvocationException(e);
+		}
+		catch (NoSuchMethodException e) {
+			throw new DynamicMethodInvocationException(e);
+		}
+
+		try {
+			return invoker.invoke();
+		}
+		catch (InvocationTargetException e) {
+			if (e.getCause() instanceof Exception) {
+				throw (Exception) e.getCause();
+			}
+			else {
+				throw new InvocationTargetThrowableWrapper(e.getCause());
+			}
+		}
+		catch (IllegalAccessException e) {
+			throw new DynamicMethodInvocationException(e);
+		}
+	}
+
+	private MethodInvoker createMethodInvoker(Object targetObject, String targetMethod) {
+		MethodInvoker invoker = new MethodInvoker();
+		invoker.setTargetObject(targetObject);
+		invoker.setTargetMethod(targetMethod);
+		return invoker;
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemWriter.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright 2012 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.data;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.adapter.AbstractMethodInvokingDelegator.InvocationTargetThrowableWrapper;
+import org.springframework.batch.item.adapter.DynamicMethodInvocationException;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.data.repository.CrudRepository;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.MethodInvoker;
+
+/**
+ * <p>
+ * A {@link org.springframework.batch.item.ItemReader} wrapper for a
+ * {@link org.springframework.data.repository.CrudRepository} from Spring Data.
+ * </p>
+ *
+ * <p>
+ * It depends on {@link org.springframework.data.repository.CrudRepository#save(Iterable)}
+ * method to store the items for the chunk.  Performance will be determined by that
+ * implementation more than this writer.
+ * </p>
+ *
+ * <p>
+ * As long as the repository provided is thread-safe, this writer is also thread-safe once
+ * properties are set (normal singleton behavior), so it can be used in multiple concurrent
+ * transactions.
+ * </p>
+ *
+ * @author Michael Minella
+ * @since 2.2
+ */
+@SuppressWarnings("rawtypes")
+public class RepositoryItemWriter implements ItemWriter, InitializingBean {
+
+	protected static final Log logger = LogFactory.getLog(RepositoryItemWriter.class);
+
+	private CrudRepository repository;
+
+	private String methodName;
+
+	/**
+	 * Specifies what method on the repository to call.  This method must the type of
+	 * object passed to this writer as the <em>sole</em> argument.
+	 *
+	 * @param methodName
+	 */
+	public void setMethodName(String methodName) {
+		this.methodName = methodName;
+	}
+
+	/**
+	 * Set the {@link org.springframework.data.repository.CrudRepository} implementation
+	 * for persistence
+	 *
+	 * @param repository the Spring Data repository to be set
+	 */
+	public void setRepository(CrudRepository repository) {
+		this.repository = repository;
+	}
+
+	/**
+	 * Write all items to the data store via a Spring Data repository.
+	 *
+	 * @see org.springframework.batch.item.ItemWriter#write(java.util.List)
+	 */
+	@Override
+	public void write(List items) throws Exception {
+		if(!CollectionUtils.isEmpty(items)) {
+			doWrite(items);
+		}
+	}
+
+	/**
+	 * Performs the actual write to the repository.  This can be overriden by
+	 * a subclass if necessary.
+	 *
+	 * @param items the list of items to be persisted.
+	 */
+	protected void doWrite(List items) throws Exception {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Writing to the repository with " + items.size() + " items.");
+		}
+
+		MethodInvoker invoker = createMethodInvoker(repository, methodName);
+
+		for (Object object : items) {
+			invoker.setArguments(new Object [] {object});
+			doInvoke(invoker);
+		}
+	}
+
+	/**
+	 * Check mandatory properties - there must be a repository.
+	 */
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.state(repository != null, "A CrudRepository implementation is required");
+	}
+
+
+	private Object doInvoke(MethodInvoker invoker) throws Exception{
+		try {
+			invoker.prepare();
+		}
+		catch (ClassNotFoundException e) {
+			throw new DynamicMethodInvocationException(e);
+		}
+		catch (NoSuchMethodException e) {
+			throw new DynamicMethodInvocationException(e);
+		}
+
+		try {
+			return invoker.invoke();
+		}
+		catch (InvocationTargetException e) {
+			if (e.getCause() instanceof Exception) {
+				throw (Exception) e.getCause();
+			}
+			else {
+				throw new InvocationTargetThrowableWrapper(e.getCause());
+			}
+		}
+		catch (IllegalAccessException e) {
+			throw new DynamicMethodInvocationException(e);
+		}
+	}
+
+	private MethodInvoker createMethodInvoker(Object targetObject, String targetMethod) {
+		MethodInvoker invoker = new MethodInvoker();
+		invoker.setTargetObject(targetObject);
+		invoker.setTargetMethod(targetMethod);
+		return invoker;
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemReaderTests.java

@@ -0,0 +1,183 @@
+package org.springframework.batch.item.data;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.data.mongodb.core.query.Query;
+
+public class MongoItemReaderTests {
+
+	private MongoItemReader reader;
+	@Mock
+	private MongoOperations template;
+	private Map<String, Sort.Direction> sortOptions;
+
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+		reader = new MongoItemReader();
+
+		sortOptions = new HashMap<String, Sort.Direction>();
+		sortOptions.put("name", Sort.Direction.DESC);
+
+		reader.setTemplate(template);
+		reader.setTargetType(String.class);
+		reader.setQuery("{ }");
+		reader.setSort(sortOptions);
+		reader.afterPropertiesSet();
+		reader.setPageSize(50);
+	}
+
+	@Test
+	public void testAfterPropertiesSet() throws Exception{
+		reader = new MongoItemReader();
+
+		try {
+			reader.afterPropertiesSet();
+			fail("Template was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("An implementation of MongoOperations is required.", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown.");
+		}
+
+		reader.setTemplate(template);
+
+		try {
+			reader.afterPropertiesSet();
+			fail("type was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("A type to convert the input into is required.", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown.");
+		}
+
+		reader.setTargetType(String.class);
+
+		try {
+			reader.afterPropertiesSet();
+			fail("Query was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("A query is required.", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown.");
+		}
+
+		reader.setQuery("");
+
+		try {
+			reader.afterPropertiesSet();
+			fail("Sort was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("A sort is required.", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown.");
+		}
+
+		reader.setSort(sortOptions);
+
+		reader.afterPropertiesSet();
+	}
+
+	@Test
+	public void testBasicQueryFirstPage() {
+		ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
+
+		when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList<String>());
+
+		assertFalse(reader.doPageRead().hasNext());
+
+		Query query = queryContainer.getValue();
+		assertEquals(50, query.getLimit());
+		assertEquals(0, query.getSkip());
+		assertEquals("{ }", query.getQueryObject().toString());
+		assertEquals("{ \"name\" : -1}", query.getSortObject().toString());
+	}
+
+	@Test
+	public void testBasicQuerySecondPage() {
+		reader.page = 2;
+		ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
+
+		when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList<String>());
+
+		assertFalse(reader.doPageRead().hasNext());
+
+		Query query = queryContainer.getValue();
+
+		assertEquals(50, query.getLimit());
+		assertEquals(100, query.getSkip());
+		assertEquals("{ }", query.getQueryObject().toString());
+		assertEquals("{ \"name\" : -1}", query.getSortObject().toString());
+		assertNull(query.getFieldsObject());
+	}
+
+	@Test
+	public void testQueryWithFields() {
+		reader.setFields("{name : 1, age : 1, _id: 0}");
+		ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
+
+		when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList<String>());
+
+		assertFalse(reader.doPageRead().hasNext());
+
+		Query query = queryContainer.getValue();
+		assertEquals(50, query.getLimit());
+		assertEquals(0, query.getSkip());
+		assertEquals("{ }", query.getQueryObject().toString());
+		assertEquals("{ \"name\" : -1}", query.getSortObject().toString());
+		assertEquals("{ \"name\" : 1 , \"age\" : 1 , \"_id\" : 0}", query.getFieldsObject().toString());
+	}
+
+	@Test
+	public void testQueryWithHint() {
+		reader.setHint("{ $natural : 1}");
+		ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
+
+		when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList<String>());
+
+		assertFalse(reader.doPageRead().hasNext());
+
+		Query query = queryContainer.getValue();
+		assertEquals(50, query.getLimit());
+		assertEquals(0, query.getSkip());
+		assertEquals("{ }", query.getQueryObject().toString());
+		assertEquals("{ \"name\" : -1}", query.getSortObject().toString());
+		assertEquals("{ $natural : 1}", query.getHint());
+	}
+
+	@Test
+	public void testQueryWithParameters() {
+		reader.setParameterValues(new ArrayList<String>(){{
+			add("foo");
+		}});
+
+		reader.setQuery("{ name : ?0 }");
+		ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
+
+		when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList<String>());
+
+		assertFalse(reader.doPageRead().hasNext());
+
+		Query query = queryContainer.getValue();
+		assertEquals(50, query.getLimit());
+		assertEquals(0, query.getSkip());
+		assertEquals("{ \"name\" : \"foo\"}", query.getQueryObject().toString());
+		assertEquals("{ \"name\" : -1}", query.getSortObject().toString());
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemWriterTests.java

@@ -0,0 +1,235 @@
+package org.springframework.batch.item.data;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
+import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@SuppressWarnings({"rawtypes", "serial", "unchecked"})
+public class MongoItemWriterTests {
+
+	private MongoItemWriter writer;
+	@Mock
+	private MongoOperations template;
+	private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
+
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+		writer = new MongoItemWriter();
+		writer.setTemplate(template);
+		writer.afterPropertiesSet();
+	}
+
+	@Test
+	public void testAfterPropertiesSet() throws Exception {
+		writer = new MongoItemWriter();
+
+		try {
+			writer.afterPropertiesSet();
+			fail("Expected exception was not thrown");
+		} catch (IllegalStateException iae) {
+		}
+
+		writer.setTemplate(template);
+		writer.afterPropertiesSet();
+	}
+
+	@Test
+	public void testWriteNoTransactionNoCollection() throws Exception {
+		List<Object> items = new ArrayList<Object>() {{
+			add(new Object());
+			add(new Object());
+		}};
+
+		writer.write(items);
+
+		verify(template).save(items.get(0));
+		verify(template).save(items.get(1));
+	}
+
+	@Test
+	public void testWriteNoTransactionWithCollection() throws Exception {
+		List<Object> items = new ArrayList<Object>() {{
+			add(new Object());
+			add(new Object());
+		}};
+
+		writer.setCollection("collection");
+
+		writer.write(items);
+
+		verify(template).save(items.get(0), "collection");
+		verify(template).save(items.get(1), "collection");
+	}
+
+	@Test
+	public void testWriteNoTransactionNoItems() throws Exception {
+		writer.write(null);
+
+		verifyZeroInteractions(template);
+	}
+
+	@Test
+	public void testWriteTransactionNoCollection() throws Exception {
+		final List<Object> items = new ArrayList<Object>() {{
+			add(new Object());
+			add(new Object());
+		}};
+
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+
+			@Override
+			public Object doInTransaction(TransactionStatus status) {
+				try {
+					writer.write(items);
+				} catch (Exception e) {
+					fail("An exception was thrown while writing: " + e.getMessage());
+				}
+
+				return null;
+			}
+		});
+
+		verify(template).save(items.get(0));
+		verify(template).save(items.get(1));
+	}
+
+	@Test
+	public void testWriteTransactionWithCollection() throws Exception {
+		final List<Object> items = new ArrayList<Object>() {{
+			add(new Object());
+			add(new Object());
+		}};
+
+		writer.setCollection("collection");
+
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+
+			@Override
+			public Object doInTransaction(TransactionStatus status) {
+				try {
+					writer.write(items);
+				} catch (Exception e) {
+					fail("An exception was thrown while writing: " + e.getMessage());
+				}
+
+				return null;
+			}
+		});
+
+		verify(template).save(items.get(0), "collection");
+		verify(template).save(items.get(1), "collection");
+	}
+
+	@Test
+	public void testWriteTransactionFails() throws Exception {
+		final List<Object> items = new ArrayList<Object>() {{
+			add(new Object());
+			add(new Object());
+		}};
+
+		writer.setCollection("collection");
+
+		try {
+			new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+
+				@Override
+				public Object doInTransaction(TransactionStatus status) {
+					try {
+						writer.write(items);
+					} catch (Exception ignore) {
+						fail("unexpected exception thrown");
+					}
+					throw new RuntimeException("force rollback");
+				}
+			});
+		} catch (RuntimeException re) {
+			assertEquals(re.getMessage(), "force rollback");
+		} catch (Throwable t) {
+			fail("Unexpected exception was thrown");
+		}
+
+		verifyZeroInteractions(template);
+	}
+
+	/**
+	 * A pointless use case but validates that the flag is still honored.
+	 *
+	 * @throws Exception
+	 */
+	@Test
+	public void testWriteTransactionReadOnly() throws Exception {
+		final List<Object> items = new ArrayList<Object>() {{
+			add(new Object());
+			add(new Object());
+		}};
+
+		writer.setCollection("collection");
+
+		try {
+			TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
+			transactionTemplate.setReadOnly(true);
+			transactionTemplate.execute(new TransactionCallback() {
+
+				@Override
+				public Object doInTransaction(TransactionStatus status) {
+					try {
+						writer.write(items);
+					} catch (Exception ignore) {
+						fail("unexpected exception thrown");
+					}
+					return null;
+				}
+			});
+		} catch (Throwable t) {
+			fail("Unexpected exception was thrown");
+		}
+
+		verifyZeroInteractions(template);
+	}
+
+	@Test
+	public void testRemoveNoTransactionNoCollection() throws Exception {
+		writer.setDelete(true);
+		List<Object> items = new ArrayList<Object>() {{
+			add(new Object());
+			add(new Object());
+		}};
+
+		writer.write(items);
+
+		verify(template).remove(items.get(0));
+		verify(template).remove(items.get(1));
+	}
+
+	@Test
+	public void testRemoveNoTransactionWithCollection() throws Exception {
+		writer.setDelete(true);
+		List<Object> items = new ArrayList<Object>() {{
+			add(new Object());
+			add(new Object());
+		}};
+
+		writer.setCollection("collection");
+
+		writer.write(items);
+
+		verify(template).remove(items.get(0), "collection");
+		verify(template).remove(items.get(1), "collection");
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/Neo4jItemReaderTests.java

@@ -0,0 +1,164 @@
+package org.springframework.batch.item.data;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.isNull;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.data.neo4j.conversion.DefaultConverter;
+import org.springframework.data.neo4j.conversion.EndResult;
+import org.springframework.data.neo4j.conversion.Result;
+import org.springframework.data.neo4j.conversion.ResultConverter;
+import org.springframework.data.neo4j.template.Neo4jOperations;
+
+public class Neo4jItemReaderTests {
+
+	private Neo4jItemReader reader;
+	@Mock
+	private Neo4jOperations template;
+	@Mock
+	private Result result;
+	@Mock
+	private EndResult endResult;
+
+	@Before
+	public void setUp() throws Exception {
+		reader = new Neo4jItemReader();
+
+		MockitoAnnotations.initMocks(this);
+
+		reader.setTemplate(template);
+		reader.setTargetType(String.class);
+		reader.setStartStatement("n=node(*)");
+		reader.setReturnStatement("*");
+		reader.setOrderByStatement("n.age");
+		reader.setPageSize(50);
+		reader.afterPropertiesSet();
+	}
+
+	@Test
+	public void testAfterPropertiesSet() throws Exception {
+		reader = new Neo4jItemReader();
+
+		try {
+			reader.afterPropertiesSet();
+			fail("Template was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("A Neo4JOperations implementation is required", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown:" + t);
+		}
+
+		reader.setTemplate(template);
+
+		try {
+			reader.afterPropertiesSet();
+			fail("type was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("The type to be returned is required", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown:" + t);
+		}
+
+		reader.setTargetType(String.class);
+
+		try {
+			reader.afterPropertiesSet();
+			fail("START was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("A START statement is required", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown:" + t);
+		}
+
+		reader.setStartStatement("n=node(*)");
+
+		try {
+			reader.afterPropertiesSet();
+			fail("RETURN was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("A RETURN statement is required", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown:" + t);
+		}
+
+		reader.setReturnStatement("n.name, n.phone");
+
+		try {
+			reader.afterPropertiesSet();
+			fail("ORDER BY was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("A ORDER BY statement is required", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown:" + t);
+		}
+
+		reader.setOrderByStatement("n.age");
+
+		reader.afterPropertiesSet();
+	}
+
+	@Test
+	public void testNullResults() {
+		ArgumentCaptor<String> query = ArgumentCaptor.forClass(String.class);
+
+		when(template.query(query.capture(), (Map<String, Object>) isNull())).thenReturn(null);
+
+		assertFalse(reader.doPageRead().hasNext());
+		assertEquals("START n=node(*) RETURN * ORDER BY n.age SKIP 0 LIMIT 50", query.getValue());
+	}
+
+	@Test
+	public void testNoResults() {
+		ArgumentCaptor<String> query = ArgumentCaptor.forClass(String.class);
+
+		when(template.query(query.capture(), (Map<String, Object>) isNull())).thenReturn(result);
+		when(result.to(String.class)).thenReturn(endResult);
+		when(endResult.iterator()).thenReturn(new ArrayList().iterator());
+
+		assertFalse(reader.doPageRead().hasNext());
+		assertEquals("START n=node(*) RETURN * ORDER BY n.age SKIP 0 LIMIT 50", query.getValue());
+	}
+
+	@Test
+	public void testResultsWithConverter() {
+		ResultConverter converter = new DefaultConverter();
+
+		reader.setResultConverter(converter);
+		ArgumentCaptor<String> query = ArgumentCaptor.forClass(String.class);
+
+		when(template.query(query.capture(), (Map<String, Object>) isNull())).thenReturn(result);
+		when(result.to(String.class, converter)).thenReturn(endResult);
+		when(endResult.iterator()).thenReturn(new ArrayList(){{
+			add(new String());
+		}}.iterator());
+
+		assertTrue(reader.doPageRead().hasNext());
+		assertEquals("START n=node(*) RETURN * ORDER BY n.age SKIP 0 LIMIT 50", query.getValue());
+	}
+
+	@Test
+	public void testResultsWithMatchAndWhere() throws Exception {
+		reader.setMatchStatement("n -- m");
+		reader.setWhereStatement("has(n.name)");
+		reader.setReturnStatement("m");
+		reader.afterPropertiesSet();
+		when(template.query("START n=node(*) MATCH n -- m WHERE has(n.name) RETURN m ORDER BY n.age SKIP 0 LIMIT 50", null)).thenReturn(result);
+		when(result.to(String.class)).thenReturn(endResult);
+		when(endResult.iterator()).thenReturn(new ArrayList(){{
+			add(new String());
+		}}.iterator());
+
+		assertTrue(reader.doPageRead().hasNext());
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/Neo4jItemWriterTests.java

@@ -0,0 +1,89 @@
+package org.springframework.batch.item.data;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.data.neo4j.template.Neo4jOperations;
+
+@SuppressWarnings("rawtypes")
+public class Neo4jItemWriterTests {
+
+	private Neo4jItemWriter writer;
+	@Mock
+	private Neo4jOperations template;
+
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+		writer = new Neo4jItemWriter();
+
+		writer.setTemplate(template);
+	}
+
+	@Test
+	public void testAfterPropertiesSet() throws Exception{
+		writer = new Neo4jItemWriter();
+
+		try {
+			writer.afterPropertiesSet();
+			fail("Template was not set but exception was not thrown.");
+		} catch (IllegalStateException iae) {
+			assertEquals("A Neo4JOperations implementation is required", iae.getMessage());
+		} catch (Throwable t) {
+			fail("Wrong exception was thrown.");
+		}
+
+		writer.setTemplate(template);
+
+		writer.afterPropertiesSet();
+	}
+
+	@Test
+	public void testWriteNull() throws Exception {
+		writer.write(null);
+
+		verifyZeroInteractions(template);
+	}
+
+	@Test
+	public void testWriteNoItems() throws Exception {
+		writer.write(new ArrayList());
+
+		verifyZeroInteractions(template);
+	}
+
+	@Test
+	public void testWriteItems() throws Exception {
+		List<String> items = new ArrayList<String>();
+		items.add("foo");
+		items.add("bar");
+
+		writer.write(items);
+
+		verify(template).save("foo");
+		verify(template).save("bar");
+	}
+
+	@Test
+	public void testDeleteItems() throws Exception {
+		List<String> items = new ArrayList<String>();
+		items.add("foo");
+		items.add("bar");
+
+		writer.setDelete(true);
+
+		writer.write(items);
+
+		verify(template).delete("foo");
+		verify(template).delete("bar");
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemReaderTests.java

@@ -0,0 +1,195 @@
+package org.springframework.batch.item.data;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.batch.item.adapter.DynamicMethodInvocationException;
+import org.springframework.data.domain.PageImpl;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
+import org.springframework.data.repository.PagingAndSortingRepository;
+
+@SuppressWarnings("rawtypes")
+public class RepositoryItemReaderTests {
+
+	private RepositoryItemReader reader;
+	@Mock
+	private PagingAndSortingRepository repository;
+	private Map<String, Sort.Direction> sorts;
+
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+		sorts = new HashMap<String, Sort.Direction>();
+		sorts.put("id", Direction.ASC);
+		reader = new RepositoryItemReader();
+		reader.setRepository(repository);
+		reader.setPageSize(1);
+		reader.setSort(sorts);
+		reader.setMethodName("findAll");
+	}
+
+	@Test
+	public void testAfterPropertiesSet() throws Exception {
+		try {
+			new RepositoryItemReader().afterPropertiesSet();
+			fail();
+		} catch (IllegalStateException e) {
+		}
+
+		try {
+			reader = new RepositoryItemReader();
+			reader.setRepository(repository);
+			reader.afterPropertiesSet();
+			fail();
+		} catch (IllegalStateException iae) {
+		}
+
+		try {
+			reader = new RepositoryItemReader();
+			reader.setRepository(repository);
+			reader.setPageSize(-1);
+			reader.afterPropertiesSet();
+			fail();
+		} catch (IllegalStateException iae) {
+		}
+
+		try {
+			reader = new RepositoryItemReader();
+			reader.setRepository(repository);
+			reader.setPageSize(1);
+			reader.afterPropertiesSet();
+			fail();
+		} catch (IllegalStateException iae) {
+		}
+
+		reader = new RepositoryItemReader();
+		reader.setRepository(repository);
+		reader.setPageSize(1);
+		reader.setSort(sorts);
+		reader.afterPropertiesSet();
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testDoReadFirstReadNoResults() throws Exception {
+		ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
+
+		when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl(new ArrayList()));
+
+		assertNull(reader.doRead());
+
+		Pageable pageRequest = pageRequestContainer.getValue();
+		assertEquals(0, pageRequest.getOffset());
+		assertEquals(0, pageRequest.getPageNumber());
+		assertEquals(1, pageRequest.getPageSize());
+		assertEquals("id: ASC", pageRequest.getSort().toString());
+	}
+
+	@Test
+	@SuppressWarnings({"serial", "unchecked"})
+	public void testDoReadFirstReadResults() throws Exception {
+		ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
+		final Object result = new Object();
+
+		when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl(new ArrayList(){{
+			add(result);
+		}}));
+
+		assertEquals(result, reader.doRead());
+
+		Pageable pageRequest = pageRequestContainer.getValue();
+		assertEquals(0, pageRequest.getOffset());
+		assertEquals(0, pageRequest.getPageNumber());
+		assertEquals(1, pageRequest.getPageSize());
+		assertEquals("id: ASC", pageRequest.getSort().toString());
+	}
+
+	@Test
+	@SuppressWarnings({"serial", "unchecked"})
+	public void testDoReadFirstReadSecondPage() throws Exception {
+		ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
+		final Object result = new Object();
+		when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl(new ArrayList(){{
+			add(new Object());
+		}})).thenReturn(new PageImpl(new ArrayList(){{
+			add(result);
+		}}));
+
+		assertFalse(reader.doRead() == result);
+		assertEquals(result, reader.doRead());
+
+		Pageable pageRequest = pageRequestContainer.getValue();
+		assertEquals(1, pageRequest.getOffset());
+		assertEquals(1, pageRequest.getPageNumber());
+		assertEquals(1, pageRequest.getPageSize());
+		assertEquals("id: ASC", pageRequest.getSort().toString());
+	}
+
+	@Test
+	@SuppressWarnings({"serial", "unchecked"})
+	public void testDoReadFirstReadExhausted() throws Exception {
+		ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
+		final Object result = new Object();
+		when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl(new ArrayList(){{
+			add(new Object());
+		}})).thenReturn(new PageImpl(new ArrayList(){{
+			add(result);
+		}})).thenReturn(new PageImpl(new ArrayList()));
+
+		assertFalse(reader.doRead() == result);
+		assertEquals(result, reader.doRead());
+		assertNull(reader.doRead());
+
+		Pageable pageRequest = pageRequestContainer.getValue();
+		assertEquals(2, pageRequest.getOffset());
+		assertEquals(2, pageRequest.getPageNumber());
+		assertEquals(1, pageRequest.getPageSize());
+		assertEquals("id: ASC", pageRequest.getSort().toString());
+	}
+
+	@Test
+	@SuppressWarnings({"serial", "unchecked"})
+	public void testJumpToItem() throws Exception {
+		reader.setPageSize(100);
+		ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
+		when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl(new ArrayList(){{
+			add(new Object());
+		}}));
+
+		reader.jumpToItem(485);
+
+		Pageable pageRequest = pageRequestContainer.getValue();
+		assertEquals(400, pageRequest.getOffset());
+		assertEquals(4, pageRequest.getPageNumber());
+		assertEquals(100, pageRequest.getPageSize());
+		assertEquals("id: ASC", pageRequest.getSort().toString());
+	}
+
+	@Test
+	public void testInvalidMethodName() throws Exception {
+		reader.setMethodName("thisMethodDoesNotExist");
+
+		try {
+			reader.doPageRead();
+			fail();
+		} catch (DynamicMethodInvocationException dmie) {
+			assertTrue(dmie.getCause() instanceof NoSuchMethodException);
+		}
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemWriterTests.java

@@ -0,0 +1,65 @@
+package org.springframework.batch.item.data;
+
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.data.repository.CrudRepository;
+
+@SuppressWarnings("rawtypes")
+public class RepositoryItemWriterTests {
+
+	@Mock
+	private CrudRepository repository;
+
+	private RepositoryItemWriter writer;
+
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+		writer = new RepositoryItemWriter();
+		writer.setMethodName("save");
+		writer.setRepository(repository);
+	}
+
+	@Test
+	public void testAfterPropertiesSet() throws Exception {
+		writer.afterPropertiesSet();
+
+		writer.setRepository(null);
+
+		try {
+			writer.afterPropertiesSet();
+			fail();
+		} catch (IllegalStateException e) {
+		}
+	}
+
+	@Test
+	public void testWriteNoItems() throws Exception {
+		writer.write(null);
+
+		writer.write(new ArrayList());
+
+		verifyZeroInteractions(repository);
+	}
+
+	@Test
+	@SuppressWarnings({"serial", "unchecked"})
+	public void testWriteItems() throws Exception {
+		List<Object> items = new ArrayList<Object>() {{
+			add("foo");
+		}};
+
+		writer.write(items);
+
+		verify(repository).save("foo");
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java

@@ -1,8 +1,6 @@
 package org.springframework.batch.item.database;
 
 import org.hibernate.SessionFactory;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.sample.Foo;
@@ -10,7 +8,6 @@ import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
 import org.springframework.orm.hibernate4.LocalSessionFactoryBean;
 
-@RunWith(JUnit4.class)
 public class HibernateCursorItemReaderCommonTests extends AbstractDatabaseItemStreamItemReaderTests {
 
 	@Override

spring-batch-parent/pom.xml

@@ -371,18 +371,6 @@
 				<version>${junit.version}</version>
 				<scope>test</scope>
 			</dependency>
-<!-- 			<dependency> -->
-<!-- 				<groupId>org.easymock</groupId> -->
-<!-- 				<artifactId>easymock</artifactId> -->
-<!-- 				<version>3.1</version> -->
-<!-- 				<scope>test</scope> -->
-<!-- 			</dependency> -->
-<!-- 			<dependency> -->
-<!-- 				<groupId>org.easymock</groupId> -->
-<!-- 				<artifactId>easymockclassextension</artifactId> -->
-<!-- 				<version>3.1</version> -->
-<!-- 				<scope>test</scope> -->
-<!-- 			</dependency> -->
 			<dependency>
 				<groupId>org.apache.geronimo.specs</groupId>
 				<artifactId>geronimo-jms_1.1_spec</artifactId>
@@ -453,6 +441,11 @@
 					</exclusion>
 				</exclusions>
 			</dependency>
+			<dependency>
+			  <groupId>org.hibernate</groupId>
+			  <artifactId>hibernate-validator</artifactId>
+			  <version>4.3.1.Final</version>
+			</dependency>
 			<dependency>
 				<groupId>org.hibernate</groupId>
 				<artifactId>hibernate-annotations</artifactId>
@@ -478,12 +471,12 @@
 			<dependency>
 				<groupId>org.slf4j</groupId>
 				<artifactId>slf4j-api</artifactId>
-				<version>1.5.8</version>
+	 			<version>1.6.6</version>
 			</dependency>
 			<dependency>
 				<groupId>org.slf4j</groupId>
 				<artifactId>slf4j-log4j12</artifactId>
-				<version>1.5.8</version>
+	 			<version>1.6.6</version>
 			</dependency>
 			<dependency>
 				<groupId>commons-lang</groupId>
@@ -696,6 +689,30 @@
 				<version>1.9.5</version>
 				<scope>test</scope>
 			</dependency>			
+			<dependency>
+				 <groupId>org.springframework.data</groupId>
+				 <artifactId>spring-data-commons-core</artifactId>
+				 <version>1.4.0.RC1</version>
+				 <optional>true</optional>
+			</dependency>
+	  		<dependency>
+				<groupId>org.springframework.data</groupId>
+				<artifactId>spring-data-jpa</artifactId>
+				<version>1.2.0.RELEASE</version>
+				<optional>true</optional>
+			</dependency>
+	 		<dependency>
+				<groupId>org.springframework.data</groupId>
+				<artifactId>spring-data-mongodb</artifactId>
+				<version>1.1.0.RELEASE</version>
+				 <optional>true</optional>
+			</dependency>
+	 		<dependency>
+				<groupId>org.springframework.data</groupId>
+				<artifactId>spring-data-neo4j</artifactId>
+				<version>2.1.0.RELEASE</version>
+				<optional>true</optional>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 	<distributionManagement>

spring-batch-samples/pom.xml

@@ -230,6 +230,15 @@
 			<optional>true</optional>
 			<scope>runtime</scope>
 		</dependency>
+		<dependency>
+			 <groupId>org.springframework.data</groupId>
+			 <artifactId>spring-data-commons-core</artifactId>
+			 <optional>true</optional>
+		</dependency>
+  		<dependency>
+			<groupId>org.springframework.data</groupId>
+			<artifactId>spring-data-jpa</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>javax.mail</groupId>
 			<artifactId>mail</artifactId>

spring-batch-samples/src/main/java/org/springframework/batch/sample/data/CustomerCreditRepository.java

@@ -0,0 +1,12 @@
+package org.springframework.batch.sample.data;
+
+import java.math.BigDecimal;
+
+import org.springframework.batch.sample.domain.trade.CustomerCredit;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.repository.PagingAndSortingRepository;
+
+public interface CustomerCreditRepository extends PagingAndSortingRepository<CustomerCredit, Long>{
+	Page<CustomerCredit> findByCreditGreaterThan(BigDecimal credit, Pageable request);
+}

spring-batch-samples/src/main/resources/jobs/iosample/repository.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xmlns:jpa="http://www.springframework.org/schema/data/jpa"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	 					http://www.springframework.org/schema/data/jpa http://www.springframework.org/schema/data/jpa/spring-jpa.xsd">
+
+	<jpa:repositories base-package="org.springframework.batch.sample.data"/>
+
+	<bean id="itemReader"
+		class="org.springframework.batch.item.data.RepositoryItemReader" scope="step">
+		<property name="pageSize" value="2"/>
+		<property name="methodName" value="findByCreditGreaterThan"/>
+		<property name="repository" ref="customerCreditRepository"/>
+		<property name="arguments">
+			<list>
+				<value>#{new java.math.BigDecimal(jobParameters[credit])}</value>
+			</list>
+		</property>
+ 		<property name="sort">
+			<map>
+				<entry key="id" value="ASC"/>
+			</map>
+ 		</property>
+	</bean>
+
+	<bean id="itemWriter"
+		class="org.springframework.batch.item.data.RepositoryItemWriter">
+		<property name="methodName" value="save"/>
+		<property name="repository" ref="customerCreditRepository"/>
+	</bean>
+	
+	<bean id="transactionManager" class="org.springframework.orm.jpa.JpaTransactionManager">
+		<property name="entityManagerFactory" ref="entityManagerFactory" />
+	</bean>
+
+	<bean id="entityManagerFactory"
+		class="org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean">
+		<property name="dataSource" ref="dataSource" />
+		<property name="persistenceUnitName" value="customerCredit" />
+		<property name="jpaVendorAdapter">
+			<bean class="org.springframework.orm.jpa.vendor.HibernateJpaVendorAdapter">
+				<property name="showSql" value="false" />
+			</bean>
+		</property>
+		<property name="jpaDialect">
+			<bean class="org.springframework.orm.jpa.vendor.HibernateJpaDialect" />
+		</property>
+	</bean>
+
+	<!--
+		Hibernate JPA dialect does not support custom tx isolation levels =>
+		overwrite with ISOLATION_DEFAULT
+	-->
+	<bean id="jobRepository"
+		class="org.springframework.batch.core.repository.support.JobRepositoryFactoryBean">
+		<property name="isolationLevelForCreate" value="ISOLATION_DEFAULT" />
+		<property name="dataSource" ref="dataSource" />
+		<property name="transactionManager" ref="transactionManager" />
+	</bean>
+</beans>

spring-batch-samples/src/test/java/org/springframework/batch/sample/iosample/RepositoryFunctionalTests.java

@@ -0,0 +1,31 @@
+package org.springframework.batch.sample.iosample;
+
+import org.junit.runner.RunWith;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.scope.context.StepSynchronizationManager;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.sample.domain.trade.CustomerCredit;
+import org.springframework.batch.test.MetaDataInstanceFactory;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(locations = "/jobs/iosample/repository.xml")
+public class RepositoryFunctionalTests extends AbstractIoSampleTests {
+
+	@Override
+	protected void pointReaderToOutput(ItemReader<CustomerCredit> reader) {
+		JobParameters jobParameters = new JobParametersBuilder(super.getUniqueJobParameters()).addDouble("credit", 0.)
+				.toJobParameters();
+		StepExecution stepExecution = MetaDataInstanceFactory.createStepExecution(jobParameters);
+		StepSynchronizationManager.close();
+		StepSynchronizationManager.register(stepExecution);
+	}
+
+	@Override
+	protected JobParameters getUniqueJobParameters() {
+		return new JobParametersBuilder(super.getUniqueJobParameters()).addString("credit", "10000").toJobParameters();
+	}
+}