DATACOUCH-401 Fix incorrect return type for reactive string based N1ql count query

Motivation
----------
Reactive string based N1QL count query incorrectly returns the map after
the single result projection.

Changes
-------
Map the observable returning from the N1ql query result deserialized to
a map by converting to the correct return type using the conversion
service. Verified by integration tests.

src/integration/java/org/springframework/data/couchbase/repository/ReactiveN1qlCouchbaseRepositoryTests.java

@@ -111,4 +111,16 @@ public class ReactiveN1qlCouchbaseRepositoryTests {
 		}
 		assertNotNull("Expected to find several parties", previousDesc);
 	}
+
+    @Test
+    public void testCustomSpelCountQuery() {
+        long count = partyRepository.countCustom().block();
+        assertEquals("Test N1QL Spel based query", 15, count);
+    }
+
+    @Test
+    public void testPartTreeQuery() {
+        long count = partyRepository.countAllByDescriptionNotNull().block();
+        assertEquals("Test N1QL part tree based query", 15, count);
+    }
 }

src/integration/java/org/springframework/data/couchbase/repository/ReactivePartyRepository.java

@@ -37,7 +37,7 @@ public interface ReactivePartyRepository extends ReactiveCouchbaseRepository<Par
 	@Query("SELECT count(*) + 5 FROM #{#n1ql.bucket} WHERE #{#n1ql.filter}")
 	Mono<Long> countCustomPlusFive();
 
-	@Query("#{#n1ql.selectEntity} WHERE #{#n1ql.filter}")
+	@Query("SELECT count(*) FROM #{#n1ql.bucket} WHERE #{#n1ql.filter}")
 	Mono<Long> countCustom();
 
 	@Query("SELECT 1 = 1")

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

@@ -16,13 +16,12 @@
 package org.springframework.data.couchbase.repository.query;
 
 import java.util.Map;
-import java.util.Optional;
-
 import com.couchbase.client.java.document.json.JsonValue;
 import com.couchbase.client.java.query.N1qlQuery;
 import com.couchbase.client.java.query.Statement;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.CouchbaseQueryExecutionException;
 import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations;
 import org.springframework.data.couchbase.repository.query.support.N1qlUtils;
 import org.springframework.data.repository.query.*;
@@ -80,7 +79,7 @@ public abstract class ReactiveAbstractN1qlBasedQuery implements RepositoryQuery
         if (queryMethod.isQueryForEntity()) {
             return execute(query, typeToRead);
         } else {
-            return executeSingleProjection(query);
+            return executeSingleProjection(query, typeToRead);
         }
     }
 
@@ -95,9 +94,17 @@ public abstract class ReactiveAbstractN1qlBasedQuery implements RepositoryQuery
         return couchbaseOperations.findByN1QL(query, typeToRead);
     }
 
-    protected Object executeSingleProjection(N1qlQuery query) {
+    protected Object executeSingleProjection(N1qlQuery query, final Class<?> typeToRead) {
         logIfNecessary(query);
-        return couchbaseOperations.findByN1QLProjection(query, Map.class);
+        return couchbaseOperations.findByN1QLProjection(query, Map.class)
+                .map(m -> {
+                        if (m.size() > 1) {
+                            throw new CouchbaseQueryExecutionException("Query returning primitive got more values than expected: "
+                                    + m.size());
+                        }
+                        Object v = m.values().iterator().next();
+                        return this.couchbaseOperations.getConverter().getConversionService().convert(v, typeToRead);
+                    });
     }
 
     @Override