, 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 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 items) {
+ if(delete) {
+ for (T t : items) {
+ template.delete(t);
+ }
+ }
+ else {
+ for (T t : items) {
+ template.save(t);
+ }
+ }
+ }
+}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemReader.java
new file mode 100644
index 000000000..cce7ad9af
--- /dev/null
+++ b/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;
+
+/**
+ *
+ * A {@link org.springframework.batch.item.ItemReader} that reads records utilizing
+ * a {@link org.springframework.data.repository.PagingAndSortingRepository}.
+ *
+ *
+ *
+ * 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.
+ *
+ *
+ *
+ * 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.
+ *
+ *
+ *
+ * This implementation is thread safe between calls to {@link #open(ExecutionContext)}, but remember to use
+ * saveState=false if used in a multi-threaded client (no restart available).
+ *
+ *
+ * @author Michael Minella
+ * @since 2.2
+ */
+@SuppressWarnings("rawtypes")
+public class RepositoryItemReader extends AbstractItemCountingItemStreamItemReader 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 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 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 last 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 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 sorts) {
+ List sortValues = new ArrayList();
+
+ for (Map.Entry 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;
+ }
+}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemWriter.java
new file mode 100644
index 000000000..24f9edb83
--- /dev/null
+++ b/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;
+
+/**
+ *
+ * A {@link org.springframework.batch.item.ItemReader} wrapper for a
+ * {@link org.springframework.data.repository.CrudRepository} from Spring Data.
+ *
+ *
+ *
+ * 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.
+ *
+ *
+ *
+ * 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.
+ *
+ *
+ * @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 sole 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;
+ }
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemReaderTests.java
new file mode 100644
index 000000000..9552d44bf
--- /dev/null
+++ b/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 sortOptions;
+
+ @Before
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ reader = new MongoItemReader();
+
+ sortOptions = new HashMap();
+ 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 queryContainer = ArgumentCaptor.forClass(Query.class);
+
+ when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList());
+
+ 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 queryContainer = ArgumentCaptor.forClass(Query.class);
+
+ when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList());
+
+ 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 queryContainer = ArgumentCaptor.forClass(Query.class);
+
+ when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList());
+
+ 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 queryContainer = ArgumentCaptor.forClass(Query.class);
+
+ when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList());
+
+ 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(){{
+ add("foo");
+ }});
+
+ reader.setQuery("{ name : ?0 }");
+ ArgumentCaptor queryContainer = ArgumentCaptor.forClass(Query.class);
+
+ when(template.find(queryContainer.capture(), eq(String.class))).thenReturn(new ArrayList());
+
+ 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());
+ }
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemWriterTests.java
new file mode 100644
index 000000000..ddfbca6fc
--- /dev/null
+++ b/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 items = new ArrayList() {{
+ 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 items = new ArrayList() {{
+ 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 items = new ArrayList() {{
+ 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 items = new ArrayList() {{
+ 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 items = new ArrayList() {{
+ 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 items = new ArrayList() {{
+ 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 items = new ArrayList() {{
+ 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 items = new ArrayList() {{
+ 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");
+ }
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/Neo4jItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/Neo4jItemReaderTests.java
new file mode 100644
index 000000000..17eb52d9b
--- /dev/null
+++ b/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 query = ArgumentCaptor.forClass(String.class);
+
+ when(template.query(query.capture(), (Map) 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 query = ArgumentCaptor.forClass(String.class);
+
+ when(template.query(query.capture(), (Map) 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 query = ArgumentCaptor.forClass(String.class);
+
+ when(template.query(query.capture(), (Map) 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());
+ }
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/Neo4jItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/Neo4jItemWriterTests.java
new file mode 100644
index 000000000..40ba364d6
--- /dev/null
+++ b/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 items = new ArrayList();
+ items.add("foo");
+ items.add("bar");
+
+ writer.write(items);
+
+ verify(template).save("foo");
+ verify(template).save("bar");
+ }
+
+ @Test
+ public void testDeleteItems() throws Exception {
+ List items = new ArrayList();
+ items.add("foo");
+ items.add("bar");
+
+ writer.setDelete(true);
+
+ writer.write(items);
+
+ verify(template).delete("foo");
+ verify(template).delete("bar");
+ }
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemReaderTests.java
new file mode 100644
index 000000000..2469ae4ea
--- /dev/null
+++ b/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 sorts;
+
+ @Before
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ sorts = new HashMap();
+ 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 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 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 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 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 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);
+ }
+ }
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemWriterTests.java
new file mode 100644
index 000000000..c291533e7
--- /dev/null
+++ b/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 items = new ArrayList() {{
+ add("foo");
+ }};
+
+ writer.write(items);
+
+ verify(repository).save("foo");
+ }
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java
index 5a97d5510..33d05e8cf 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java
+++ b/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
diff --git a/spring-batch-parent/pom.xml b/spring-batch-parent/pom.xml
index 5bd6374f4..53eec5b43 100644
--- a/spring-batch-parent/pom.xml
+++ b/spring-batch-parent/pom.xml
@@ -371,18 +371,6 @@
${junit.version}
test