reactive flow: add support for returning stream, add reactive Custome…

…r negative tests repository

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

@@ -102,9 +102,10 @@ public Object execute(Object[] parameters) {
                 return getPage(unprocessedResults, size, pageable, query);
             });
         } else if (queryMethod.isStreamQuery()) {
-            throw new UnsupportedOperationException(
-                "Automatic converting of an async stream to a blocking sync Stream is not supported");
+            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();
         }
          else if (queryMethod.isQueryForEntity() || !isEntityAssignableFromReturnType(queryMethod)) {

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

@@ -2,11 +2,14 @@
 
 import org.junit.jupiter.api.Test;
 import org.springframework.data.aerospike.repository.query.reactive.ReactiveCustomerRepositoryQueryTests;
+import org.springframework.data.aerospike.sample.Customer;
 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.
@@ -37,5 +40,12 @@ public void findBySimplePropertyNotEqual_String() {
                 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

@@ -23,6 +23,7 @@
 
 import java.util.Collection;
 import java.util.List;
+import java.util.stream.Stream;
 
 /**
  * Simple reactive repository interface managing {@link Customer}s.
@@ -43,6 +44,8 @@ public interface ReactiveCustomerRepository extends ReactiveAerospikeRepository<
 
     Mono<Customer> findOneByLastNameNot(String lastName);
 
+    Stream<Customer> findByFirstNameNot(String lastName);
+
     Mono<Collection<Customer>> findByFirstNameNotIgnoreCase(String lastName);
 
     Mono<Customer> findOneByLastName(String lastName);