Refine contribution #4323
- Update parameter types - Update tests - Update Javadocs - Fix code formatting
This commit is contained in:
Mahmoud Ben Hassine
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2023 the original author or authors.
|
||||
* Copyright 2023 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.
|
||||
@@ -16,7 +16,6 @@
|
||||
package org.springframework.batch.item.data;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
@@ -26,6 +25,7 @@ import java.util.stream.Stream;
|
||||
import org.bson.Document;
|
||||
import org.bson.codecs.DecoderContext;
|
||||
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.data.domain.Sort;
|
||||
@@ -40,258 +40,268 @@ import org.springframework.util.ClassUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Cursor-based {@link ItemReader} implementation for MongoDB.
|
||||
*
|
||||
* @author LEE Juchan
|
||||
* @since 5.0
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 5.1
|
||||
*/
|
||||
public class MongoCursorItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements InitializingBean {
|
||||
|
||||
private MongoOperations template;
|
||||
private MongoOperations template;
|
||||
|
||||
private Class<? extends T> targetType;
|
||||
private Class<? extends T> targetType;
|
||||
|
||||
private String collection;
|
||||
private String collection;
|
||||
|
||||
private Query query;
|
||||
private Query query;
|
||||
|
||||
private String queryString;
|
||||
private String queryString;
|
||||
|
||||
private List<Object> parameterValues = new ArrayList<>();
|
||||
private List<Object> parameterValues = new ArrayList<>();
|
||||
|
||||
private String fields;
|
||||
private String fields;
|
||||
|
||||
private Sort sort;
|
||||
private Sort sort;
|
||||
|
||||
private String hint;
|
||||
private String hint;
|
||||
|
||||
private Integer batchSize;
|
||||
private int batchSize;
|
||||
|
||||
private Integer limit;
|
||||
private int limit;
|
||||
|
||||
private Integer maxTimeMs;
|
||||
private Duration maxTime;
|
||||
|
||||
private CloseableIterator<? extends T> cursor;
|
||||
private CloseableIterator<? extends T> cursor;
|
||||
|
||||
public MongoCursorItemReader() {
|
||||
super();
|
||||
setName(ClassUtils.getShortName(MongoCursorItemReader.class));
|
||||
}
|
||||
/**
|
||||
* Create a new {@link MongoCursorItemReader}.
|
||||
*/
|
||||
public MongoCursorItemReader() {
|
||||
super();
|
||||
setName(ClassUtils.getShortName(MongoCursorItemReader.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;
|
||||
}
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
|
||||
/**
|
||||
* The targetType of object to be returned for each {@link #read()} call.
|
||||
* @param targetType the targetType of object to return
|
||||
*/
|
||||
public void setTargetType(Class<? extends T> targetType) {
|
||||
this.targetType = targetType;
|
||||
}
|
||||
/**
|
||||
* The targetType of object to be returned for each {@link #read()} call.
|
||||
* @param targetType the targetType of object to return
|
||||
*/
|
||||
public void setTargetType(Class<? extends T> targetType) {
|
||||
this.targetType = targetType;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param collection Mongo collection to be queried.
|
||||
*/
|
||||
public void setCollection(String collection) {
|
||||
this.collection = collection;
|
||||
}
|
||||
/**
|
||||
* @param collection Mongo collection to be queried.
|
||||
*/
|
||||
public void setCollection(String collection) {
|
||||
this.collection = collection;
|
||||
}
|
||||
|
||||
/**
|
||||
* A Mongo Query to be used.
|
||||
* @param query Mongo Query to be used.
|
||||
*/
|
||||
public void setQuery(Query query) {
|
||||
this.query = query;
|
||||
}
|
||||
/**
|
||||
* A Mongo Query to be used.
|
||||
* @param query Mongo Query to be used.
|
||||
*/
|
||||
public void setQuery(Query query) {
|
||||
this.query = query;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 queryString JSON formatted Mongo query
|
||||
*/
|
||||
public void setQuery(String queryString) {
|
||||
this.queryString = queryString;
|
||||
}
|
||||
/**
|
||||
* 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 queryString JSON formatted Mongo query
|
||||
*/
|
||||
public void setQuery(String queryString) {
|
||||
this.queryString = queryString;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@link List} of values to be substituted in for each of the parameters in the
|
||||
* query.
|
||||
* @param parameterValues values
|
||||
*/
|
||||
public void setParameterValues(List<Object> parameterValues) {
|
||||
Assert.notNull(parameterValues, "Parameter values must not be null");
|
||||
this.parameterValues = parameterValues;
|
||||
}
|
||||
/**
|
||||
* {@link List} of values to be substituted in for each of the parameters in the
|
||||
* query.
|
||||
* @param parameterValues values
|
||||
*/
|
||||
public void setParameterValues(List<Object> parameterValues) {
|
||||
Assert.notNull(parameterValues, "Parameter values must not be null");
|
||||
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 sort by.
|
||||
*/
|
||||
public void setFields(String fields) {
|
||||
this.fields = fields;
|
||||
}
|
||||
/**
|
||||
* JSON defining the fields to be returned from the matching documents by MongoDB.
|
||||
* @param fields JSON string that identifies the fields to sort by.
|
||||
*/
|
||||
public void setFields(String fields) {
|
||||
this.fields = fields;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@link Map} of property
|
||||
* names/{@link org.springframework.data.domain.Sort.Direction} values to sort the
|
||||
* input by.
|
||||
* @param sorts map of properties and direction to sort each.
|
||||
*/
|
||||
public void setSort(Map<String, Sort.Direction> sorts) {
|
||||
Assert.notNull(sorts, "Sorts must not be null");
|
||||
this.sort = convertToSort(sorts);
|
||||
}
|
||||
/**
|
||||
* {@link Map} of property
|
||||
* names/{@link org.springframework.data.domain.Sort.Direction} values to sort the
|
||||
* input by.
|
||||
* @param sorts map of properties and direction to sort each.
|
||||
*/
|
||||
public void setSort(Map<String, Sort.Direction> sorts) {
|
||||
Assert.notNull(sorts, "Sorts must not be null");
|
||||
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;
|
||||
}
|
||||
/**
|
||||
* JSON String telling MongoDB what index to use.
|
||||
* @param hint string indicating what index to use.
|
||||
*/
|
||||
public void setHint(String hint) {
|
||||
this.hint = hint;
|
||||
}
|
||||
|
||||
/**
|
||||
* The size of batches to use when iterating over results.
|
||||
* @param batchSize size the batch size to apply to the cursor
|
||||
*/
|
||||
public void setBatchSize(Integer batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
/**
|
||||
* The size of batches to use when iterating over results.
|
||||
* @param batchSize size the batch size to apply to the cursor
|
||||
* @see Query#cursorBatchSize(int)
|
||||
*/
|
||||
public void setBatchSize(int batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* The query limit
|
||||
* @param limit The limit
|
||||
*/
|
||||
public void setLimit(Integer limit) {
|
||||
this.limit = limit;
|
||||
}
|
||||
/**
|
||||
* The query limit.
|
||||
* @param limit The limit
|
||||
* @see Query#limit(int)
|
||||
*/
|
||||
public void setLimit(int limit) {
|
||||
this.limit = limit;
|
||||
}
|
||||
|
||||
/**
|
||||
* The maximum execution time for the aggregation command
|
||||
* @param maxTimeMs The max time
|
||||
*/
|
||||
public void setMaxTimeMs(Integer maxTimeMs) {
|
||||
this.maxTimeMs = maxTimeMs;
|
||||
}
|
||||
/**
|
||||
* The maximum execution time for the query
|
||||
* @param maxTime The max time
|
||||
* @see Query#maxTime(Duration)
|
||||
*/
|
||||
public void setMaxTime(Duration maxTime) {
|
||||
Assert.notNull(maxTime, "maxTime must not be null.");
|
||||
this.maxTime = maxTime;
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks mandatory properties
|
||||
*
|
||||
* @see InitializingBean#afterPropertiesSet()
|
||||
*/
|
||||
@Override
|
||||
public void afterPropertiesSet() {
|
||||
Assert.state(template != null, "An implementation of MongoOperations is required.");
|
||||
Assert.state(targetType != null, "A targetType to convert the input into is required.");
|
||||
Assert.state(queryString != null || query != null, "A query is required.");
|
||||
/**
|
||||
* Checks mandatory properties
|
||||
*
|
||||
* @see InitializingBean#afterPropertiesSet()
|
||||
*/
|
||||
@Override
|
||||
public void afterPropertiesSet() {
|
||||
Assert.state(template != null, "An implementation of MongoOperations is required.");
|
||||
Assert.state(targetType != null, "A targetType to convert the input into is required.");
|
||||
Assert.state(queryString != null || query != null, "A query is required.");
|
||||
|
||||
if (queryString != null) {
|
||||
Assert.state(sort != null, "A sort is required.");
|
||||
}
|
||||
}
|
||||
if (queryString != null) {
|
||||
Assert.state(sort != null, "A sort is required.");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doOpen() throws Exception {
|
||||
Query mongoQuery;
|
||||
if (queryString != null) {
|
||||
mongoQuery = createQuery();
|
||||
} else {
|
||||
mongoQuery = query;
|
||||
}
|
||||
@Override
|
||||
protected void doOpen() throws Exception {
|
||||
Query mongoQuery;
|
||||
if (queryString != null) {
|
||||
mongoQuery = createQuery();
|
||||
}
|
||||
else {
|
||||
mongoQuery = query;
|
||||
}
|
||||
|
||||
Stream<? extends T> stream;
|
||||
if (StringUtils.hasText(collection)) {
|
||||
stream = template.stream(mongoQuery, targetType, collection);
|
||||
} else {
|
||||
stream = template.stream(mongoQuery, targetType);
|
||||
}
|
||||
Stream<? extends T> stream;
|
||||
if (StringUtils.hasText(collection)) {
|
||||
stream = template.stream(mongoQuery, targetType, collection);
|
||||
}
|
||||
else {
|
||||
stream = template.stream(mongoQuery, targetType);
|
||||
}
|
||||
|
||||
this.cursor = streamToIterator(stream);
|
||||
}
|
||||
this.cursor = streamToIterator(stream);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected T doRead() throws Exception {
|
||||
return cursor.hasNext() ? cursor.next() : null;
|
||||
}
|
||||
@Override
|
||||
protected T doRead() throws Exception {
|
||||
return cursor.hasNext() ? cursor.next() : null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doClose() throws Exception {
|
||||
this.cursor.close();
|
||||
}
|
||||
@Override
|
||||
protected void doClose() throws Exception {
|
||||
this.cursor.close();
|
||||
}
|
||||
|
||||
private Sort convertToSort(Map<String, Sort.Direction> sorts) {
|
||||
List<Sort.Order> sortValues = new ArrayList<>(sorts.size());
|
||||
private Sort convertToSort(Map<String, Sort.Direction> sorts) {
|
||||
List<Sort.Order> sortValues = new ArrayList<>(sorts.size());
|
||||
|
||||
for (Map.Entry<String, Sort.Direction> curSort : sorts.entrySet()) {
|
||||
sortValues.add(new Sort.Order(curSort.getValue(), curSort.getKey()));
|
||||
}
|
||||
for (Map.Entry<String, Sort.Direction> curSort : sorts.entrySet()) {
|
||||
sortValues.add(new Sort.Order(curSort.getValue(), curSort.getKey()));
|
||||
}
|
||||
|
||||
return Sort.by(sortValues);
|
||||
}
|
||||
return Sort.by(sortValues);
|
||||
}
|
||||
|
||||
private Query createQuery() {
|
||||
String populatedQuery = replacePlaceholders(queryString, parameterValues);
|
||||
private Query createQuery() {
|
||||
String populatedQuery = replacePlaceholders(queryString, parameterValues);
|
||||
|
||||
Query mongoQuery;
|
||||
if (StringUtils.hasText(fields)) {
|
||||
mongoQuery = new BasicQuery(populatedQuery, fields);
|
||||
} else {
|
||||
mongoQuery = new BasicQuery(populatedQuery);
|
||||
}
|
||||
Query mongoQuery;
|
||||
if (StringUtils.hasText(fields)) {
|
||||
mongoQuery = new BasicQuery(populatedQuery, fields);
|
||||
}
|
||||
else {
|
||||
mongoQuery = new BasicQuery(populatedQuery);
|
||||
}
|
||||
|
||||
if (sort != null) {
|
||||
mongoQuery.with(sort);
|
||||
}
|
||||
if (StringUtils.hasText(hint)) {
|
||||
mongoQuery.withHint(hint);
|
||||
}
|
||||
if (batchSize != null) {
|
||||
mongoQuery.cursorBatchSize(batchSize);
|
||||
}
|
||||
if (limit != null) {
|
||||
mongoQuery.limit(limit);
|
||||
}
|
||||
if (maxTimeMs != null) {
|
||||
mongoQuery.maxTime(Duration.of(maxTimeMs, ChronoUnit.MILLIS));
|
||||
} else {
|
||||
mongoQuery.noCursorTimeout();
|
||||
}
|
||||
if (sort != null) {
|
||||
mongoQuery.with(sort);
|
||||
}
|
||||
if (StringUtils.hasText(hint)) {
|
||||
mongoQuery.withHint(hint);
|
||||
}
|
||||
mongoQuery.cursorBatchSize(batchSize);
|
||||
mongoQuery.limit(limit);
|
||||
if (maxTime != null) {
|
||||
mongoQuery.maxTime(maxTime);
|
||||
}
|
||||
else {
|
||||
mongoQuery.noCursorTimeout();
|
||||
}
|
||||
|
||||
return mongoQuery;
|
||||
}
|
||||
return mongoQuery;
|
||||
}
|
||||
|
||||
private String replacePlaceholders(String input, List<Object> values) {
|
||||
ParameterBindingJsonReader reader = new ParameterBindingJsonReader(input, values.toArray());
|
||||
DecoderContext decoderContext = DecoderContext.builder().build();
|
||||
Document document = new ParameterBindingDocumentCodec().decode(reader, decoderContext);
|
||||
return document.toJson();
|
||||
}
|
||||
private String replacePlaceholders(String input, List<Object> values) {
|
||||
ParameterBindingJsonReader reader = new ParameterBindingJsonReader(input, values.toArray());
|
||||
DecoderContext decoderContext = DecoderContext.builder().build();
|
||||
Document document = new ParameterBindingDocumentCodec().decode(reader, decoderContext);
|
||||
return document.toJson();
|
||||
}
|
||||
|
||||
private CloseableIterator<? extends T> streamToIterator(Stream<? extends T> stream) {
|
||||
return new CloseableIterator<>() {
|
||||
final private Iterator<? extends T> delegate = stream.iterator();
|
||||
private CloseableIterator<? extends T> streamToIterator(Stream<? extends T> stream) {
|
||||
return new CloseableIterator<>() {
|
||||
final private Iterator<? extends T> delegate = stream.iterator();
|
||||
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
return delegate.hasNext();
|
||||
}
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
return delegate.hasNext();
|
||||
}
|
||||
|
||||
@Override
|
||||
public T next() {
|
||||
return delegate.next();
|
||||
}
|
||||
@Override
|
||||
public T next() {
|
||||
return delegate.next();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
stream.close();
|
||||
}
|
||||
};
|
||||
}
|
||||
@Override
|
||||
public void close() {
|
||||
stream.close();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2023 the original author or authors.
|
||||
* Copyright 2023 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.
|
||||
@@ -16,6 +16,7 @@
|
||||
|
||||
package org.springframework.batch.item.data.builder;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
@@ -29,276 +30,282 @@ import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* @author LEE Juchan
|
||||
* @since 5.0
|
||||
* @author Mahmoud Ben Hassine
|
||||
* @since 5.1
|
||||
* @see MongoCursorItemReader
|
||||
*/
|
||||
public class MongoCursorItemReaderBuilder<T> {
|
||||
|
||||
private boolean saveState = true;
|
||||
private boolean saveState = true;
|
||||
|
||||
private String name;
|
||||
private String name;
|
||||
|
||||
private int maxItemCount = Integer.MAX_VALUE;
|
||||
private int maxItemCount = Integer.MAX_VALUE;
|
||||
|
||||
private int currentItemCount;
|
||||
private int currentItemCount;
|
||||
|
||||
private MongoOperations template;
|
||||
private MongoOperations template;
|
||||
|
||||
private Class<? extends T> targetType;
|
||||
private Class<? extends T> targetType;
|
||||
|
||||
private String collection;
|
||||
private String collection;
|
||||
|
||||
private Query query;
|
||||
private Query query;
|
||||
|
||||
private String jsonQuery;
|
||||
private String jsonQuery;
|
||||
|
||||
private List<Object> parameterValues = new ArrayList<>();
|
||||
private List<Object> parameterValues = new ArrayList<>();
|
||||
|
||||
private String fields;
|
||||
private String fields;
|
||||
|
||||
private Map<String, Sort.Direction> sorts;
|
||||
private Map<String, Sort.Direction> sorts;
|
||||
|
||||
private String hint;
|
||||
private String hint;
|
||||
|
||||
private Integer batchSize;
|
||||
private int batchSize;
|
||||
|
||||
private Integer limit;
|
||||
private int limit;
|
||||
|
||||
private Integer maxTimeMs;
|
||||
private Duration maxTime;
|
||||
|
||||
/**
|
||||
* Configure if the state of the
|
||||
* {@link org.springframework.batch.item.ItemStreamSupport} should be persisted within
|
||||
* the {@link org.springframework.batch.item.ExecutionContext} for restart purposes.
|
||||
* @param saveState defaults to true
|
||||
* @return The current instance of the builder.
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> saveState(boolean saveState) {
|
||||
this.saveState = saveState;
|
||||
/**
|
||||
* Configure if the state of the
|
||||
* {@link org.springframework.batch.item.ItemStreamSupport} should be persisted within
|
||||
* the {@link org.springframework.batch.item.ExecutionContext} for restart purposes.
|
||||
* @param saveState defaults to true
|
||||
* @return The current instance of the builder.
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> saveState(boolean saveState) {
|
||||
this.saveState = saveState;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The name used to calculate the key within the
|
||||
* {@link org.springframework.batch.item.ExecutionContext}. Required if
|
||||
* {@link #saveState(boolean)} is set to true.
|
||||
* @param name name of the reader instance
|
||||
* @return The current instance of the builder.
|
||||
* @see org.springframework.batch.item.ItemStreamSupport#setName(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> name(String name) {
|
||||
this.name = name;
|
||||
/**
|
||||
* The name used to calculate the key within the
|
||||
* {@link org.springframework.batch.item.ExecutionContext}. Required if
|
||||
* {@link #saveState(boolean)} is set to true.
|
||||
* @param name name of the reader instance
|
||||
* @return The current instance of the builder.
|
||||
* @see org.springframework.batch.item.ItemStreamSupport#setName(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> name(String name) {
|
||||
this.name = name;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Configure the max number of items to be read.
|
||||
* @param maxItemCount the max items to be read
|
||||
* @return The current instance of the builder.
|
||||
* @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setMaxItemCount(int)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> maxItemCount(int maxItemCount) {
|
||||
this.maxItemCount = maxItemCount;
|
||||
/**
|
||||
* Configure the max number of items to be read.
|
||||
* @param maxItemCount the max items to be read
|
||||
* @return The current instance of the builder.
|
||||
* @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setMaxItemCount(int)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> maxItemCount(int maxItemCount) {
|
||||
this.maxItemCount = maxItemCount;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Index for the current item. Used on restarts to indicate where to start from.
|
||||
* @param currentItemCount current index
|
||||
* @return this instance for method chaining
|
||||
* @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setCurrentItemCount(int)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> currentItemCount(int currentItemCount) {
|
||||
this.currentItemCount = currentItemCount;
|
||||
/**
|
||||
* Index for the current item. Used on restarts to indicate where to start from.
|
||||
* @param currentItemCount current index
|
||||
* @return this instance for method chaining
|
||||
* @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setCurrentItemCount(int)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> currentItemCount(int currentItemCount) {
|
||||
this.currentItemCount = currentItemCount;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setTemplate(MongoOperations)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> template(MongoOperations template) {
|
||||
this.template = template;
|
||||
/**
|
||||
* 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
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setTemplate(MongoOperations)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> template(MongoOperations template) {
|
||||
this.template = template;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The targetType of object to be returned for each {@link MongoCursorItemReader#read()} call.
|
||||
* @param targetType the targetType of object to return
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setTargetType(Class)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> targetType(Class<? extends T> targetType) {
|
||||
this.targetType = targetType;
|
||||
/**
|
||||
* The targetType of object to be returned for each
|
||||
* {@link MongoCursorItemReader#read()} call.
|
||||
* @param targetType the targetType of object to return
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setTargetType(Class)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> targetType(Class<? extends T> targetType) {
|
||||
this.targetType = targetType;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Establish an optional collection that can be queried.
|
||||
* @param collection Mongo collection to be queried.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setCollection(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> collection(String collection) {
|
||||
this.collection = collection;
|
||||
/**
|
||||
* Establish an optional collection that can be queried.
|
||||
* @param collection Mongo collection to be queried.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setCollection(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> collection(String collection) {
|
||||
this.collection = collection;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provide a Spring Data Mongo {@link Query}. This will take precedence over a JSON
|
||||
* configured query.
|
||||
* @param query Query to execute
|
||||
* @return this instance for method chaining
|
||||
* @see MongoCursorItemReader#setQuery(Query)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> query(Query query) {
|
||||
this.query = query;
|
||||
/**
|
||||
* Provide a Spring Data Mongo {@link Query}. This will take precedence over a JSON
|
||||
* configured query.
|
||||
* @param query Query to execute
|
||||
* @return this instance for method chaining
|
||||
* @see MongoCursorItemReader#setQuery(Query)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> query(Query query) {
|
||||
this.query = query;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* A JSON formatted MongoDB jsonQuery. Parameterization of the provided jsonQuery is
|
||||
* allowed via ?<index> placeholders where the <index> indicates the index
|
||||
* of the parameterValue to substitute.
|
||||
* @param query JSON formatted Mongo jsonQuery
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setQuery(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> jsonQuery(String query) {
|
||||
this.jsonQuery = query;
|
||||
/**
|
||||
* A JSON formatted MongoDB jsonQuery. Parameterization of the provided jsonQuery is
|
||||
* allowed via ?<index> placeholders where the <index> indicates the index
|
||||
* of the parameterValue to substitute.
|
||||
* @param query JSON formatted Mongo jsonQuery
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setQuery(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> jsonQuery(String query) {
|
||||
this.jsonQuery = query;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Values to be substituted in for each of the parameters in the query.
|
||||
* @param parameterValues values
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setParameterValues(List)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> parameterValues(List<Object> parameterValues) {
|
||||
this.parameterValues = parameterValues;
|
||||
/**
|
||||
* Values to be substituted in for each of the parameters in the query.
|
||||
* @param parameterValues values
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setParameterValues(List)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> parameterValues(List<Object> parameterValues) {
|
||||
this.parameterValues = parameterValues;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* JSON defining the fields to be returned from the matching documents by MongoDB.
|
||||
* @param fields JSON string that identifies the fields to sort by.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setFields(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> fields(String fields) {
|
||||
this.fields = fields;
|
||||
/**
|
||||
* JSON defining the fields to be returned from the matching documents by MongoDB.
|
||||
* @param fields JSON string that identifies the fields to sort by.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setFields(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> fields(String fields) {
|
||||
this.fields = fields;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@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.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setSort(Map)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> sorts(Map<String, Sort.Direction> sorts) {
|
||||
this.sorts = sorts;
|
||||
/**
|
||||
* {@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.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setSort(Map)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> sorts(Map<String, Sort.Direction> sorts) {
|
||||
this.sorts = sorts;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* JSON String telling MongoDB what index to use.
|
||||
* @param hint string indicating what index to use.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setHint(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> hint(String hint) {
|
||||
this.hint = hint;
|
||||
/**
|
||||
* JSON String telling MongoDB what index to use.
|
||||
* @param hint string indicating what index to use.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setHint(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> hint(String hint) {
|
||||
this.hint = hint;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The size of batches to use when iterating over results.
|
||||
* @param batchSize string indicating what index to use.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setHint(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> batchSize(Integer batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
/**
|
||||
* The size of batches to use when iterating over results.
|
||||
* @param batchSize string indicating what index to use.
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setHint(String)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> batchSize(int batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The query limit
|
||||
* @param limit The limit
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setLimit(Integer)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> limit(Integer limit) {
|
||||
this.limit = limit;
|
||||
/**
|
||||
* The query limit
|
||||
* @param limit The limit
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setLimit(int)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> limit(int limit) {
|
||||
this.limit = limit;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The maximum execution time for the aggregation command
|
||||
* @param maxTimeMs The max time
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setMaxTimeMs(Integer)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> maxTimeMs(Integer maxTimeMs) {
|
||||
this.maxTimeMs = maxTimeMs;
|
||||
/**
|
||||
* The maximum execution time for the query
|
||||
* @param maxTime The max time
|
||||
* @return The current instance of the builder
|
||||
* @see MongoCursorItemReader#setMaxTime(Duration)
|
||||
*/
|
||||
public MongoCursorItemReaderBuilder<T> maxTime(Duration maxTime) {
|
||||
Assert.notNull(maxTime, "maxTime must not be null.");
|
||||
this.maxTime = maxTime;
|
||||
|
||||
return this;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
public MongoCursorItemReader<T> build() {
|
||||
Assert.notNull(this.template, "template is required.");
|
||||
if (this.saveState) {
|
||||
Assert.hasText(this.name, "A name is required when saveState is set to true");
|
||||
}
|
||||
Assert.notNull(this.targetType, "targetType is required.");
|
||||
Assert.state(StringUtils.hasText(this.jsonQuery) || this.query != null, "A query is required");
|
||||
public MongoCursorItemReader<T> build() {
|
||||
Assert.notNull(this.template, "template is required.");
|
||||
if (this.saveState) {
|
||||
Assert.hasText(this.name, "A name is required when saveState is set to true");
|
||||
}
|
||||
Assert.notNull(this.targetType, "targetType is required.");
|
||||
Assert.state(StringUtils.hasText(this.jsonQuery) || this.query != null, "A query is required");
|
||||
|
||||
if (StringUtils.hasText(this.jsonQuery) || this.query != null) {
|
||||
Assert.notNull(this.sorts, "sorts map is required.");
|
||||
}
|
||||
if (StringUtils.hasText(this.jsonQuery) || this.query != null) {
|
||||
Assert.notNull(this.sorts, "sorts map is required.");
|
||||
}
|
||||
|
||||
MongoCursorItemReader<T> reader = new MongoCursorItemReader<>();
|
||||
reader.setSaveState(this.saveState);
|
||||
reader.setName(this.name);
|
||||
reader.setCurrentItemCount(this.currentItemCount);
|
||||
reader.setMaxItemCount(this.maxItemCount);
|
||||
MongoCursorItemReader<T> reader = new MongoCursorItemReader<>();
|
||||
reader.setSaveState(this.saveState);
|
||||
reader.setName(this.name);
|
||||
reader.setCurrentItemCount(this.currentItemCount);
|
||||
reader.setMaxItemCount(this.maxItemCount);
|
||||
|
||||
reader.setTemplate(this.template);
|
||||
reader.setTargetType(this.targetType);
|
||||
reader.setCollection(this.collection);
|
||||
reader.setQuery(this.query);
|
||||
reader.setQuery(this.jsonQuery);
|
||||
reader.setParameterValues(this.parameterValues);
|
||||
reader.setFields(this.fields);
|
||||
reader.setSort(this.sorts);
|
||||
reader.setHint(this.hint);
|
||||
reader.setBatchSize(this.batchSize);
|
||||
reader.setLimit(this.limit);
|
||||
reader.setMaxTimeMs(this.maxTimeMs);
|
||||
reader.setTemplate(this.template);
|
||||
reader.setTargetType(this.targetType);
|
||||
reader.setCollection(this.collection);
|
||||
reader.setQuery(this.query);
|
||||
reader.setQuery(this.jsonQuery);
|
||||
reader.setParameterValues(this.parameterValues);
|
||||
reader.setFields(this.fields);
|
||||
reader.setSort(this.sorts);
|
||||
reader.setHint(this.hint);
|
||||
reader.setBatchSize(this.batchSize);
|
||||
reader.setLimit(this.limit);
|
||||
if (this.maxTime != null) {
|
||||
reader.setMaxTime(this.maxTime);
|
||||
}
|
||||
|
||||
return reader;
|
||||
}
|
||||
|
||||
return reader;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2023 the original author or authors.
|
||||
* Copyright 2023 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.
|
||||
@@ -15,6 +15,7 @@
|
||||
*/
|
||||
package org.springframework.batch.item.data;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
@@ -37,274 +38,277 @@ import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* Test class for {@link MongoCursorItemReader}.
|
||||
*
|
||||
* @author LEE Juchan
|
||||
* @since 5.0
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class MongoCursorItemReaderTest {
|
||||
|
||||
private MongoCursorItemReader<String> reader;
|
||||
private MongoCursorItemReader<String> reader;
|
||||
|
||||
@Mock
|
||||
private MongoTemplate template;
|
||||
@Mock
|
||||
private MongoTemplate template;
|
||||
|
||||
private Map<String, Sort.Direction> sortOptions;
|
||||
private Map<String, Sort.Direction> sortOptions;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
|
||||
sortOptions = new HashMap<>();
|
||||
sortOptions.put("name", Sort.Direction.DESC);
|
||||
sortOptions = new HashMap<>();
|
||||
sortOptions.put("name", Sort.Direction.DESC);
|
||||
|
||||
reader.setTemplate(template);
|
||||
reader.setTargetType(String.class);
|
||||
reader.setQuery("{ }");
|
||||
reader.setSort(sortOptions);
|
||||
reader.afterPropertiesSet();
|
||||
}
|
||||
reader.setTemplate(template);
|
||||
reader.setTargetType(String.class);
|
||||
reader.setQuery("{ }");
|
||||
reader.setSort(sortOptions);
|
||||
reader.afterPropertiesSet();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testAfterPropertiesSetForQueryString() {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
Exception exception = assertThrows(IllegalStateException.class, reader::afterPropertiesSet);
|
||||
assertEquals("An implementation of MongoOperations is required.", exception.getMessage());
|
||||
@Test
|
||||
void testAfterPropertiesSetForQueryString() {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
Exception exception = assertThrows(IllegalStateException.class, reader::afterPropertiesSet);
|
||||
assertEquals("An implementation of MongoOperations is required.", exception.getMessage());
|
||||
|
||||
reader.setTemplate(template);
|
||||
reader.setTemplate(template);
|
||||
|
||||
exception = assertThrows(IllegalStateException.class, reader::afterPropertiesSet);
|
||||
assertEquals("A type to convert the input into is required.", exception.getMessage());
|
||||
exception = assertThrows(IllegalStateException.class, reader::afterPropertiesSet);
|
||||
assertEquals("A targetType to convert the input into is required.", exception.getMessage());
|
||||
|
||||
reader.setTargetType(String.class);
|
||||
reader.setTargetType(String.class);
|
||||
|
||||
exception = assertThrows(IllegalStateException.class, reader::afterPropertiesSet);
|
||||
assertEquals("A query is required.", exception.getMessage());
|
||||
exception = assertThrows(IllegalStateException.class, reader::afterPropertiesSet);
|
||||
assertEquals("A query is required.", exception.getMessage());
|
||||
|
||||
reader.setQuery("");
|
||||
reader.setQuery("");
|
||||
|
||||
exception = assertThrows(IllegalStateException.class, reader::afterPropertiesSet);
|
||||
assertEquals("A sort is required.", exception.getMessage());
|
||||
exception = assertThrows(IllegalStateException.class, reader::afterPropertiesSet);
|
||||
assertEquals("A sort is required.", exception.getMessage());
|
||||
|
||||
reader.setSort(sortOptions);
|
||||
reader.afterPropertiesSet();
|
||||
}
|
||||
reader.setSort(sortOptions);
|
||||
reader.afterPropertiesSet();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testAfterPropertiesSetForQueryObject() {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
@Test
|
||||
void testAfterPropertiesSetForQueryObject() {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
|
||||
reader.setTemplate(template);
|
||||
reader.setTargetType(String.class);
|
||||
reader.setTemplate(template);
|
||||
reader.setTargetType(String.class);
|
||||
|
||||
Query query = new Query().with(Sort.by(new Sort.Order(Sort.Direction.ASC, "_id")));
|
||||
reader.setQuery(query);
|
||||
Query query = new Query().with(Sort.by(new Sort.Order(Sort.Direction.ASC, "_id")));
|
||||
reader.setQuery(query);
|
||||
|
||||
reader.afterPropertiesSet();
|
||||
}
|
||||
reader.afterPropertiesSet();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testBasicQuery() throws Exception {
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
@Test
|
||||
void testBasicQuery() throws Exception {
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of("hello world"));
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of("hello world"));
|
||||
|
||||
reader.doOpen();
|
||||
assertEquals(reader.doRead(), "hello world");
|
||||
reader.doOpen();
|
||||
assertEquals(reader.doRead(), "hello world");
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
}
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryWithFields() throws Exception {
|
||||
reader.setFields("{name : 1, age : 1, _id: 0}");
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
@Test
|
||||
void testQueryWithFields() throws Exception {
|
||||
reader.setFields("{name : 1, age : 1, _id: 0}");
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals(1, query.getFieldsObject().get("name"));
|
||||
assertEquals(1, query.getFieldsObject().get("age"));
|
||||
assertEquals(0, query.getFieldsObject().get("_id"));
|
||||
}
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals(1, query.getFieldsObject().get("name"));
|
||||
assertEquals(1, query.getFieldsObject().get("age"));
|
||||
assertEquals(0, query.getFieldsObject().get("_id"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryWithHint() throws Exception {
|
||||
reader.setHint("{ $natural : 1}");
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
@Test
|
||||
void testQueryWithHint() throws Exception {
|
||||
reader.setHint("{ $natural : 1}");
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals("{ $natural : 1}", query.getHint());
|
||||
}
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals("{ $natural : 1}", query.getHint());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryWithParameters() throws Exception {
|
||||
reader.setParameterValues(Collections.singletonList("foo"));
|
||||
@Test
|
||||
void testQueryWithParameters() throws Exception {
|
||||
reader.setParameterValues(Collections.singletonList("foo"));
|
||||
|
||||
reader.setQuery("{ name : ?0 }");
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
reader.setQuery("{ name : ?0 }");
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{\"name\": \"foo\"}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
}
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{\"name\": \"foo\"}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryWithBatchSize() throws Exception {
|
||||
reader.setBatchSize(50);
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
@Test
|
||||
void testQueryWithBatchSize() throws Exception {
|
||||
reader.setBatchSize(50);
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals(50, query.getMeta().getCursorBatchSize());
|
||||
}
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals(50, query.getMeta().getCursorBatchSize());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryWithLimit() throws Exception {
|
||||
reader.setLimit(200);
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
@Test
|
||||
void testQueryWithLimit() throws Exception {
|
||||
reader.setLimit(200);
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals(200, query.getLimit());
|
||||
}
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals(200, query.getLimit());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryWithMaxTime() throws Exception {
|
||||
reader.setMaxTimeMs(3000);
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
@Test
|
||||
void testQueryWithMaxTime() throws Exception {
|
||||
reader.setMaxTime(Duration.ofMillis(3000));
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals(3000, query.getMeta().getMaxTimeMsec());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryWithCollection() throws Exception {
|
||||
reader.setParameterValues(Collections.singletonList("foo"));
|
||||
|
||||
reader.setQuery("{ name : ?0 }");
|
||||
reader.setCollection("collection");
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
ArgumentCaptor<String> collectionContainer = ArgumentCaptor.forClass(String.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class), collectionContainer.capture()))
|
||||
.thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{\"name\": \"foo\"}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals("collection", collectionContainer.getValue());
|
||||
}
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals(3000, query.getMeta().getMaxTimeMsec());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryWithCollection() throws Exception {
|
||||
reader.setParameterValues(Collections.singletonList("foo"));
|
||||
|
||||
reader.setQuery("{ name : ?0 }");
|
||||
reader.setCollection("collection");
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
ArgumentCaptor<String> collectionContainer = ArgumentCaptor.forClass(String.class);
|
||||
|
||||
when(template.stream(queryContainer.capture(), eq(String.class), collectionContainer.capture()))
|
||||
.thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query query = queryContainer.getValue();
|
||||
assertEquals("{\"name\": \"foo\"}", query.getQueryObject().toJson());
|
||||
assertEquals("{\"name\": -1}", query.getSortObject().toJson());
|
||||
assertEquals("collection", collectionContainer.getValue());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryObject() throws Exception {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
reader.setTemplate(template);
|
||||
|
||||
@Test
|
||||
void testQueryObject() throws Exception {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
reader.setTemplate(template);
|
||||
Query query = new Query().with(Sort.by(new Sort.Order(Sort.Direction.ASC, "_id")));
|
||||
reader.setQuery(query);
|
||||
reader.setTargetType(String.class);
|
||||
|
||||
reader.afterPropertiesSet();
|
||||
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
|
||||
Query query = new Query().with(Sort.by(new Sort.Order(Sort.Direction.ASC, "_id")));
|
||||
reader.setQuery(query);
|
||||
reader.setTargetType(String.class);
|
||||
|
||||
reader.afterPropertiesSet();
|
||||
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
when(template.stream(queryContainer.capture(), eq(String.class))).thenReturn(Stream.of());
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query actualQuery = queryContainer.getValue();
|
||||
assertEquals("{}", actualQuery.getQueryObject().toJson());
|
||||
assertEquals("{\"_id\": 1}", actualQuery.getSortObject().toJson());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryObjectWithCollection() throws Exception {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
reader.setTemplate(template);
|
||||
|
||||
Query query = new Query().with(Sort.by(new Sort.Order(Sort.Direction.ASC, "_id")));
|
||||
reader.setQuery(query);
|
||||
reader.setTargetType(String.class);
|
||||
reader.setCollection("collection");
|
||||
|
||||
reader.afterPropertiesSet();
|
||||
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
ArgumentCaptor<String> stringContainer = ArgumentCaptor.forClass(String.class);
|
||||
when(template.stream(queryContainer.capture(), eq(String.class), stringContainer.capture())).thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query actualQuery = queryContainer.getValue();
|
||||
assertEquals("{}", actualQuery.getQueryObject().toJson());
|
||||
assertEquals("{\"_id\": 1}", actualQuery.getSortObject().toJson());
|
||||
assertEquals("collection", stringContainer.getValue());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testSortThrowsExceptionWhenInvokedWithNull() {
|
||||
// given
|
||||
reader = new MongoCursorItemReader<>();
|
||||
|
||||
// when + then
|
||||
assertThatIllegalArgumentException().isThrownBy(() -> reader.setSort(null))
|
||||
.withMessage("Sorts must not be null");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCursorRead() throws Exception {
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
when(template.stream(queryContainer.capture(), eq(String.class)))
|
||||
.thenReturn(Stream.of("first", "second", "third"));
|
||||
|
||||
reader.doOpen();
|
||||
|
||||
assertEquals("first", reader.doRead());
|
||||
assertEquals("second", reader.doRead());
|
||||
assertEquals("third", reader.doRead());
|
||||
assertNull(reader.doRead());
|
||||
}
|
||||
Query actualQuery = queryContainer.getValue();
|
||||
assertEquals("{}", actualQuery.getQueryObject().toJson());
|
||||
assertEquals("{\"_id\": 1}", actualQuery.getSortObject().toJson());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQueryObjectWithCollection() throws Exception {
|
||||
reader = new MongoCursorItemReader<>();
|
||||
reader.setTemplate(template);
|
||||
|
||||
Query query = new Query().with(Sort.by(new Sort.Order(Sort.Direction.ASC, "_id")));
|
||||
reader.setQuery(query);
|
||||
reader.setTargetType(String.class);
|
||||
reader.setCollection("collection");
|
||||
|
||||
reader.afterPropertiesSet();
|
||||
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
ArgumentCaptor<String> stringContainer = ArgumentCaptor.forClass(String.class);
|
||||
when(template.stream(queryContainer.capture(), eq(String.class), stringContainer.capture()))
|
||||
.thenReturn(Stream.of());
|
||||
|
||||
reader.doOpen();
|
||||
assertNull(reader.doRead());
|
||||
|
||||
Query actualQuery = queryContainer.getValue();
|
||||
assertEquals("{}", actualQuery.getQueryObject().toJson());
|
||||
assertEquals("{\"_id\": 1}", actualQuery.getSortObject().toJson());
|
||||
assertEquals("collection", stringContainer.getValue());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testSortThrowsExceptionWhenInvokedWithNull() {
|
||||
// given
|
||||
reader = new MongoCursorItemReader<>();
|
||||
|
||||
// when + then
|
||||
assertThatIllegalArgumentException().isThrownBy(() -> reader.setSort(null))
|
||||
.withMessage("Sorts must not be null");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCursorRead() throws Exception {
|
||||
ArgumentCaptor<Query> queryContainer = ArgumentCaptor.forClass(Query.class);
|
||||
when(template.stream(queryContainer.capture(), eq(String.class)))
|
||||
.thenReturn(Stream.of("first", "second", "third"));
|
||||
|
||||
reader.doOpen();
|
||||
|
||||
assertEquals("first", reader.doRead());
|
||||
assertEquals("second", reader.doRead());
|
||||
assertEquals("third", reader.doRead());
|
||||
assertNull(reader.doRead());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
/*
|
||||
* Copyright 2023 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
|
||||
*
|
||||
* https://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.builder;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.batch.item.data.MongoCursorItemReader;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import org.springframework.data.mongodb.core.MongoTemplate;
|
||||
import org.springframework.data.mongodb.core.query.Query;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
/**
|
||||
* Test class for {@link MongoCursorItemReaderBuilder}.
|
||||
*
|
||||
* @author Mahmoud Ben Hassine
|
||||
*/
|
||||
public class MongoCursorItemReaderBuilderTests {
|
||||
|
||||
@Test
|
||||
void testBuild() {
|
||||
// given
|
||||
MongoTemplate template = mock();
|
||||
Class<String> targetType = String.class;
|
||||
Query query = mock();
|
||||
Map<String, Sort.Direction> sorts = mock();
|
||||
int batchSize = 100;
|
||||
int limit = 10000;
|
||||
Duration maxTime = Duration.ofMillis(1000);
|
||||
|
||||
// when
|
||||
MongoCursorItemReader<String> reader = new MongoCursorItemReaderBuilder<String>().name("reader")
|
||||
.template(template)
|
||||
.targetType(targetType)
|
||||
.query(query)
|
||||
.sorts(sorts)
|
||||
.batchSize(batchSize)
|
||||
.limit(limit)
|
||||
.maxTime(maxTime)
|
||||
.build();
|
||||
|
||||
// then
|
||||
Assertions.assertEquals(template, ReflectionTestUtils.getField(reader, "template"));
|
||||
Assertions.assertEquals(targetType, ReflectionTestUtils.getField(reader, "targetType"));
|
||||
Assertions.assertEquals(query, ReflectionTestUtils.getField(reader, "query"));
|
||||
Assertions.assertEquals(batchSize, ReflectionTestUtils.getField(reader, "batchSize"));
|
||||
Assertions.assertEquals(limit, ReflectionTestUtils.getField(reader, "limit"));
|
||||
Assertions.assertEquals(maxTime, ReflectionTestUtils.getField(reader, "maxTime"));
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user