Adapt returned Publisher in ReactiveSearchBasedCouchbaseQuery and add tests. Minor cleanup.

Signed-off-by: Emilien Bevierre <emilien.bevierre@couchbase.com>

Author: emilienbev

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

@@ -15,16 +15,16 @@
  */
 package org.springframework.data.couchbase.repository;
 
+import com.couchbase.client.java.analytics.AnalyticsScanConsistency;
+import com.couchbase.client.java.query.QueryScanConsistency;
+import com.couchbase.client.java.search.SearchScanConsistency;
+
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
-import com.couchbase.client.java.analytics.AnalyticsScanConsistency;
-import com.couchbase.client.java.query.QueryScanConsistency;
-import com.couchbase.client.java.search.SearchScanConsistency;
-
 /**
  * Scan Consistency Annotation
  *

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

@@ -15,13 +15,13 @@
  */
 package org.springframework.data.couchbase.repository.query;
 
+import org.reactivestreams.Publisher;
 import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
 import org.springframework.data.couchbase.core.ReactiveFindBySearchOperation;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.repository.query.ParametersParameterAccessor;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
-import org.springframework.data.repository.util.ReactiveWrapperConverters;
 import org.springframework.util.Assert;
 
 import org.springframework.data.couchbase.repository.Search;
@@ -31,6 +31,9 @@
 import com.couchbase.client.java.search.SearchRequest;
 import com.couchbase.client.java.search.SearchScanConsistency;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 /**
  * Reactive {@link RepositoryQuery} implementation for FTS {@link Search}-annotated methods.
  * <p>
@@ -73,16 +76,21 @@ public ReactiveSearchBasedCouchbaseQuery(ReactiveCouchbaseQueryMethod method,
 	public Object execute(Object[] parameters) {
 		ReactiveCouchbaseParameterAccessor accessor = new ReactiveCouchbaseParameterAccessor(method, parameters);
 
-		return accessor.resolveParameters().flatMapMany(resolvedAccessor -> {
-			ResultProcessor processor = method.getResultProcessor().withDynamicProjection(resolvedAccessor);
-			SearchRepositoryQuerySupport.validateSort(resolvedAccessor);
+		if (method.isCollectionQuery()) {
+			return accessor.resolveParameters().flatMapMany(resolvedAccessor -> Flux.from(doExecute(resolvedAccessor)));
+		}
+		return accessor.resolveParameters().flatMap(resolvedAccessor -> Mono.from(doExecute(resolvedAccessor)));
+	}
+
+	private Publisher<?> doExecute(ParametersParameterAccessor resolvedAccessor) {
+		ResultProcessor processor = method.getResultProcessor().withDynamicProjection(resolvedAccessor);
+		SearchRepositoryQuerySupport.validateSort(resolvedAccessor);
 
-			String resolvedQuery = SearchBasedCouchbaseQuery.resolveParameters(searchQueryTemplate, resolvedAccessor);
-			SearchRequest request = SearchRequest.create(SearchQuery.queryString(resolvedQuery));
-			Object result = executeDependingOnType(resolvedAccessor, request);
+		String resolvedQuery = SearchBasedCouchbaseQuery.resolveParameters(searchQueryTemplate, resolvedAccessor);
+		SearchRequest request = SearchRequest.create(SearchQuery.queryString(resolvedQuery));
+		Object result = executeDependingOnType(resolvedAccessor, request);
 
-			return ReactiveWrapperConverters.toWrapper(processor.processResult(result), reactor.core.publisher.Flux.class);
-		});
+		return (Publisher<?>) processor.processResult(result);
 	}
 
 	@Override

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

@@ -15,14 +15,16 @@
  */
 package org.springframework.data.couchbase.core;
 
-import static org.junit.jupiter.api.Assertions.*;
+import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 
-import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * Unit tests for {@link SearchResult}.