diff --git a/src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperation.java b/src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperation.java index 2fc6e428..5078d701 100644 --- a/src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperation.java +++ b/src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperation.java @@ -126,7 +126,7 @@ public interface ExecutableFindByQueryOperation { * * @param scanConsistency the custom scan consistency to use for this query. */ - FindByQueryWithQuery consistentWith(QueryScanConsistency scanConsistency); + FindByQueryConsistentWith consistentWith(QueryScanConsistency scanConsistency); } diff --git a/src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperationSupport.java b/src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperationSupport.java index 46859bd8..7f569501 100644 --- a/src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperationSupport.java +++ b/src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperationSupport.java @@ -21,6 +21,7 @@ import java.util.stream.Stream; import org.springframework.data.couchbase.core.query.Query; import com.couchbase.client.java.query.QueryScanConsistency; +import org.springframework.data.couchbase.core.ReactiveFindByQueryOperationSupport.ReactiveFindByQuerySupport; public class ExecutableFindByQueryOperationSupport implements ExecutableFindByQueryOperation { @@ -50,8 +51,7 @@ public class ExecutableFindByQueryOperationSupport implements ExecutableFindByQu this.template = template; this.domainType = domainType; this.query = query; - this.reactiveSupport = new ReactiveFindByQueryOperationSupport.ReactiveFindByQuerySupport( - template.reactive(), domainType, query, scanConsistency); + this.reactiveSupport = new ReactiveFindByQuerySupport(template.reactive(), domainType, query, scanConsistency); this.scanConsistency = scanConsistency; } @@ -72,11 +72,17 @@ public class ExecutableFindByQueryOperationSupport implements ExecutableFindByQu @Override public TerminatingFindByQuery matching(final Query query) { - return new ExecutableFindByQuerySupport<>(template, domainType, query, scanConsistency); + QueryScanConsistency scanCons; + if (query.getScanConsistency() != null) { + scanCons = query.getScanConsistency(); + } else { + scanCons = scanConsistency; + } + return new ExecutableFindByQuerySupport<>(template, domainType, query, scanCons); } @Override - public FindByQueryWithQuery consistentWith(final QueryScanConsistency scanConsistency) { + public FindByQueryConsistentWith consistentWith(final QueryScanConsistency scanConsistency) { return new ExecutableFindByQuerySupport<>(template, domainType, query, scanConsistency); } diff --git a/src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperation.java b/src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperation.java index 50e333a9..309431f9 100644 --- a/src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperation.java +++ b/src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperation.java @@ -100,7 +100,7 @@ public interface ReactiveFindByQueryOperation { * * @param scanConsistency the custom scan consistency to use for this query. */ - FindByQueryWithQuery consistentWith(QueryScanConsistency scanConsistency); + FindByQueryConsistentWith consistentWith(QueryScanConsistency scanConsistency); } diff --git a/src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperationSupport.java b/src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperationSupport.java index 37777ed0..a538be87 100644 --- a/src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperationSupport.java +++ b/src/main/java/org/springframework/data/couchbase/core/ReactiveFindByQueryOperationSupport.java @@ -60,11 +60,17 @@ public class ReactiveFindByQueryOperationSupport implements ReactiveFindByQueryO @Override public TerminatingFindByQuery matching(Query query) { - return new ReactiveFindByQuerySupport<>(template, domainType, query, scanConsistency); + QueryScanConsistency scanCons; + if (query.getScanConsistency() != null) { + scanCons = query.getScanConsistency(); + } else { + scanCons = scanConsistency; + } + return new ReactiveFindByQuerySupport<>(template, domainType, query, scanCons); } @Override - public FindByQueryWithQuery consistentWith(QueryScanConsistency scanConsistency) { + public FindByQueryConsistentWith consistentWith(QueryScanConsistency scanConsistency) { return new ReactiveFindByQuerySupport<>(template, domainType, query, scanConsistency); } diff --git a/src/main/java/org/springframework/data/couchbase/core/ReactiveInsertByIdOperationSupport.java b/src/main/java/org/springframework/data/couchbase/core/ReactiveInsertByIdOperationSupport.java index c977511f..c76acf37 100644 --- a/src/main/java/org/springframework/data/couchbase/core/ReactiveInsertByIdOperationSupport.java +++ b/src/main/java/org/springframework/data/couchbase/core/ReactiveInsertByIdOperationSupport.java @@ -15,7 +15,6 @@ */ package org.springframework.data.couchbase.core; -import com.couchbase.client.java.kv.UpsertOptions; import org.springframework.data.couchbase.core.mapping.Document; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -98,7 +97,7 @@ public class ReactiveInsertByIdOperationSupport implements ReactiveInsertByIdOpe } else if (durabilityLevel != DurabilityLevel.NONE) { options.durability(durabilityLevel); } - if (expiry != null) { + if (expiry != null && ! expiry.isZero()) { options.expiry(expiry); } else if (domainType.isAnnotationPresent(Document.class)) { Document documentAnn = domainType.getAnnotation(Document.class); diff --git a/src/main/java/org/springframework/data/couchbase/core/ReactiveUpsertByIdOperationSupport.java b/src/main/java/org/springframework/data/couchbase/core/ReactiveUpsertByIdOperationSupport.java index bb07b2c7..9dee8776 100644 --- a/src/main/java/org/springframework/data/couchbase/core/ReactiveUpsertByIdOperationSupport.java +++ b/src/main/java/org/springframework/data/couchbase/core/ReactiveUpsertByIdOperationSupport.java @@ -15,7 +15,6 @@ */ package org.springframework.data.couchbase.core; -import com.couchbase.client.java.kv.Expiry; import org.springframework.data.couchbase.core.mapping.Document; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -98,7 +97,7 @@ public class ReactiveUpsertByIdOperationSupport implements ReactiveUpsertByIdOpe } else if (durabilityLevel != DurabilityLevel.NONE) { options.durability(durabilityLevel); } - if (expiry != null) { + if (expiry != null && !expiry.isZero()) { options.expiry(expiry); } else if (domainType.isAnnotationPresent(Document.class)) { Document documentAnn = domainType.getAnnotation(Document.class); diff --git a/src/main/java/org/springframework/data/couchbase/core/query/Query.java b/src/main/java/org/springframework/data/couchbase/core/query/Query.java index 6d699d40..b2a0b960 100644 --- a/src/main/java/org/springframework/data/couchbase/core/query/Query.java +++ b/src/main/java/org/springframework/data/couchbase/core/query/Query.java @@ -47,6 +47,7 @@ public class Query { private long skip; private int limit; private Sort sort = Sort.unsorted(); + private QueryScanConsistency queryScanConsistency; static private final Pattern WHERE_PATTERN = Pattern.compile("\\sWHERE\\s"); @@ -127,6 +128,27 @@ public class Query { return with(pageable.getSort()); } + /** + * queryScanConsistency + * + * @return queryScanConsistency + */ + public QueryScanConsistency getScanConsistency() { + return queryScanConsistency; + } + + + /** + * Sets the given scan consistency on the {@link Query} instance. + * + * @param queryScanConsistency + * @return this + */ + public Query scanConsistency(final QueryScanConsistency queryScanConsistency) { + this.queryScanConsistency = queryScanConsistency; + return this; + } + /** * Adds a {@link Sort} to the {@link Query} instance. * @@ -280,7 +302,7 @@ public class Query { } /** - * build QueryOptions forom parameters and scanConsistency + * build QueryOptions from parameters and scanConsistency * * @param scanConsistency * @return QueryOptions diff --git a/src/main/java/org/springframework/data/couchbase/repository/CouchbaseRepository.java b/src/main/java/org/springframework/data/couchbase/repository/CouchbaseRepository.java index aab74e92..5544d7e5 100644 --- a/src/main/java/org/springframework/data/couchbase/repository/CouchbaseRepository.java +++ b/src/main/java/org/springframework/data/couchbase/repository/CouchbaseRepository.java @@ -18,6 +18,7 @@ package org.springframework.data.couchbase.repository; import java.util.List; +import com.couchbase.client.java.query.QueryScanConsistency; import org.springframework.data.domain.Sort; import org.springframework.data.repository.NoRepositoryBean; import org.springframework.data.repository.PagingAndSortingRepository; @@ -34,6 +35,8 @@ public interface CouchbaseRepository extends PagingAndSortingRepository findAll(Sort sort); + List findAll(QueryScanConsistency queryScanConsistency); + @Override List findAll(); diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/CouchbaseQueryMethod.java b/src/main/java/org/springframework/data/couchbase/repository/query/CouchbaseQueryMethod.java index 43223010..d230572a 100644 --- a/src/main/java/org/springframework/data/couchbase/repository/query/CouchbaseQueryMethod.java +++ b/src/main/java/org/springframework/data/couchbase/repository/query/CouchbaseQueryMethod.java @@ -25,6 +25,7 @@ import org.springframework.data.couchbase.core.query.Dimensional; import org.springframework.data.couchbase.core.query.View; import org.springframework.data.couchbase.core.query.WithConsistency; import org.springframework.data.couchbase.repository.Query; +import org.springframework.data.couchbase.repository.ScanConsistency; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.projection.ProjectionFactory; import org.springframework.data.repository.core.RepositoryMetadata; @@ -152,6 +153,24 @@ public class CouchbaseQueryMethod extends QueryMethod { return method.getAnnotation(WithConsistency.class); } + /** + * If the method has a @ScanConsistency annotation + * + * @return true if this has the @ScanConsistency annotation + */ + public boolean hasScanConsistencyAnnotation() { + return getScanConsistencyAnnotation() != null; + } + + /** + * ScanConsistency annotation + * + * @return the @ScanConsistency annotation + */ + public ScanConsistency getScanConsistencyAnnotation() { + return method.getAnnotation(ScanConsistency.class); + } + /** * Returns the query string declared in a {@link Query} annotation or {@literal null} if neither the annotation found * nor the attribute was specified. diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/N1qlRepositoryQueryExecutor.java b/src/main/java/org/springframework/data/couchbase/repository/query/N1qlRepositoryQueryExecutor.java index d9d3eed3..9f19847c 100644 --- a/src/main/java/org/springframework/data/couchbase/repository/query/N1qlRepositoryQueryExecutor.java +++ b/src/main/java/org/springframework/data/couchbase/repository/query/N1qlRepositoryQueryExecutor.java @@ -15,9 +15,7 @@ */ package org.springframework.data.couchbase.repository.query; -import java.util.ArrayList; -import java.util.List; - +import com.couchbase.client.java.query.QueryScanConsistency; import org.springframework.data.couchbase.core.CouchbaseOperations; import org.springframework.data.couchbase.core.ExecutableFindByQueryOperation; import org.springframework.data.couchbase.core.query.Query; @@ -65,7 +63,8 @@ public class N1qlRepositoryQueryExecutor { final PartTree tree = new PartTree(queryMethod.getName(), domainClass); query = new N1qlQueryCreator(tree, accessor, queryMethod, operations.getConverter()).createQuery(); } - q = (ExecutableFindByQueryOperation.ExecutableFindByQuery) operations.findByQuery(domainClass).matching(query); + q = (ExecutableFindByQueryOperation.ExecutableFindByQuery) operations.findByQuery(domainClass) + .consistentWith(buildQueryScanConsistency()).matching(query); if (queryMethod.isCountQuery()) { return q.count(); } else if (queryMethod.isCollectionQuery()) { @@ -76,4 +75,14 @@ public class N1qlRepositoryQueryExecutor { } + private QueryScanConsistency buildQueryScanConsistency() { + QueryScanConsistency scanConsistency = QueryScanConsistency.NOT_BOUNDED; + if (queryMethod.hasConsistencyAnnotation()) { + scanConsistency = queryMethod.getConsistencyAnnotation().value(); + } else if (queryMethod.hasScanConsistencyAnnotation()) { + scanConsistency = queryMethod.getScanConsistencyAnnotation().query(); + } + return scanConsistency; + } + } diff --git a/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveN1qlRepositoryQueryExecutor.java b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveN1qlRepositoryQueryExecutor.java index 2a540980..bcc11eeb 100644 --- a/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveN1qlRepositoryQueryExecutor.java +++ b/src/main/java/org/springframework/data/couchbase/repository/query/ReactiveN1qlRepositoryQueryExecutor.java @@ -15,6 +15,7 @@ */ package org.springframework.data.couchbase.repository.query; +import com.couchbase.client.java.query.QueryScanConsistency; import org.springframework.data.couchbase.core.ExecutableFindByQueryOperation; import org.springframework.data.couchbase.core.ReactiveFindByQueryOperation; import org.springframework.data.mapping.context.MappingContext; @@ -69,7 +70,8 @@ public class ReactiveN1qlRepositoryQueryExecutor { final PartTree tree = new PartTree(queryMethod.getName(), domainClass); query = new N1qlQueryCreator(tree, accessor, queryMethod, operations.getConverter()).createQuery(); } - q = (ReactiveFindByQueryOperation.ReactiveFindByQuery) operations.findByQuery(domainClass).matching(query); + q = (ReactiveFindByQueryOperation.ReactiveFindByQuery) operations.findByQuery(domainClass) + .consistentWith(buildQueryScanConsistency()).matching(query); if (queryMethod.isCountQuery()) { return q.count(); } else if (queryMethod.isCollectionQuery()) { @@ -79,4 +81,14 @@ public class ReactiveN1qlRepositoryQueryExecutor { } } + private QueryScanConsistency buildQueryScanConsistency() { + QueryScanConsistency scanConsistency = QueryScanConsistency.NOT_BOUNDED; + if (queryMethod.hasConsistencyAnnotation()) { + scanConsistency = queryMethod.getConsistencyAnnotation().value(); + } else if (queryMethod.hasScanConsistencyAnnotation()) { + scanConsistency = queryMethod.getScanConsistencyAnnotation().query(); + } + return scanConsistency; + } + } diff --git a/src/main/java/org/springframework/data/couchbase/repository/support/SimpleCouchbaseRepository.java b/src/main/java/org/springframework/data/couchbase/repository/support/SimpleCouchbaseRepository.java index f6b7ee7c..94e336a5 100644 --- a/src/main/java/org/springframework/data/couchbase/repository/support/SimpleCouchbaseRepository.java +++ b/src/main/java/org/springframework/data/couchbase/repository/support/SimpleCouchbaseRepository.java @@ -147,6 +147,11 @@ public class SimpleCouchbaseRepository implements CouchbaseRepository findAll(final QueryScanConsistency queryScanConsistency) { + return findAll(new Query().scanConsistency(queryScanConsistency)); + } + @Override public Page findAll(final Pageable pageable) { List results = findAll(new Query().with(pageable)); diff --git a/src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java b/src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java index 1ddf6103..9f0aa782 100644 --- a/src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java +++ b/src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java @@ -55,6 +55,7 @@ class CouchbaseTemplateKeyValueIntegrationTests extends ClusterAwareIntegrationT private static CouchbaseClientFactory couchbaseClientFactory; private CouchbaseTemplate couchbaseTemplate; + private ReactiveCouchbaseTemplate reactiveCouchbaseTemplate; @BeforeAll static void beforeAll() { @@ -71,6 +72,7 @@ class CouchbaseTemplateKeyValueIntegrationTests extends ClusterAwareIntegrationT void beforeEach() { ApplicationContext ac = new AnnotationConfigApplicationContext(Config.class); couchbaseTemplate = (CouchbaseTemplate) ac.getBean(COUCHBASE_TEMPLATE); + reactiveCouchbaseTemplate = (ReactiveCouchbaseTemplate) ac.getBean(REACTIVE_COUCHBASE_TEMPLATE); } @Test @@ -89,6 +91,7 @@ class CouchbaseTemplateKeyValueIntegrationTests extends ClusterAwareIntegrationT assertEquals(user, found); couchbaseTemplate.removeById().one(user.getId()); + reactiveCouchbaseTemplate.replaceById(User.class).withDurability(PersistTo.ACTIVE, ReplicateTo.THREE).one(user); } @Test diff --git a/src/test/java/org/springframework/data/couchbase/domain/AirportRepository.java b/src/test/java/org/springframework/data/couchbase/domain/AirportRepository.java index 06ce1fec..a441f009 100644 --- a/src/test/java/org/springframework/data/couchbase/domain/AirportRepository.java +++ b/src/test/java/org/springframework/data/couchbase/domain/AirportRepository.java @@ -40,15 +40,19 @@ public interface AirportRepository extends PagingAndSortingRepository findAll(); @Override - @ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS) Airport save(Airport airport); @ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS) List findAllByIata(String iata); + @ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS) + Airport findByIata(String iata); + @Query("#{#n1ql.selectEntity} where iata = $1") + @ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS) List getAllByIata(String iata); + @ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS) long countByIataIn(String... iata); @ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS) diff --git a/src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java b/src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java index 54f56573..5d707cf1 100644 --- a/src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java +++ b/src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java @@ -16,6 +16,7 @@ package org.springframework.data.couchbase.domain; +import org.springframework.data.couchbase.repository.Query; import reactor.core.publisher.Flux; import org.springframework.data.couchbase.repository.ScanConsistency; @@ -44,6 +45,10 @@ public interface ReactiveAirportRepository extends ReactiveSortingRepository findAllByIata(String iata); + @Query("#{#n1ql.selectEntity} where iata = $1") + @ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS) + Flux getAllByIata(String iata); + @ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS) Mono countByIataIn(String... iatas); diff --git a/src/test/java/org/springframework/data/couchbase/repository/CouchbaseRepositoryQueryIntegrationTests.java b/src/test/java/org/springframework/data/couchbase/repository/CouchbaseRepositoryQueryIntegrationTests.java index c4ea6cd8..48218706 100644 --- a/src/test/java/org/springframework/data/couchbase/repository/CouchbaseRepositoryQueryIntegrationTests.java +++ b/src/test/java/org/springframework/data/couchbase/repository/CouchbaseRepositoryQueryIntegrationTests.java @@ -25,8 +25,6 @@ import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import java.util.stream.Collectors; -import java.util.stream.StreamSupport; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -93,7 +91,6 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr airportRepository.save(vie); xxx = new Airport("airports::xxx", "xxx", "xxxx"); airportRepository.save(xxx); - sleep(1000); List airports; airports = airportRepository.findAllByIata("1\" or iata=iata or iata=\"1"); assertEquals(0, airports.size()); @@ -112,7 +109,6 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr try { vie = new Airport("airports::vie", "vie", "loww"); airportRepository.save(vie); - sleep(1000); List airports = airportRepository.findAllByIata("vie"); assertEquals(vie.getId(), airports.get(0).getId()); } finally { @@ -170,7 +166,7 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr iatas[i].toLowerCase(Locale.ROOT) /* lcao */); airportRepository.save(airport); } - sleep(1000); + for (int k = 0; k < 50; k++) { Callable[] suppliers = new Callable[iatas.length]; for (int i = 0; i < iatas.length; i++) { @@ -211,7 +207,7 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr Airport airport = new Airport("airports::" + iatas[i], iatas[i] /*iata*/, iatas[i].toLowerCase() /* lcao */); airportRepository.save(airport); } - sleep(1000); + for (int k = 0; k < 100; k++) { Callable[] suppliers = new Callable[iatas.length]; for (int i = 0; i < iatas.length; i++) { diff --git a/src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java b/src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java index dd04aeb9..75dd831a 100644 --- a/src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java +++ b/src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java @@ -88,9 +88,10 @@ public class ReactiveCouchbaseRepositoryQueryIntegrationTests extends ClusterAwa try { vie = new Airport("airports::vie", "vie", "loww"); airportRepository.save(vie).block(); - List airports = airportRepository.findAllByIata("vie").collectList().block(); - // TODO - System.err.println(airports); + List airports1 = airportRepository.findAllByIata("vie").collectList().block(); + assertEquals(1,airports1.size()); + List airports2 = airportRepository.findAllByIata("vie").collectList().block(); + assertEquals(1,airports2.size()); } finally { airportRepository.delete(vie).block(); }