BATCH-1728: Initial commit of Spring Data ItemReader/ItemWriter implementations

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

@@ -0,0 +1,70 @@
+package org.springframework.batch.item.data;
+
+import java.util.Iterator;
+
+import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
+
+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();
+
+	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;
+			}
+		}
+	}
+
+	protected abstract Iterator 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 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,153 @@
+/*
+ * 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)
+	 */
+	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.
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.state(repository != null, "A CRUDRepository 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;
+	}
+}