DATACOUCH-588 - Implement pageable and realign repo query

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

@@ -21,6 +21,7 @@
 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,8 @@ static class ExecutableFindByQuerySupport<T> implements ExecutableFindByQuery<T>
 			this.template = template;
 			this.domainType = domainType;
 			this.query = query;
-			this.reactiveSupport = new ReactiveFindByQueryOperationSupport.ReactiveFindByQuerySupport<T>(
-					template.reactive(), domainType, query, scanConsistency);
+			this.reactiveSupport = new ReactiveFindByQuerySupport<T>(template.reactive(),
+					domainType, query, scanConsistency);
 			this.scanConsistency = scanConsistency;
 		}
 
@@ -72,7 +73,13 @@ public List<T> all() {
 
 		@Override
 		public TerminatingFindByQuery<T> matching(final Query query) {
-			return new ExecutableFindByQuerySupport<>(template, domainType, query, scanConsistency);
+			QueryScanConsistency scanCons;
+			if (query.getConsistency() != null) {
+				scanCons = query.getConsistency();
+			} else {
+				scanCons = scanConsistency;
+			}
+			return new ExecutableFindByQuerySupport<>(template, domainType, query, scanCons);
 		}
 
 		@Override

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

@@ -104,6 +104,129 @@ interface FindByQueryConsistentWith<T> extends FindByQueryWithQuery<T> {
 
 	}
 
-	interface ReactiveFindByQuery<T> extends FindByQueryConsistentWith<T> {}
+	/**
+	 * Collection override (optional).
+	 */
+	interface FindWithCollection<T> extends FindByQueryWithQuery<T> {
+
+		/**
+		 * Explicitly set the name of the collection to perform the query on. <br />
+		 * Skip this step to use the default collection derived from the domain type.
+		 *
+		 * @param collection must not be {@literal null} nor {@literal empty}.
+		 * @return new instance of {@link FindWithProjection}.
+		 * @throws IllegalArgumentException if collection is {@literal null}.
+		 */
+		FindWithProjection<T> inCollection(String collection);
+	}
+
+	/**
+	 * Result type override (optional).
+	 */
+	interface FindWithProjection<T> extends FindByQueryWithQuery<T>, FindDistinct {
+
+		/**
+		 * Define the target type fields should be mapped to. <br />
+		 * Skip this step if you are anyway only interested in the original domain type.
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return new instance of {@link FindWithProjection}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		<R> FindByQueryWithQuery<R> as(Class<R> resultType);
+	}
+
+	/**
+	 * Distinct Find support.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface FindDistinct {
+
+		/**
+		 * Finds the distinct values for a specified {@literal field} across a single {@link } or view.
+		 *
+		 * @param field name of the field. Must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if field is {@literal null}.
+		 */
+		TerminatingDistinct<Object> distinct(String field);
+	}
+
+	/**
+	 * Result type override. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithProjection {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 * <dl>
+		 * <dt>{@link Object} (the default)</dt>
+		 * <dd>Result is mapped according to the {@link } converting eg. {@link } into plain {@link String}, {@link } to
+		 * {@link Long}, etc. always picking the most concrete type with respect to the domain types property.<br />
+		 * Any {@link } is run through the {@link org.springframework.data.convert.EntityReader} to obtain the domain type.
+		 * <br />
+		 * Using {@link Object} also works for non strictly typed fields. Eg. a mixture different types like fields using
+		 * {@link String} in one {@link } while {@link Long} in another.</dd>
+		 * <dt>Any Simple type like {@link String}, {@link Long}, ...</dt>
+		 * <dd>The result is mapped directly by the Couchbase Java driver and the {@link } in place. This works only for
+		 * results where all documents considered for the operation use the very same type for the field.</dd>
+		 * <dt>Any Domain type</dt>
+		 * <dd>Domain types can only be mapped if the if the result of the actual {@code distinct()} operation returns
+		 * {@link }.</dd>
+		 * <dt>{@link }</dt>
+		 * <dd>Using {@link } allows retrieval of the raw driver specific format, which returns eg. {@link }.</dd>
+		 * </dl>
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		<R> TerminatingDistinct<R> as(Class<R> resultType);
+	}
+
+	/**
+	 * Result restrictions. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithQuery<T> extends DistinctWithProjection {
+
+		/**
+		 * Set the filter {@link Query criteria} to be used.
+		 *
+		 * @param query must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if criteria is {@literal null}.
+		 * @since 3.0
+		 */
+		TerminatingDistinct<T> matching(Query query);
+	}
+
+	/**
+	 * Terminating distinct find operations.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface TerminatingDistinct<T> extends DistinctWithQuery<T> {
+
+		/**
+		 * Get all matching distinct field values.
+		 *
+		 * @return empty {@link Flux} if not match found. Never {@literal null}.
+		 */
+		Flux<T> all();
+	}
+
+	interface ReactiveFindByQuery<T> extends FindByQueryConsistentWith<T>, FindWithCollection<T>, FindDistinct {}
 
 }

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

@@ -60,7 +60,13 @@ static class ReactiveFindByQuerySupport<T> implements ReactiveFindByQuery<T> {
 
 		@Override
 		public TerminatingFindByQuery<T> matching(Query query) {
-			return new ReactiveFindByQuerySupport<>(template, domainType, query, scanConsistency);
+			QueryScanConsistency scanCons;
+			if (query.getConsistency() != null) {
+				scanCons = query.getConsistency();
+			} else {
+				scanCons = scanConsistency;
+			}
+			return new ReactiveFindByQuerySupport<>(template, domainType, query, scanCons);
 		}
 
 		@Override
@@ -94,6 +100,7 @@ public Flux<T> all() {
 							long cas = row.getLong(TemplateUtils.SELECT_CAS);
 							row.removeKey(TemplateUtils.SELECT_ID);
 							row.removeKey(TemplateUtils.SELECT_CAS);
+							System.out.println("Row -> " + row.toString());
 							return template.support().decodeEntity(id, row.toString(), cas, domainType);
 						});
 			});
@@ -110,8 +117,10 @@ public Mono<Long> count() {
 							} else {
 								return throwable;
 							}
-						}).flatMapMany(ReactiveQueryResult::rowsAsObject).map(row -> row.getLong(TemplateUtils.SELECT_COUNT))
-						.next();
+						}).flatMapMany(ReactiveQueryResult::rowsAsObject).map(row -> {
+							System.out.println("count: " + row.getLong(TemplateUtils.SELECT_COUNT));
+							return row.getLong(TemplateUtils.SELECT_COUNT);
+						}).next();
 			});
 		}
 
@@ -124,6 +133,15 @@ private String assembleEntityQuery(final boolean count) {
 			return query.toN1qlString(template, this.domainType, count);
 		}
 
+		@Override
+		public FindWithProjection<T> inCollection(String collection) {
+			throw new RuntimeException(("not implemented"));
+		}
+
+		@Override
+		public TerminatingDistinct<Object> distinct(String field) {
+			throw new RuntimeException(("not implemented"));
+		}
 	}
 
 }

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");
 
@@ -104,13 +105,22 @@ public Query skip(long skip) {
 	 * Limit the number of returned documents to {@code limit}.
 	 *
 	 * @param limit
-	 * @return
+	 * @return this
 	 */
 	public Query limit(int limit) {
 		this.limit = limit;
 		return this;
 	}
 
+	/**
+	 * limit
+	 *
+	 * @return limit
+	 */
+	public int getLimit() {
+		return limit;
+	}
+
 	/**
 	 * Sets the given pagination information on the {@link Query} instance. Will transparently set {@code skip} and
 	 * {@code limit} as well as applying the {@link Sort} instance defined with the {@link Pageable}.
@@ -127,6 +137,27 @@ public Query with(final Pageable pageable) {
 		return with(pageable.getSort());
 	}
 
+	/**
+	 * queryScanconsistency
+	 *
+	 * @return queryScanConsistency
+	 */
+	public QueryScanConsistency getConsistency() {
+		return queryScanConsistency;
+	}
+
+	/**
+	 * Sets the given pagination information on the {@link Query} instance. Will transparently set {@code skip} and
+	 * {@code limit} as well as applying the {@link Sort} instance defined with the {@link Pageable}.
+	 *
+	 * @param queryScanConsistency
+	 * @return
+	 */
+	public Query with(final QueryScanConsistency queryScanConsistency) {
+		this.queryScanConsistency = queryScanConsistency;
+		return this;
+	}
+
 	/**
 	 * Adds a {@link Sort} to the {@link Query} instance.
 	 *