DATACOUCH-588 - Refactoring part 1 of n.

pom.xml

@@ -85,6 +85,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <version>3.1.0.RELEASE</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>

src/main/java/org/springframework/data/couchbase/core/ExecutableExistsByIdOperationSupport.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.couchbase.core;
 
+import org.springframework.data.couchbase.core.ReactiveExistsByIdOperationSupport.ReactiveExistsByIdSupport;
+
 import java.util.Collection;
 import java.util.Map;
 
@@ -36,12 +38,11 @@ public ExecutableExistsById existsById() {
 	static class ExecutableExistsByIdSupport implements ExecutableExistsById {
 
 		private final CouchbaseTemplate template;
-		private final ReactiveExistsByIdOperationSupport.ReactiveExistsByIdSupport reactiveSupport;
+		private final ReactiveExistsByIdSupport reactiveSupport;
 
 		ExecutableExistsByIdSupport(final CouchbaseTemplate template, final String collection) {
 			this.template = template;
-			this.reactiveSupport = new ReactiveExistsByIdOperationSupport.ReactiveExistsByIdSupport(template.reactive(),
-					collection);
+			this.reactiveSupport = new ReactiveExistsByIdSupport(template.reactive(), collection);
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByAnalyticsOperationSupport.java

@@ -15,13 +15,14 @@
  */
 package org.springframework.data.couchbase.core;
 
+import org.springframework.data.couchbase.core.ReactiveFindByAnalyticsOperationSupport.ReactiveFindByAnalyticsSupport;
+
 import java.util.List;
 import java.util.stream.Stream;
 
-import com.couchbase.client.java.analytics.AnalyticsScanConsistency;
-import com.couchbase.client.java.query.QueryScanConsistency;
 import org.springframework.data.couchbase.core.query.AnalyticsQuery;
-import org.springframework.data.couchbase.core.query.Query;
+
+import com.couchbase.client.java.analytics.AnalyticsScanConsistency;
 
 public class ExecutableFindByAnalyticsOperationSupport implements ExecutableFindByAnalyticsOperation {
 
@@ -35,14 +36,15 @@ public ExecutableFindByAnalyticsOperationSupport(final CouchbaseTemplate templat
 
 	@Override
 	public <T> ExecutableFindByAnalytics<T> findByAnalytics(final Class<T> domainType) {
-		return new ExecutableFindByAnalyticsSupport<>(template, domainType, ALL_QUERY, AnalyticsScanConsistency.NOT_BOUNDED);
+		return new ExecutableFindByAnalyticsSupport<>(template, domainType, ALL_QUERY,
+				AnalyticsScanConsistency.NOT_BOUNDED);
 	}
 
 	static class ExecutableFindByAnalyticsSupport<T> implements ExecutableFindByAnalytics<T> {
 
 		private final CouchbaseTemplate template;
 		private final Class<T> domainType;
-		private final ReactiveFindByAnalyticsOperationSupport.ReactiveFindByAnalyticsSupport<T> reactiveSupport;
+		private final ReactiveFindByAnalyticsSupport<T> reactiveSupport;
 		private final AnalyticsQuery query;
 		private final AnalyticsScanConsistency scanConsistency;
 
@@ -51,8 +53,8 @@ static class ExecutableFindByAnalyticsSupport<T> implements ExecutableFindByAnal
 			this.template = template;
 			this.domainType = domainType;
 			this.query = query;
-			this.reactiveSupport = new ReactiveFindByAnalyticsOperationSupport.ReactiveFindByAnalyticsSupport<>(
-					template.reactive(), domainType, query, scanConsistency);
+			this.reactiveSupport = new ReactiveFindByAnalyticsSupport<>(template.reactive(), domainType, query,
+					scanConsistency);
 			this.scanConsistency = scanConsistency;
 		}
 

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByIdOperationSupport.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.couchbase.core;
 
+import org.springframework.data.couchbase.core.ReactiveFindByIdOperationSupport.ReactiveFindByIdSupport;
+
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
@@ -40,15 +42,14 @@ static class ExecutableFindByIdSupport<T> implements ExecutableFindById<T> {
 		private final Class<T> domainType;
 		private final String collection;
 		private final List<String> fields;
-		private final ReactiveFindByIdOperationSupport.ReactiveFindByIdSupport<T> reactiveSupport;
+		private final ReactiveFindByIdSupport<T> reactiveSupport;
 
 		ExecutableFindByIdSupport(CouchbaseTemplate template, Class<T> domainType, String collection, List<String> fields) {
 			this.template = template;
 			this.domainType = domainType;
 			this.collection = collection;
 			this.fields = fields;
-			this.reactiveSupport = new ReactiveFindByIdOperationSupport.ReactiveFindByIdSupport<>(template.reactive(),
-					domainType, collection, fields);
+			this.reactiveSupport = new ReactiveFindByIdSupport<>(template.reactive(), domainType, collection, fields);
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperation.java

@@ -130,6 +130,17 @@ interface FindByQueryConsistentWith<T> extends FindByQueryWithQuery<T> {
 
 	}
 
-	interface ExecutableFindByQuery<T> extends FindByQueryConsistentWith<T> {}
+	interface FindByQueryInCollection<T> extends FindByQueryConsistentWith<T> {
+
+		/**
+		 * Allows to override the default scan consistency.
+		 *
+		 * @param collection the collection to use for this query.
+		 */
+		FindByQueryInCollection<T> inCollection(String collection);
+
+	}
+
+	interface ExecutableFindByQuery<T> extends FindByQueryInCollection<T> {}
 
 }

src/main/java/org/springframework/data/couchbase/core/ExecutableFindByQueryOperationSupport.java

@@ -18,11 +18,17 @@
 import java.util.List;
 import java.util.stream.Stream;
 
+import org.springframework.data.couchbase.core.ReactiveFindByQueryOperationSupport.ReactiveFindByQuerySupport;
 import org.springframework.data.couchbase.core.query.Query;
 
 import com.couchbase.client.java.query.QueryScanConsistency;
-import org.springframework.data.couchbase.core.ReactiveFindByQueryOperationSupport.ReactiveFindByQuerySupport;
 
+/**
+ * {@link ExecutableFindByQueryOperation} implementations for Couchbase.
+ *
+ * @author Michael Nitschinger
+ * @author Michael Reiche
+ */
 public class ExecutableFindByQueryOperationSupport implements ExecutableFindByQueryOperation {
 
 	private static final Query ALL_QUERY = new Query();
@@ -35,24 +41,28 @@ public ExecutableFindByQueryOperationSupport(final CouchbaseTemplate template) {
 
 	@Override
 	public <T> ExecutableFindByQuery<T> findByQuery(final Class<T> domainType) {
-		return new ExecutableFindByQuerySupport<>(template, domainType, ALL_QUERY, QueryScanConsistency.NOT_BOUNDED);
+		return new ExecutableFindByQuerySupport<>(template, domainType, ALL_QUERY, QueryScanConsistency.NOT_BOUNDED,
+				"_default._default");
 	}
 
 	static class ExecutableFindByQuerySupport<T> implements ExecutableFindByQuery<T> {
 
 		private final CouchbaseTemplate template;
 		private final Class<T> domainType;
 		private final Query query;
-		private final ReactiveFindByQueryOperationSupport.ReactiveFindByQuerySupport<T> reactiveSupport;
+		private final ReactiveFindByQuerySupport<T> reactiveSupport;
 		private final QueryScanConsistency scanConsistency;
+		private final String collection;
 
 		ExecutableFindByQuerySupport(final CouchbaseTemplate template, final Class<T> domainType, final Query query,
-				final QueryScanConsistency scanConsistency) {
+				final QueryScanConsistency scanConsistency, final String collection) {
 			this.template = template;
 			this.domainType = domainType;
 			this.query = query;
-			this.reactiveSupport = new ReactiveFindByQuerySupport<T>(template.reactive(), domainType, query, scanConsistency);
+			this.reactiveSupport = new ReactiveFindByQuerySupport<T>(template.reactive(), domainType, query, scanConsistency,
+					collection);
 			this.scanConsistency = scanConsistency;
+			this.collection = collection;
 		}
 
 		@Override
@@ -78,12 +88,17 @@ public TerminatingFindByQuery<T> matching(final Query query) {
 			} else {
 				scanCons = scanConsistency;
 			}
-			return new ExecutableFindByQuerySupport<>(template, domainType, query, scanCons);
+			return new ExecutableFindByQuerySupport<>(template, domainType, query, scanCons, collection);
 		}
 
 		@Override
 		public FindByQueryConsistentWith<T> consistentWith(final QueryScanConsistency scanConsistency) {
-			return new ExecutableFindByQuerySupport<>(template, domainType, query, scanConsistency);
+			return new ExecutableFindByQuerySupport<>(template, domainType, query, scanConsistency, collection);
+		}
+
+		@Override
+		public FindByQueryInCollection<T> inCollection(final String collection) {
+			return new ExecutableFindByQuerySupport<>(template, domainType, query, scanConsistency, collection);
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableFindFromReplicasByIdOperationSupport.java

@@ -17,6 +17,7 @@
 
 import java.util.Collection;
 
+import org.springframework.data.couchbase.core.ReactiveFindFromReplicasByIdOperationSupport.ReactiveFindFromReplicasByIdSupport;
 import org.springframework.util.Assert;
 
 public class ExecutableFindFromReplicasByIdOperationSupport implements ExecutableFindFromReplicasByIdOperation {
@@ -37,14 +38,13 @@ static class ExecutableFindFromReplicasByIdSupport<T> implements ExecutableFindF
 		private final CouchbaseTemplate template;
 		private final Class<T> domainType;
 		private final String collection;
-		private final ReactiveFindFromReplicasByIdOperationSupport.ReactiveFindFromReplicasByIdSupport<T> reactiveSupport;
+		private final ReactiveFindFromReplicasByIdSupport<T> reactiveSupport;
 
 		ExecutableFindFromReplicasByIdSupport(CouchbaseTemplate template, Class<T> domainType, String collection) {
 			this.template = template;
 			this.domainType = domainType;
 			this.collection = collection;
-			this.reactiveSupport = new ReactiveFindFromReplicasByIdOperationSupport.ReactiveFindFromReplicasByIdSupport<>(
-					template.reactive(), domainType, collection);
+			this.reactiveSupport = new ReactiveFindFromReplicasByIdSupport<>(template.reactive(), domainType, collection);
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableInsertByIdOperationSupport.java

@@ -18,6 +18,7 @@
 import java.time.Duration;
 import java.util.Collection;
 
+import org.springframework.data.couchbase.core.ReactiveInsertByIdOperationSupport.ReactiveInsertByIdSupport;
 import org.springframework.util.Assert;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;
@@ -48,7 +49,7 @@ static class ExecutableInsertByIdSupport<T> implements ExecutableInsertById<T> {
 		private final ReplicateTo replicateTo;
 		private final DurabilityLevel durabilityLevel;
 		private final Duration expiry;
-		private final ReactiveInsertByIdOperationSupport.ReactiveInsertByIdSupport<T> reactiveSupport;
+		private final ReactiveInsertByIdSupport<T> reactiveSupport;
 
 		ExecutableInsertByIdSupport(final CouchbaseTemplate template, final Class<T> domainType, final String collection,
 				final PersistTo persistTo, final ReplicateTo replicateTo, final DurabilityLevel durabilityLevel,
@@ -60,8 +61,8 @@ static class ExecutableInsertByIdSupport<T> implements ExecutableInsertById<T> {
 			this.replicateTo = replicateTo;
 			this.durabilityLevel = durabilityLevel;
 			this.expiry = expiry;
-			this.reactiveSupport = new ReactiveInsertByIdOperationSupport.ReactiveInsertByIdSupport<>(template.reactive(),
-					domainType, collection, persistTo, replicateTo, durabilityLevel, expiry);
+			this.reactiveSupport = new ReactiveInsertByIdSupport<>(template.reactive(), domainType, collection, persistTo,
+					replicateTo, durabilityLevel, expiry);
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableRemoveByIdOperationSupport.java

@@ -18,6 +18,7 @@
 import java.util.Collection;
 import java.util.List;
 
+import org.springframework.data.couchbase.core.ReactiveRemoveByIdOperationSupport.ReactiveRemoveByIdSupport;
 import org.springframework.util.Assert;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;
@@ -44,7 +45,7 @@ static class ExecutableRemoveByIdSupport implements ExecutableRemoveById {
 		private final PersistTo persistTo;
 		private final ReplicateTo replicateTo;
 		private final DurabilityLevel durabilityLevel;
-		private final ReactiveRemoveByIdOperationSupport.ReactiveRemoveByIdSupport reactiveRemoveByIdSupport;
+		private final ReactiveRemoveByIdSupport reactiveRemoveByIdSupport;
 
 		ExecutableRemoveByIdSupport(final CouchbaseTemplate template, final String collection, final PersistTo persistTo,
 				final ReplicateTo replicateTo, final DurabilityLevel durabilityLevel) {
@@ -53,8 +54,8 @@ static class ExecutableRemoveByIdSupport implements ExecutableRemoveById {
 			this.persistTo = persistTo;
 			this.replicateTo = replicateTo;
 			this.durabilityLevel = durabilityLevel;
-			this.reactiveRemoveByIdSupport = new ReactiveRemoveByIdOperationSupport.ReactiveRemoveByIdSupport(
-					template.reactive(), collection, persistTo, replicateTo, durabilityLevel);
+			this.reactiveRemoveByIdSupport = new ReactiveRemoveByIdSupport(template.reactive(), collection, persistTo,
+					replicateTo, durabilityLevel);
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ExecutableRemoveByQueryOperation.java

@@ -43,6 +43,12 @@ interface RemoveByQueryConsistentWith<T> extends RemoveByQueryWithQuery<T> {
 
 	}
 
-	interface ExecutableRemoveByQuery<T> extends RemoveByQueryConsistentWith<T> {}
+	interface RemoveByQueryInCollection<T> extends RemoveByQueryConsistentWith<T> {
+
+		RemoveByQueryConsistentWith<T> inCollection(String collection);
+
+	}
+
+	interface ExecutableRemoveByQuery<T> extends RemoveByQueryInCollection<T> {}
 
 }

src/main/java/org/springframework/data/couchbase/core/ExecutableRemoveByQueryOperationSupport.java

@@ -17,6 +17,7 @@
 
 import java.util.List;
 
+import org.springframework.data.couchbase.core.ReactiveRemoveByQueryOperationSupport.ReactiveRemoveByQuerySupport;
 import org.springframework.data.couchbase.core.query.Query;
 
 import com.couchbase.client.java.query.QueryScanConsistency;
@@ -33,25 +34,28 @@ public ExecutableRemoveByQueryOperationSupport(final CouchbaseTemplate template)
 
 	@Override
 	public <T> ExecutableRemoveByQuery<T> removeByQuery(Class<T> domainType) {
-		return new ExecutableRemoveByQuerySupport<>(template, domainType, ALL_QUERY, QueryScanConsistency.NOT_BOUNDED);
+		return new ExecutableRemoveByQuerySupport<>(template, domainType, ALL_QUERY, QueryScanConsistency.NOT_BOUNDED,
+				"_default._default");
 	}
 
 	static class ExecutableRemoveByQuerySupport<T> implements ExecutableRemoveByQuery<T> {
 
 		private final CouchbaseTemplate template;
 		private final Class<T> domainType;
 		private final Query query;
-		private final ReactiveRemoveByQueryOperationSupport.ReactiveRemoveByQuerySupport<T> reactiveSupport;
+		private final ReactiveRemoveByQuerySupport<T> reactiveSupport;
 		private final QueryScanConsistency scanConsistency;
+		private final String collection;
 
 		ExecutableRemoveByQuerySupport(final CouchbaseTemplate template, final Class<T> domainType, final Query query,
-				final QueryScanConsistency scanConsistency) {
+				final QueryScanConsistency scanConsistency, String collection) {
 			this.template = template;
 			this.domainType = domainType;
 			this.query = query;
-			this.reactiveSupport = new ReactiveRemoveByQueryOperationSupport.ReactiveRemoveByQuerySupport<>(
-					template.reactive(), domainType, query, scanConsistency);
+			this.reactiveSupport = new ReactiveRemoveByQuerySupport<>(template.reactive(), domainType, query, scanConsistency,
+					collection);
 			this.scanConsistency = scanConsistency;
+			this.collection = collection;
 		}
 
 		@Override
@@ -61,12 +65,17 @@ public List<RemoveResult> all() {
 
 		@Override
 		public TerminatingRemoveByQuery<T> matching(final Query query) {
-			return new ExecutableRemoveByQuerySupport<>(template, domainType, query, scanConsistency);
+			return new ExecutableRemoveByQuerySupport<>(template, domainType, query, scanConsistency, collection);
 		}
 
 		@Override
 		public RemoveByQueryWithQuery<T> consistentWith(final QueryScanConsistency scanConsistency) {
-			return new ExecutableRemoveByQuerySupport<>(template, domainType, query, scanConsistency);
+			return new ExecutableRemoveByQuerySupport<>(template, domainType, query, scanConsistency, collection);
+		}
+
+		@Override
+		public RemoveByQueryInCollection<T> inCollection(final String collection) {
+			return new ExecutableRemoveByQuerySupport<>(template, domainType, query, scanConsistency, collection);
 		}
 
 	}

src/main/java/org/springframework/data/couchbase/core/ExecutableReplaceByIdOperationSupport.java

@@ -19,6 +19,7 @@
 import java.util.Collection;
 
 import org.springframework.util.Assert;
+import org.springframework.data.couchbase.core.ReactiveReplaceByIdOperationSupport.ReactiveReplaceByIdSupport;
 
 import com.couchbase.client.core.msg.kv.DurabilityLevel;
 import com.couchbase.client.java.kv.PersistTo;
@@ -48,7 +49,7 @@ static class ExecutableReplaceByIdSupport<T> implements ExecutableReplaceById<T>
 		private final ReplicateTo replicateTo;
 		private final DurabilityLevel durabilityLevel;
 		private final Duration expiry;
-		private final ReactiveReplaceByIdOperationSupport.ReactiveReplaceByIdSupport<T> reactiveSupport;
+		private final ReactiveReplaceByIdSupport<T> reactiveSupport;
 
 		ExecutableReplaceByIdSupport(final CouchbaseTemplate template, final Class<T> domainType, final String collection,
 				final PersistTo persistTo, final ReplicateTo replicateTo, final DurabilityLevel durabilityLevel,
@@ -60,7 +61,7 @@ static class ExecutableReplaceByIdSupport<T> implements ExecutableReplaceById<T>
 			this.replicateTo = replicateTo;
 			this.durabilityLevel = durabilityLevel;
 			this.expiry = expiry;
-			this.reactiveSupport = new ReactiveReplaceByIdOperationSupport.ReactiveReplaceByIdSupport<>(template.reactive(),
+			this.reactiveSupport = new ReactiveReplaceByIdSupport<>(template.reactive(),
 					domainType, collection, persistTo, replicateTo, durabilityLevel, expiry);
 		}