FMWK-549 Align executing reactive query method with sync flow (#792)

src/main/java/org/springframework/data/aerospike/repository/query/AerospikePartTreeQuery.java

@@ -86,6 +86,7 @@ public Object execute(Object[] parameters) {
         } else if (queryMethod.isStreamQuery()) {
             return findByQuery(query, targetClass);
         } else if (queryMethod.isCollectionQuery()) {
+            // All queries with Collection return type including projections
             return findByQuery(query, targetClass).collect(Collectors.toList());
         } else if (queryMethod.isQueryForEntity()) {
             Stream<?> result = findByQuery(query, targetClass);

src/main/java/org/springframework/data/aerospike/repository/query/BaseAerospikePartTreeQuery.java

@@ -114,7 +114,7 @@ Class<?> getTargetClass(ParametersParameterAccessor accessor) {
             return accessor.findDynamicProjection();
         }
         // DTO projection
-        if (queryMethod.getReturnedObjectType() != queryMethod.getEntityInformation().getJavaType()) {
+        if (!isEntityAssignableFromReturnType(queryMethod)) {
             return queryMethod.getReturnedObjectType();
         }
         // No projection - target class will be the entity class.
@@ -170,4 +170,15 @@ protected boolean isCountQuery(QueryMethod queryMethod) {
     protected boolean isDeleteQuery(QueryMethod queryMethod) {
         return queryMethod.getName().startsWith("deleteBy") || queryMethod.getName().startsWith("removeBy");
     }
+
+    /**
+     * Find whether entity domain class is assignable from query method's returned object class.
+     * Not assignable when using a detached DTO (data transfer object, e.g., for projections).
+     *
+     * @param queryMethod QueryMethod in use
+     * @return true when entity is assignable from query method's return class, otherwise false
+     */
+    protected boolean isEntityAssignableFromReturnType(QueryMethod queryMethod) {
+        return queryMethod.getEntityInformation().getJavaType().isAssignableFrom(queryMethod.getReturnedObjectType());
+    }
 }

src/main/java/org/springframework/data/aerospike/repository/query/ReactiveAerospikePartTreeQuery.java

@@ -55,6 +55,7 @@ public ReactiveAerospikePartTreeQuery(QueryMethod queryMethod,
     }
 
     @Override
+    @SuppressWarnings({"NullableProblems"})
     public Object execute(Object[] parameters) {
         ParametersParameterAccessor accessor = new ParametersParameterAccessor(queryMethod.getParameters(), parameters);
         Query query = prepareQuery(parameters, accessor);
@@ -100,8 +101,19 @@ public Object execute(Object[] parameters) {
                 }
                 return getPage(unprocessedResults, size, pageable, query);
             });
+        } else if (queryMethod.isStreamQuery()) {
+            return findByQuery(query, targetClass).toStream();
+        } else if (queryMethod.isCollectionQuery()) {
+            // Currently there seems to be no way to distinguish return type Collection from Mono<Collection> etc.,
+            // so a query method with return type Collection will compile but throw ClassCastException in runtime
+            return findByQuery(query, targetClass).collectList();
         }
-        return findByQuery(query, targetClass);
+         else if (queryMethod.isQueryForEntity() || !isEntityAssignableFromReturnType(queryMethod)) {
+            // Queries with Flux<Entity> and Mono<Entity> return types including projection queries
+            return findByQuery(query, targetClass);
+        }
+        throw new UnsupportedOperationException("Query method " + queryMethod.getNamedQueryName() + " is not " +
+            "supported");
     }
 
     protected Object runQueryWithIds(Class<?> targetClass, List<Object> ids, Query query) {

src/test/java/org/springframework/data/aerospike/repository/query/reactive/ReactiveCustomerRepositoryQueryTests.java

@@ -6,6 +6,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.aerospike.BaseReactiveIntegrationTests;
 import org.springframework.data.aerospike.sample.Customer;
+import org.springframework.data.aerospike.sample.ReactiveCustomerNegativeTestsRepository;
 import org.springframework.data.aerospike.sample.ReactiveCustomerRepository;
 
 import java.util.List;
@@ -34,6 +35,8 @@ public class ReactiveCustomerRepositoryQueryTests extends BaseReactiveIntegratio
 
     @Autowired
     protected ReactiveCustomerRepository reactiveRepository;
+    @Autowired
+    protected ReactiveCustomerNegativeTestsRepository negativeTestsReactiveRepository;
 
     @BeforeAll
     void beforeAll() {

src/test/java/org/springframework/data/aerospike/repository/query/reactive/find/NotEqualTests.java

@@ -3,22 +3,49 @@
 import org.junit.jupiter.api.Test;
 import org.springframework.data.aerospike.repository.query.reactive.ReactiveCustomerRepositoryQueryTests;
 import org.springframework.data.aerospike.sample.Customer;
-import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
 
 import java.util.List;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /**
  * Tests for the "Is not equal" reactive repository query. Keywords: Not, IsNot.
  */
 public class NotEqualTests extends ReactiveCustomerRepositoryQueryTests {
 
     @Test
-    public void findBySimplePropertyNot() {
-        List<Customer> results = reactiveRepository.findByLastNameNot("Simpson")
-            .collectList().block();
+    public void findBySimplePropertyNotEqual_String() {
+        StepVerifier.create(reactiveRepository.findByLastNameNot("Simpson"))
+            .recordWith(List::of)
+            .consumeRecordedWith(customers -> {
+                assertThat(customers).containsExactlyInAnyOrderElementsOf(List.of(matt, leela, fry));
+            })
+            .expectComplete();
 
-        assertThat(results).contains(matt);
+        StepVerifier.create(reactiveRepository.findByFirstNameNotIgnoreCase("SimpSon"))
+            // this query returns Mono<Collection>
+            .expectNextMatches(customers -> {
+                assertThat(customers).containsExactlyInAnyOrderElementsOf(List.of(matt, leela, fry));
+                return false;
+            })
+            .expectComplete();
+
+        StepVerifier.create(reactiveRepository.findOneByLastNameNot("Simpson"))
+            // this query returns Mono<Customer>
+            .expectNextMatches(customer -> {
+                assertThat(customer).isIn(List.of(matt, leela, fry));
+                return false;
+            })
+            .expectComplete();
+
+        Stream<Customer> customersStream = reactiveRepository.findByFirstNameNot("Simpson");
+        assertThat(customersStream.toList()).containsExactlyInAnyOrderElementsOf(allCustomers);
+
+        assertThatThrownBy(() -> negativeTestsReactiveRepository.findByLastNameNotIgnoreCase("Simpson"))
+            .isInstanceOf(ClassCastException.class)
+            .hasMessageContaining("cannot be cast");
     }
 }

src/test/java/org/springframework/data/aerospike/sample/ReactiveCustomerNegativeTestsRepository.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2012-2024 the original author or authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 		https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.aerospike.sample;
+
+import org.springframework.data.aerospike.repository.ReactiveAerospikeRepository;
+
+import java.util.List;
+
+/**
+ * This repository acts as a storage for invalid method names used for testing. For actual repository see
+ * {@link ReactiveCustomerRepository}
+ */
+public interface ReactiveCustomerNegativeTestsRepository extends ReactiveAerospikeRepository<Customer, String> {
+
+    /**
+     * ClassCastException, cannot be automatically cast to sync List using reactive Repository.
+     * See {@link ReactiveCustomerRepository} for examples of used query methods return types
+     */
+    List<Customer> findByLastNameNotIgnoreCase(String lastName);
+}

src/test/java/org/springframework/data/aerospike/sample/ReactiveCustomerRepository.java

@@ -21,7 +21,9 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.Collection;
 import java.util.List;
+import java.util.stream.Stream;
 
 /**
  * Simple reactive repository interface managing {@link Customer}s.
@@ -40,6 +42,12 @@ public interface ReactiveCustomerRepository extends ReactiveAerospikeRepository<
 
     Flux<Customer> findByLastNameNot(String lastName);
 
+    Mono<Customer> findOneByLastNameNot(String lastName);
+
+    Stream<Customer> findByFirstNameNot(String lastName);
+
+    Mono<Collection<Customer>> findByFirstNameNotIgnoreCase(String lastName);
+
     Mono<Customer> findOneByLastName(String lastName);
 
     Flux<Customer> findByLastNameOrderByFirstNameAsc(String lastName);