From 91deb12a839dede16e820c25e97444a5f23feffa Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Tue, 4 Dec 2012 14:44:24 -0600 Subject: [PATCH] BATCH-1728: Initial commit of Spring Data ItemReader/ItemWriter implementations --- spring-batch-infrastructure/pom.xml | 28 +- .../data/AbstractPaginatedDataItemReader.java | 70 +++++ .../batch/item/data/MongoItemReader.java | 221 +++++++++++++++ .../batch/item/data/MongoItemWriter.java | 175 ++++++++++++ .../batch/item/data/Neo4jItemReader.java | 226 +++++++++++++++ .../batch/item/data/Neo4jItemWriter.java | 103 +++++++ .../batch/item/data/RepositoryItemReader.java | 263 ++++++++++++++++++ .../batch/item/data/RepositoryItemWriter.java | 153 ++++++++++ .../batch/item/data/MongoItemReaderTests.java | 183 ++++++++++++ .../batch/item/data/MongoItemWriterTests.java | 235 ++++++++++++++++ .../batch/item/data/Neo4jItemReaderTests.java | 164 +++++++++++ .../batch/item/data/Neo4jItemWriterTests.java | 89 ++++++ .../item/data/RepositoryItemReaderTests.java | 195 +++++++++++++ .../item/data/RepositoryItemWriterTests.java | 65 +++++ .../HibernateCursorItemReaderCommonTests.java | 3 - spring-batch-parent/pom.xml | 45 ++- spring-batch-samples/pom.xml | 9 + .../sample/data/CustomerCreditRepository.java | 12 + .../resources/jobs/iosample/repository.xml | 61 ++++ .../iosample/RepositoryFunctionalTests.java | 31 +++ 20 files changed, 2306 insertions(+), 25 deletions(-) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/AbstractPaginatedDataItemReader.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemReader.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemWriter.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/Neo4jItemReader.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/Neo4jItemWriter.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemReader.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemWriter.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemReaderTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemWriterTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/Neo4jItemReaderTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/Neo4jItemWriterTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemReaderTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemWriterTests.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/data/CustomerCreditRepository.java create mode 100644 spring-batch-samples/src/main/resources/jobs/iosample/repository.xml create mode 100644 spring-batch-samples/src/test/java/org/springframework/batch/sample/iosample/RepositoryFunctionalTests.java diff --git a/spring-batch-infrastructure/pom.xml b/spring-batch-infrastructure/pom.xml index c3c7c6c4c..6f1c075b8 100644 --- a/spring-batch-infrastructure/pom.xml +++ b/spring-batch-infrastructure/pom.xml @@ -39,14 +39,6 @@ junit junit - - - - - - - - org.aspectj aspectjrt @@ -123,6 +115,11 @@ hibernate-annotations true + + org.hibernate + hibernate-validator + true + org.apache.geronimo.specs geronimo-jta_1.1_spec @@ -191,6 +188,21 @@ spring-tx true + + org.springframework.data + spring-data-commons-core + true + + + org.springframework.data + spring-data-mongodb + true + + + org.springframework.data + spring-data-neo4j + true + org.codehaus.woodstox woodstox-core-asl diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/AbstractPaginatedDataItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/AbstractPaginatedDataItemReader.java new file mode 100644 index 000000000..6e5c09a01 --- /dev/null +++ b/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 extends + AbstractItemCountingItemStreamItemReader { + + protected volatile int page = 0; + + protected int pageSize = 10; + + protected Iterator 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(); + } + } + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemReader.java new file mode 100644 index 000000000..00231a53f --- /dev/null +++ b/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; + +/** + *

+ * Restartable {@link ItemReader} that reads documents from MongoDB + * via a paging technique. + *

+ * + *

+ * 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. + *

+ * + *

+ * The JSON query provided supports parameter substitution via ?<index> + * placeholders where the <index> indicates the index of the + * parameterValue to substitute. + *

+ * + *

+ * The 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 + */ +public class MongoItemReader extends AbstractPaginatedDataItemReader implements InitializingBean { + + private static final Pattern PLACEHOLDER = Pattern.compile("\\?(\\d+)"); + private MongoOperations template; + private String query; + private Class type; + private Sort sort; + private String hint; + private String fields; + private List 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 ?<index> placeholders where the <index> 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 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 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 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 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) 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 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 values, int index) { + return JSON.serialize(values.get(index)); + } + + 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); + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemWriter.java new file mode 100644 index 000000000..491eb8440 --- /dev/null +++ b/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; + +/** + *

+ * 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. + *

+ * + *

+ * This writer is thread safe once all properties are set (normal singleton behavior) so it can be used in multiple + * concurrent transactions. + *

+ * + * @author Michael Minella + * + */ +public class MongoItemWriter implements ItemWriter, 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 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 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 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."); + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/Neo4jItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/Neo4jItemReader.java new file mode 100644 index 000000000..6d7f8ec6e --- /dev/null +++ b/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; + +/** + *

+ * Restartable {@link ItemReader} that reads objects from the graph database Neo4j + * via a paging technique. + *

+ * + *

+ * 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. + *

+ * + *

+ * 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. + *

+ * + *

+ * This implementation is thread-safe between calls to + * {@link #open(org.springframework.batch.item.ExecutionContext)}, however you + * should set saveState=false if used in a multi-threaded + * environment (no restart available). + *

+ * + * @author Michael Minella + * + */ +public class Neo4jItemReader extends AbstractPaginatedDataItemReader 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 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 not 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 not 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 not + * 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 not + * 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 not 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 doPageRead() { + Result> 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"); + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/Neo4jItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/Neo4jItemWriter.java new file mode 100644 index 000000000..47e5b6b04 --- /dev/null +++ b/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; + +/** + *

+ * A {@link ItemWriter} implementation that writes to a Neo4j database using an + * implementation of Spring Data's {@link Neo4jOperations}. + *

+ * + *

+ * This writer is thread safe once all properties are set (normal singleton + * behavior) so it can be used in multiple concurrent transactions. + *

+ * + * @author Michael Minella + * + */ +public class Neo4jItemWriter implements ItemWriter, 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 - - - - - - - - - - - - org.apache.geronimo.specs geronimo-jms_1.1_spec @@ -453,6 +441,11 @@ + + org.hibernate + hibernate-validator + 4.3.1.Final + org.hibernate hibernate-annotations @@ -478,12 +471,12 @@ org.slf4j slf4j-api - 1.5.8 + 1.6.6 org.slf4j slf4j-log4j12 - 1.5.8 + 1.6.6 commons-lang @@ -696,6 +689,30 @@ 1.9.5 test + + org.springframework.data + spring-data-commons-core + 1.4.0.RC1 + true + + + org.springframework.data + spring-data-jpa + 1.2.0.RELEASE + true + + + org.springframework.data + spring-data-mongodb + 1.1.0.RELEASE + true + + + org.springframework.data + spring-data-neo4j + 2.1.0.RELEASE + true + diff --git a/spring-batch-samples/pom.xml b/spring-batch-samples/pom.xml index 860be73b1..41e050dd2 100644 --- a/spring-batch-samples/pom.xml +++ b/spring-batch-samples/pom.xml @@ -230,6 +230,15 @@ true runtime + + org.springframework.data + spring-data-commons-core + true + + + org.springframework.data + spring-data-jpa + javax.mail mail diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/data/CustomerCreditRepository.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/data/CustomerCreditRepository.java new file mode 100644 index 000000000..fea3a65e9 --- /dev/null +++ b/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{ + Page findByCreditGreaterThan(BigDecimal credit, Pageable request); +} diff --git a/spring-batch-samples/src/main/resources/jobs/iosample/repository.xml b/spring-batch-samples/src/main/resources/jobs/iosample/repository.xml new file mode 100644 index 000000000..73f4f462d --- /dev/null +++ b/spring-batch-samples/src/main/resources/jobs/iosample/repository.xml @@ -0,0 +1,61 @@ + + + + + + + + + + + + #{new java.math.BigDecimal(jobParameters[credit])} + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/iosample/RepositoryFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/iosample/RepositoryFunctionalTests.java new file mode 100644 index 000000000..4bf4f3216 --- /dev/null +++ b/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 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(); + } +}