After 1490, changes needed so integration tests pass.

Closes #1498.

Author: mikereiche

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

@@ -18,7 +18,6 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import org.reactivestreams.Publisher;
 import org.springframework.data.couchbase.core.CouchbaseOperations;
 import org.springframework.data.couchbase.core.ExecutableFindByQueryOperation.ExecutableFindByQuery;
 import org.springframework.data.couchbase.core.query.Query;
@@ -112,14 +111,14 @@ public Object execute(Object[] parameters) {
 
 		ReactiveCouchbaseParameterAccessor accessor = new ReactiveCouchbaseParameterAccessor(getQueryMethod(), parameters);
 
-		return accessor.resolveParameters().flatMapMany(this::executeDeferred);
+		Object result = accessor.resolveParameters()./* flatMapMany */map(this::executeDeferred);
+		result = result instanceof Flux ? ((Flux<Object>) result).collectList().block()
+				: result instanceof Mono ? ((Mono<Object>) result).block() : result;
+		return result;
 	}
 
-	private Publisher<Object> executeDeferred(ReactiveCouchbaseParameterAccessor parameterAccessor) {
-		if (getQueryMethod().isCollectionQuery()) {
-			return Flux.defer(() -> (Publisher<Object>) execute(parameterAccessor));
-		}
-		return Mono.defer(() -> (Mono<Object>) execute(parameterAccessor));
+	private Object executeDeferred(ReactiveCouchbaseParameterAccessor parameterAccessor) {
+		return execute(parameterAccessor);
 	}
 
 	private Object execute(ParametersParameterAccessor parameterAccessor) {
@@ -131,7 +130,9 @@ private Object execute(ParametersParameterAccessor parameterAccessor) {
 		if (typeToRead == null && returnType.getComponentType() != null) {
 			typeToRead = returnType.getComponentType().getType();
 		}
-		return doExecute(getQueryMethod(), processor, parameterAccessor, typeToRead);
+		Object result = doExecute(getQueryMethod(), processor, parameterAccessor, typeToRead);
+		System.err.println("execute parameterAccessor result: " + result);
+		return result;
 	}
 
 	/**