DATACOUCH-504 - Make sure reactive repositories can be used

This changeset fixes an issue where while blocking repositories would
work properly, the reactive repositories won't. This has been an
oversight in the transition before RC1.

src/main/java/org/springframework/data/couchbase/config/AbstractCouchbaseConfiguration.java

@@ -38,6 +38,7 @@ import org.springframework.data.couchbase.core.convert.translation.JacksonTransl
 import org.springframework.data.couchbase.core.convert.translation.TranslationService;
 import org.springframework.data.couchbase.core.mapping.CouchbaseMappingContext;
 import org.springframework.data.couchbase.core.mapping.Document;
+import org.springframework.data.couchbase.repository.config.ReactiveRepositoryOperationsMapping;
 import org.springframework.data.couchbase.repository.config.RepositoryOperationsMapping;
 import org.springframework.data.mapping.model.CamelCaseAbbreviatingFieldNamingStrategy;
 import org.springframework.data.mapping.model.FieldNamingStrategy;
@@ -116,7 +117,7 @@ public abstract class AbstractCouchbaseConfiguration {
 		return new ReactiveCouchbaseTemplate(couchbaseClientFactory, mappingCouchbaseConverter);
 	}
 
-	@Bean
+	@Bean(name = BeanNames.COUCHBASE_OPERATIONS_MAPPING)
 	public RepositoryOperationsMapping couchbaseRepositoryOperationsMapping(CouchbaseTemplate couchbaseTemplate) {
 		// create a base mapping that associates all repositories to the default template
 		RepositoryOperationsMapping baseMapping = new RepositoryOperationsMapping(couchbaseTemplate);
@@ -135,6 +136,25 @@ public abstract class AbstractCouchbaseConfiguration {
 		// NO_OP
 	}
 
+	@Bean(name = BeanNames.REACTIVE_COUCHBASE_OPERATIONS_MAPPING)
+	public ReactiveRepositoryOperationsMapping reactiveCouchbaseRepositoryOperationsMapping(ReactiveCouchbaseTemplate reactiveCouchbaseTemplate) {
+		// create a base mapping that associates all repositories to the default template
+		ReactiveRepositoryOperationsMapping baseMapping = new ReactiveRepositoryOperationsMapping(reactiveCouchbaseTemplate);
+		// let the user tune it
+		configureReactiveRepositoryOperationsMapping(baseMapping);
+		return baseMapping;
+	}
+
+	/**
+	 * In order to customize the mapping between repositories/entity types to couchbase templates, use the provided
+	 * mapping's api (eg. in order to have different buckets backing different repositories).
+	 *
+	 * @param mapping the default mapping (will associate all repositories to the default template).
+	 */
+	protected void configureReactiveRepositoryOperationsMapping(ReactiveRepositoryOperationsMapping mapping) {
+		// NO_OP
+	}
+
 	/**
 	 * Scans the mapping base package for classes annotated with {@link Document}.
 	 *

src/main/java/org/springframework/data/couchbase/repository/CouchbaseRepository.java

@@ -31,11 +31,6 @@ import org.springframework.data.repository.PagingAndSortingRepository;
 @NoRepositoryBean
 public interface CouchbaseRepository<T, ID> extends PagingAndSortingRepository<T, ID> {
 
-	/**
-	 * @return a reference to the underlying {@link CouchbaseOperations operation template}.
-	 */
-	CouchbaseOperations getCouchbaseOperations();
-
 	@Override
 	List<T> findAll(Sort sort);
 

src/main/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepository.java

@@ -15,8 +15,6 @@
  */
 package org.springframework.data.couchbase.repository;
 
-import org.springframework.data.couchbase.core.CouchbaseOperations;
-import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
 import org.springframework.data.repository.NoRepositoryBean;
 import org.springframework.data.repository.reactive.ReactiveSortingRepository;
 
@@ -26,8 +24,5 @@ import org.springframework.data.repository.reactive.ReactiveSortingRepository;
  */
 @NoRepositoryBean
 public interface ReactiveCouchbaseRepository<T, ID> extends ReactiveSortingRepository<T, ID> {
-	/**
-	 * @return a reference to the underlying {@link CouchbaseOperations operation template}.
-	 */
-	ReactiveCouchbaseOperations getReactiveCouchbaseOperations();
+
 }

src/main/java/org/springframework/data/couchbase/repository/config/EnableReactiveCouchbaseRepositories.java

@@ -47,8 +47,8 @@ public @interface EnableReactiveCouchbaseRepositories {
 
 	/**
 	 * Alias for the {@link #basePackages()} attribute. Allows for more concise annotation declarations e.g.:
-	 * {@code @EnableCouchbaseRepositories("org.my.pkg")} instead of
-	 * {@code @EnableCouchbaseRepositories(basePackages="org.my.pkg")}.
+	 * {@code @EnableReactiveCouchbaseRepositories("org.my.pkg")} instead of
+	 * {@code @EnableReactiveCouchbaseRepositories(basePackages="org.my.pkg")}.
 	 */
 	String[] value() default {};
 
@@ -119,6 +119,6 @@ public @interface EnableReactiveCouchbaseRepositories {
 	 *
 	 * @return
 	 */
-	String couchbaseTemplateRef() default BeanNames.COUCHBASE_TEMPLATE;
+	String couchbaseTemplateRef() default BeanNames.REACTIVE_COUCHBASE_TEMPLATE;
 
 }

src/main/java/org/springframework/data/couchbase/repository/config/ReactiveCouchbaseRepositoryConfigurationExtension.java

@@ -109,7 +109,7 @@ public class ReactiveCouchbaseRepositoryConfigurationExtension extends Repositor
 	public void postProcess(BeanDefinitionBuilder builder, AnnotationRepositoryConfigurationSource config) {
 
 		builder.addDependsOn(BeanNames.REACTIVE_COUCHBASE_OPERATIONS_MAPPING);
-		builder.addPropertyReference("couchbaseOperationsMapping", BeanNames.REACTIVE_COUCHBASE_OPERATIONS_MAPPING);
+		builder.addPropertyReference("reactiveCouchbaseOperationsMapping", BeanNames.REACTIVE_COUCHBASE_OPERATIONS_MAPPING);
 	}
 
 	/*

src/main/java/org/springframework/data/couchbase/repository/query/ReactiveCouchbaseRepositoryQuery.java

@@ -0,0 +1,27 @@
+package org.springframework.data.couchbase.repository.query;
+
+import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
+import org.springframework.data.repository.query.QueryMethod;
+import org.springframework.data.repository.query.RepositoryQuery;
+
+public class ReactiveCouchbaseRepositoryQuery implements RepositoryQuery {
+
+	private final ReactiveCouchbaseOperations operations;
+	private final QueryMethod queryMethod;
+
+	public ReactiveCouchbaseRepositoryQuery(final ReactiveCouchbaseOperations operations, final QueryMethod queryMethod) {
+		this.operations = operations;
+		this.queryMethod = queryMethod;
+	}
+
+	@Override
+	public Object execute(final Object[] parameters) {
+		return new ReactiveN1qlRepositoryQueryExecutor(operations, queryMethod).execute(parameters);
+	}
+
+	@Override
+	public QueryMethod getQueryMethod() {
+		return queryMethod;
+	}
+
+}

src/main/java/org/springframework/data/couchbase/repository/query/ReactiveN1qlRepositoryQueryExecutor.java

@@ -0,0 +1,35 @@
+package org.springframework.data.couchbase.repository.query;
+
+import java.util.List;
+
+import org.springframework.data.couchbase.core.CouchbaseOperations;
+import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
+import org.springframework.data.couchbase.core.query.Query;
+import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.data.repository.query.ParametersParameterAccessor;
+import org.springframework.data.repository.query.QueryMethod;
+import org.springframework.data.repository.query.parser.PartTree;
+import reactor.core.publisher.Flux;
+
+public class ReactiveN1qlRepositoryQueryExecutor {
+
+	private final ReactiveCouchbaseOperations operations;
+	private final QueryMethod queryMethod;
+
+	public ReactiveN1qlRepositoryQueryExecutor(final ReactiveCouchbaseOperations operations, final QueryMethod queryMethod) {
+		this.operations = operations;
+		this.queryMethod = queryMethod;
+	}
+
+	public Object execute(final Object[] parameters) {
+		final Class<?> domainClass = queryMethod.getResultProcessor().getReturnedType().getDomainType();
+		final ParameterAccessor accessor = new ParametersParameterAccessor(queryMethod.getParameters(), parameters);
+
+		final PartTree tree = new PartTree(queryMethod.getName(), domainClass);
+		Query query = new N1qlQueryCreator(tree, accessor, operations.getConverter().getMappingContext()).createQuery();
+
+		Flux<?> all = operations.findByQuery(domainClass).matching(query).all();
+		return all;
+	}
+
+}

src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactory.java

@@ -19,12 +19,15 @@ import java.io.Serializable;
 import java.lang.reflect.Method;
 import java.util.Optional;
 
+import org.springframework.data.couchbase.core.CouchbaseOperations;
 import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
 import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
 import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
 import org.springframework.data.couchbase.repository.config.ReactiveRepositoryOperationsMapping;
 import org.springframework.data.couchbase.repository.query.CouchbaseEntityInformation;
 import org.springframework.data.couchbase.repository.query.CouchbaseQueryMethod;
+import org.springframework.data.couchbase.repository.query.CouchbaseRepositoryQuery;
+import org.springframework.data.couchbase.repository.query.ReactiveCouchbaseRepositoryQuery;
 import org.springframework.data.couchbase.repository.query.ReactivePartTreeN1qlBasedQuery;
 import org.springframework.data.couchbase.repository.query.ReactiveStringN1qlBasedQuery;
 import org.springframework.data.mapping.context.MappingContext;
@@ -34,6 +37,7 @@ import org.springframework.data.repository.core.RepositoryInformation;
 import org.springframework.data.repository.core.RepositoryMetadata;
 import org.springframework.data.repository.core.support.ReactiveRepositoryFactorySupport;
 import org.springframework.data.repository.query.QueryLookupStrategy;
+import org.springframework.data.repository.query.QueryMethod;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -57,6 +61,8 @@ public class ReactiveCouchbaseRepositoryFactory extends ReactiveRepositoryFactor
 	 */
 	private final MappingContext<? extends CouchbasePersistentEntity<?>, CouchbasePersistentProperty> mappingContext;
 
+	private final CrudMethodMetadataPostProcessor crudMethodMetadataPostProcessor;
+
 	/**
 	 * Create a new factory.
 	 *
@@ -66,8 +72,16 @@ public class ReactiveCouchbaseRepositoryFactory extends ReactiveRepositoryFactor
 		Assert.notNull(couchbaseOperationsMapping);
 
 		this.couchbaseOperationsMapping = couchbaseOperationsMapping;
+		this.crudMethodMetadataPostProcessor = new CrudMethodMetadataPostProcessor();
 		mappingContext = this.couchbaseOperationsMapping.getMappingContext();
 
+		addRepositoryProxyPostProcessor(crudMethodMetadataPostProcessor);
+	}
+
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		super.setBeanClassLoader(classLoader);
+		this.crudMethodMetadataPostProcessor.setBeanClassLoader(classLoader);
 	}
 
 	/**
@@ -80,9 +94,9 @@ public class ReactiveCouchbaseRepositoryFactory extends ReactiveRepositoryFactor
 	 */
 	@Override
 	public <T, ID> CouchbaseEntityInformation<T, ID> getEntityInformation(Class<T> domainClass) {
-		CouchbasePersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(domainClass);
-
-		return new MappingCouchbaseEntityInformation<>((CouchbasePersistentEntity<T>) entity);
+		CouchbasePersistentEntity<T> entity = (CouchbasePersistentEntity<T>) mappingContext
+			.getRequiredPersistentEntity(domainClass);
+		return new MappingCouchbaseEntityInformation<>(entity);
 	}
 
 	/**
@@ -97,15 +111,12 @@ public class ReactiveCouchbaseRepositoryFactory extends ReactiveRepositoryFactor
 	@Override
 	protected final Object getTargetRepository(final RepositoryInformation metadata) {
 		ReactiveCouchbaseOperations couchbaseOperations = couchbaseOperationsMapping.resolve(metadata.getRepositoryInterface(),
-				metadata.getDomainType());
-		// boolean isN1qlAvailable =
-		// couchbaseOperations.getCouchbaseClusterConfig().clusterCapabilities().containsKey(ServiceType.QUERY);
-
+			metadata.getDomainType());
 		CouchbaseEntityInformation<?, Serializable> entityInformation = getEntityInformation(metadata.getDomainType());
-		SimpleReactiveCouchbaseRepository repo = getTargetRepositoryViaReflection(metadata, entityInformation,
-				couchbaseOperations);
-		// repo.setViewMetadataProvider(viewPostProcessor.getViewMetadataProvider());
-		return repo;
+		SimpleReactiveCouchbaseRepository repository = getTargetRepositoryViaReflection(metadata, entityInformation,
+			couchbaseOperations);
+		repository.setRepositoryMethodMetadata(crudMethodMetadataPostProcessor.getCrudMethodMetadata());
+		return repository;
 	}
 
 	/**
@@ -119,8 +130,6 @@ public class ReactiveCouchbaseRepositoryFactory extends ReactiveRepositoryFactor
 	 */
 	@Override
 	protected final Class<?> getRepositoryBaseClass(final RepositoryMetadata repositoryMetadata) {
-		// Since we always need n1ql (we eliminated use of views for findAll, etc...), lets just
-		// always return the n1ql repo
 		return SimpleReactiveCouchbaseRepository.class;
 	}
 
@@ -144,23 +153,10 @@ public class ReactiveCouchbaseRepositoryFactory extends ReactiveRepositoryFactor
 		@Override
 		public RepositoryQuery resolveQuery(Method method, RepositoryMetadata metadata, ProjectionFactory factory,
 				NamedQueries namedQueries) {
-			ReactiveCouchbaseOperations couchbaseOperations = couchbaseOperationsMapping.resolve(metadata.getRepositoryInterface(),
-					metadata.getDomainType());
+			final ReactiveCouchbaseOperations couchbaseOperations = couchbaseOperationsMapping
+				.resolve(metadata.getRepositoryInterface(), metadata.getDomainType());
 
-			CouchbaseQueryMethod queryMethod = new CouchbaseQueryMethod(method, metadata, factory, mappingContext);
-			String namedQueryName = queryMethod.getNamedQueryName();
-
-			if (queryMethod.hasN1qlAnnotation()) {
-				if (queryMethod.hasInlineN1qlQuery()) {
-					return new ReactiveStringN1qlBasedQuery(queryMethod.getInlineN1qlQuery(), queryMethod, couchbaseOperations,
-							SPEL_PARSER, evaluationContextProvider);
-				} else if (namedQueries.hasQuery(namedQueryName)) {
-					String namedQuery = namedQueries.getQuery(namedQueryName);
-					return new ReactiveStringN1qlBasedQuery(namedQuery, queryMethod, couchbaseOperations, SPEL_PARSER,
-							evaluationContextProvider);
-				} // otherwise will do default, queryDerivation
-			}
-			return new ReactivePartTreeN1qlBasedQuery(queryMethod, couchbaseOperations);
+			return new ReactiveCouchbaseRepositoryQuery(couchbaseOperations, new QueryMethod(method, metadata, factory));
 		}
 	}
 

src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactoryBean.java

@@ -51,10 +51,10 @@ public class ReactiveCouchbaseRepositoryFactoryBean<T extends Repository<S, ID>,
 	 * @param reactiveCouchbaseOperations the reference to the operations template.
 	 */
 	public void setCouchbaseOperations(final ReactiveCouchbaseOperations reactiveCouchbaseOperations) {
-		setCouchbaseOperationsMapping(new ReactiveRepositoryOperationsMapping(reactiveCouchbaseOperations));
+		setReactiveCouchbaseOperationsMapping(new ReactiveRepositoryOperationsMapping(reactiveCouchbaseOperations));
 	}
 
-	public void setCouchbaseOperationsMapping(final ReactiveRepositoryOperationsMapping couchbaseOperationsMapping) {
+	public void setReactiveCouchbaseOperationsMapping(final ReactiveRepositoryOperationsMapping couchbaseOperationsMapping) {
 		this.couchbaseOperationsMapping = couchbaseOperationsMapping;
 		setMappingContext(couchbaseOperationsMapping.getMappingContext());
 	}

src/main/java/org/springframework/data/couchbase/repository/support/SimpleCouchbaseRepository.java

@@ -153,11 +153,6 @@ public class SimpleCouchbaseRepository<T, ID> implements CouchbaseRepository<T,
 		return new PageImpl<>(results, pageable, count());
 	}
 
-	@Override
-	public CouchbaseOperations getCouchbaseOperations() {
-		return couchbaseOperations;
-	}
-
 	/**
 	 * Returns the information for the underlying template.
 	 *

src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java

@@ -16,6 +16,7 @@
 
 package org.springframework.data.couchbase.repository.support;
 
+import com.couchbase.client.java.query.QueryScanConsistency;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -55,18 +56,20 @@ public class SimpleReactiveCouchbaseRepository<T, ID> implements ReactiveCouchba
 	 */
 	private final CouchbaseEntityInformation<T, String> entityInformation;
 
+	private CrudMethodMetadata crudMethodMetadata;
+
 	/**
 	 * Create a new Repository.
 	 *
-	 * @param metadata the Metadata for the entity.
+	 * @param entityInformation the Metadata for the entity.
 	 * @param operations the reference to the reactive template used.
 	 */
-	public SimpleReactiveCouchbaseRepository(final CouchbaseEntityInformation<T, String> metadata,
+	public SimpleReactiveCouchbaseRepository(final CouchbaseEntityInformation<T, String> entityInformation,
 			final ReactiveCouchbaseOperations operations) {
-		Assert.notNull(operations, "RxJavaCouchbaseOperations must not be null!");
-		Assert.notNull(metadata, "CouchbaseEntityInformation must not be null!");
+		Assert.notNull(operations, "ReactiveCouchbaseOperations must not be null!");
+		Assert.notNull(entityInformation, "CouchbaseEntityInformation must not be null!");
 
-		this.entityInformation = metadata;
+		this.entityInformation = entityInformation;
 		this.operations = operations;
 	}
 
@@ -195,13 +198,25 @@ public class SimpleReactiveCouchbaseRepository<T, ID> implements ReactiveCouchba
 		return entityInformation;
 	}
 
-	@Override
-	public ReactiveCouchbaseOperations getReactiveCouchbaseOperations() {
-		return operations;
+	private Flux<T> findAll(final Query query) {
+		return operations.findByQuery(entityInformation.getJavaType()).consistentWith(buildQueryScanConsistency()).matching(query).all();
 	}
 
-	private Flux<T> findAll(final Query query) {
-		return operations.findByQuery(entityInformation.getJavaType()).matching(query).all();
+	private QueryScanConsistency buildQueryScanConsistency() {
+		QueryScanConsistency scanConsistency = QueryScanConsistency.NOT_BOUNDED;
+		if (crudMethodMetadata.getScanConsistency() != null) {
+			scanConsistency = crudMethodMetadata.getScanConsistency().query();
+		}
+		return scanConsistency;
+	}
+
+	/**
+	 * Setter for the repository metadata, contains annotations on the overidden methods.
+	 *
+	 * @param crudMethodMetadata the injected repository metadata.
+	 */
+	void setRepositoryMethodMetadata(final CrudMethodMetadata crudMethodMetadata) {
+		this.crudMethodMetadata = crudMethodMetadata;
 	}
 
 }

src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java

@@ -19,6 +19,7 @@ package org.springframework.data.couchbase.core;
 import static org.junit.jupiter.api.Assertions.*;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.UUID;
 
 import org.junit.jupiter.api.AfterAll;
@@ -44,6 +45,7 @@ class CouchbaseTemplateKeyValueIntegrationTests extends ClusterAwareIntegrationT
 	@BeforeAll
 	static void beforeAll() {
 		couchbaseClientFactory = new SimpleCouchbaseClientFactory(connectionString(), authenticator(), bucketName());
+		couchbaseClientFactory.getBucket().waitUntilReady(Duration.ofSeconds(10));
 	}
 
 	@AfterAll

src/test/java/org/springframework/data/couchbase/core/CustomTypeKeyIntegrationTests.java

@@ -18,6 +18,7 @@ package org.springframework.data.couchbase.core;
 
 import static org.junit.jupiter.api.Assertions.*;
 
+import java.time.Duration;
 import java.util.UUID;
 
 import org.junit.jupiter.api.Test;
@@ -44,6 +45,8 @@ public class CustomTypeKeyIntegrationTests extends ClusterAwareIntegrationTests
 
 	@Test
 	void saveSimpleEntityCorrectlyWithDifferentTypeKey() {
+		clientFactory.getBucket().waitUntilReady(Duration.ofSeconds(5));
+
 		User user = new User(UUID.randomUUID().toString(), "firstname", "lastname");
 		User modified = operations.upsertById(User.class).one(user);
 		assertEquals(user, modified);

src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2019 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.
@@ -13,17 +13,23 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.couchbase.repository;
 
-import java.io.Serializable;
+package org.springframework.data.couchbase.domain;
+
+import org.springframework.data.couchbase.repository.ScanConsistency;
 import org.springframework.data.repository.reactive.ReactiveSortingRepository;
+import org.springframework.stereotype.Repository;
+
+import com.couchbase.client.java.query.QueryScanConsistency;
+import reactor.core.publisher.Flux;
+
+@Repository
+public interface ReactiveAirportRepository extends ReactiveSortingRepository<Airport, String> {
+
+	@Override
+	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
+	Flux<Airport> findAll();
+
+	Flux<Airport> findAllByIata(String iata);
 
-/**
- * Couchbase specific {@link org.springframework.data.repository.Repository} interface that is
- * a {@link ReactiveSortingRepository}.
- *
- * @author Subhashni Balakrishnan
- */
-public interface ReactiveCouchbaseSortingRepository<T, ID extends Serializable>
-        extends ReactiveCouchbaseRepository<T, ID>, ReactiveSortingRepository<T, ID> {
 }

src/test/java/org/springframework/data/couchbase/domain/ReactiveUserRepository.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2012-2020 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.data.couchbase.domain;
+
+import org.springframework.data.repository.reactive.ReactiveSortingRepository;
+import org.springframework.stereotype.Repository;
+import reactor.core.publisher.Flux;
+
+@Repository
+public interface ReactiveUserRepository extends ReactiveSortingRepository<User, String> {
+
+	Flux<User> findByFirstname(String firstname);
+
+	Flux<User> findByFirstnameAndLastname(String firstname, String lastname);
+
+}

src/test/java/org/springframework/data/couchbase/repository/CouchbaseRepositoryKeyValueIntegrationTests.java

@@ -29,9 +29,12 @@ import org.springframework.data.couchbase.domain.User;
 import org.springframework.data.couchbase.domain.UserRepository;
 import org.springframework.data.couchbase.repository.config.EnableCouchbaseRepositories;
 import org.springframework.data.couchbase.util.ClusterAwareIntegrationTests;
+import org.springframework.data.couchbase.util.ClusterType;
+import org.springframework.data.couchbase.util.IgnoreWhen;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 @SpringJUnitConfig(CouchbaseRepositoryKeyValueIntegrationTests.Config.class)
+@IgnoreWhen(clusterTypes = ClusterType.MOCKED)
 public class CouchbaseRepositoryKeyValueIntegrationTests extends ClusterAwareIntegrationTests {
 
 	@Autowired UserRepository userRepository;

src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryKeyValueIntegrationTests.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2012-2020 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.data.couchbase.repository;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.Optional;
+import java.util.UUID;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.couchbase.config.AbstractCouchbaseConfiguration;
+import org.springframework.data.couchbase.domain.ReactiveUserRepository;
+import org.springframework.data.couchbase.domain.User;
+import org.springframework.data.couchbase.repository.config.EnableReactiveCouchbaseRepositories;
+import org.springframework.data.couchbase.util.ClusterAwareIntegrationTests;
+import org.springframework.data.couchbase.util.ClusterType;
+import org.springframework.data.couchbase.util.IgnoreWhen;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+@SpringJUnitConfig(ReactiveCouchbaseRepositoryKeyValueIntegrationTests.Config.class)
+@IgnoreWhen(clusterTypes = ClusterType.MOCKED)
+public class ReactiveCouchbaseRepositoryKeyValueIntegrationTests extends ClusterAwareIntegrationTests {
+
+	@Autowired
+	ReactiveUserRepository userRepository;
+
+	@Test
+	void saveAndFindById() {
+		User user = new User(UUID.randomUUID().toString(), "f", "l");
+
+		assertFalse(userRepository.existsById(user.getId()).block());
+
+		userRepository.save(user).block();
+
+		Optional<User> found = userRepository.findById(user.getId()).blockOptional();
+		assertTrue(found.isPresent());
+		found.ifPresent(u -> assertEquals(user, u));
+
+		assertTrue(userRepository.existsById(user.getId()).block());
+	}
+
+	@Configuration
+	@EnableReactiveCouchbaseRepositories("org.springframework.data.couchbase")
+	static class Config extends AbstractCouchbaseConfiguration {
+
+		@Override
+		public String getConnectionString() {
+			return connectionString();
+		}
+
+		@Override
+		public String getUserName() {
+			return config().adminUsername();
+		}
+
+		@Override
+		public String getPassword() {
+			return config().adminPassword();
+		}
+
+		@Override
+		public String getBucketName() {
+			return bucketName();
+		}
+
+	}
+
+}

src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2017-2019 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.data.couchbase.repository;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.couchbase.CouchbaseClientFactory;
+import org.springframework.data.couchbase.config.AbstractCouchbaseConfiguration;
+import org.springframework.data.couchbase.domain.Airport;
+import org.springframework.data.couchbase.domain.AirportRepository;
+import org.springframework.data.couchbase.domain.ReactiveAirportRepository;
+import org.springframework.data.couchbase.repository.config.EnableCouchbaseRepositories;
+import org.springframework.data.couchbase.repository.config.EnableReactiveCouchbaseRepositories;
+import org.springframework.data.couchbase.util.Capabilities;
+import org.springframework.data.couchbase.util.ClusterAwareIntegrationTests;
+import org.springframework.data.couchbase.util.ClusterType;
+import org.springframework.data.couchbase.util.IgnoreWhen;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import com.couchbase.client.core.error.IndexExistsException;
+
+@SpringJUnitConfig(ReactiveCouchbaseRepositoryQueryIntegrationTests.Config.class)
+@IgnoreWhen(missesCapabilities = Capabilities.QUERY, clusterTypes = ClusterType.MOCKED)
+public class ReactiveCouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegrationTests {
+
+	@Autowired CouchbaseClientFactory clientFactory;
+
+	@Autowired ReactiveAirportRepository airportRepository;
+
+	@BeforeEach
+	void beforeEach() {
+		try {
+			clientFactory.getCluster().queryIndexes().createPrimaryIndex(bucketName());
+		} catch (IndexExistsException ex) {
+			// ignore, all good.
+		}
+	}
+
+	@Test
+	void shouldSaveAndFindAll() {
+		Airport vie = new Airport("airports::vie", "vie", "loww");
+		airportRepository.save(vie).block();
+
+		List<Airport> all = airportRepository.findAll().toStream()
+				.collect(Collectors.toList());
+
+		assertFalse(all.isEmpty());
+		assertTrue(all.stream().anyMatch(a -> a.getId().equals("airports::vie")));
+	}
+
+	@Test
+	void findBySimpleProperty() {
+		List<Airport> airports = airportRepository.findAllByIata("vie").collectList().block();
+		// TODO
+		System.err.println(airports);
+	}
+
+	@Configuration
+	@EnableReactiveCouchbaseRepositories("org.springframework.data.couchbase")
+	static class Config extends AbstractCouchbaseConfiguration {
+
+		@Override
+		public String getConnectionString() {
+			return connectionString();
+		}
+
+		@Override
+		public String getUserName() {
+			return config().adminUsername();
+		}
+
+		@Override
+		public String getPassword() {
+			return config().adminPassword();
+		}
+
+		@Override
+		public String getBucketName() {
+			return bucketName();
+		}
+
+	}
+
+}