Correct RxJava2 validation

settings.gradle

@@ -28,5 +28,6 @@ include 'test-suite-kotlin'
 micronautBuild {
     importMicronautCatalog()
     importMicronautCatalog("micronaut-reactor")
+    importMicronautCatalog("micronaut-rxjava2")
 }
 

validation/build.gradle

@@ -23,6 +23,7 @@ dependencies {
 
     testImplementation mn.micronaut.http.client
     testImplementation mn.micronaut.http.server.netty
+    testImplementation mnRxjava2.micronaut.rxjava2
     testImplementation libs.groovy.json
     testImplementation mn.micronaut.inject.java.test
     testImplementation mn.micronaut.jackson.databind

validation/src/main/java/io/micronaut/validation/validator/DefaultValidator.java

@@ -476,8 +476,11 @@ public <T> Publisher<T> validatePublisher(@NonNull ReturnType<?> returnType,
                     Flux.error(new ConstraintViolationException(violations));
             });
         }
-
-        return Publishers.convertPublisher(conversionService, output, ((ReturnType<Publisher>) returnType).getType());
+        Class<?> returnClass = returnType.getType();
+        if (!Publisher.class.isAssignableFrom(returnClass)) {
+            return (Publisher<T>) output;
+        }
+        return Publishers.convertPublisher(conversionService, output, (Class<Publisher>) returnClass);
     }
 
     /**

validation/src/test/groovy/io/micronaut/validation/validator/reactive/BookService.java

@@ -36,8 +36,8 @@ class BookService {
     }
 
     @Executable
-    void rxValidWithTypeParameter(Mono<List<@Valid Book>> books) {
-        books.block();
+    Mono<Void> rxValidWithTypeParameter(Mono<List<@Valid Book>> books) {
+        return books.then();
     }
 
     @Executable

validation/src/test/groovy/io/micronaut/validation/validator/reactive/BookServiceRxJava2.java

@@ -0,0 +1,49 @@
+package io.micronaut.validation.validator.reactive;
+
+import io.micronaut.context.annotation.Executable;
+import io.reactivex.Completable;
+import io.reactivex.Flowable;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+import jakarta.inject.Singleton;
+import org.reactivestreams.Publisher;
+
+import javax.validation.Valid;
+import javax.validation.constraints.NotBlank;
+import java.util.List;
+
+@Singleton
+class BookServiceRxJava2 {
+
+    @Executable
+    Publisher<@Valid Book> rxSimple(Publisher<@NotBlank String> title) {
+        return Single.fromPublisher(title).map(Book::new).toFlowable();
+    }
+
+    @Executable
+    Observable<@Valid Book> rxValid(Observable<@Valid Book> book) {
+        return book;
+    }
+
+    @Executable
+    Completable rxValidWithTypeParameter(Single<List<@Valid Book>> books) {
+        return books.ignoreElement();
+    }
+
+    @Executable
+    Maybe<@Valid Book> rxValidMaybe(Maybe<@Valid Book> book) { return book; }
+
+    @Executable
+    Publisher<@Valid Book> rxReturnInvalid(Publisher<@Valid Book> book) {
+        return Flowable.fromPublisher(book).map(b -> new Book(""));
+    }
+
+    @Executable
+    Maybe<Book> rxReturnInvalidWithoutValidation(Flowable<@Valid Book> books) {
+        return books.firstElement().map(v -> new Book(""));
+    }
+
+}
+
+

validation/src/test/groovy/io/micronaut/validation/validator/reactive/ReactiveMethodValidationSpec.groovy

@@ -57,6 +57,10 @@ class ReactiveMethodValidationSpec extends Specification {
                 [Mono.just("")] as Object[]
         )
 
+        then: "No errors because publisher is not executed"
+        violations.size() == 0
+
+        when:
         Mono.from(bookService.rxSimple(Mono.just(""))).block()
 
         then:

validation/src/test/groovy/io/micronaut/validation/validator/reactive/RxJava2MethodValidationSpec.groovy

@@ -0,0 +1,133 @@
+package io.micronaut.validation.validator.reactive
+
+import io.micronaut.context.ApplicationContext
+import io.micronaut.validation.validator.Validator
+import io.reactivex.Flowable
+import io.reactivex.Maybe
+import io.reactivex.Observable
+import io.reactivex.Single
+import org.reactivestreams.Publisher
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import spock.lang.AutoCleanup
+import spock.lang.Shared
+import spock.lang.Specification
+
+import javax.validation.ConstraintViolationException
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ExecutionException
+import java.util.regex.Pattern
+
+class RxJava2MethodValidationSpec extends Specification {
+
+    @Shared
+    @AutoCleanup
+    ApplicationContext applicationContext = ApplicationContext.run()
+
+    void "test reactive return type validation"() {
+        given:
+        BookServiceRxJava2 bookService = applicationContext.getBean(BookServiceRxJava2)
+
+        when:
+        Single<Book> single = Single.just(new Book("It"))
+        Single.fromPublisher(bookService.rxReturnInvalid(single.toFlowable())).blockingGet()
+
+        then:
+        ConstraintViolationException e = thrown()
+        e.message == 'publisher[]<T Book>.title: must not be blank'
+        e.getConstraintViolations().first().propertyPath.toString() == 'publisher[]<T Book>.title'
+    }
+
+    void "test reactive return type no validation"() {
+        given:
+        BookServiceRxJava2 bookService = applicationContext.getBean(BookServiceRxJava2)
+
+        when:
+        Single<Book> single = Single.just(new Book("It"))
+        bookService.rxReturnInvalidWithoutValidation(single.toFlowable()).blockingGet()
+
+        then:
+        noExceptionThrown()
+    }
+
+    void "test reactive validation with invalid simple argument"() {
+        given:
+        BookServiceRxJava2 bookService = applicationContext.getBean(BookServiceRxJava2)
+
+        when:
+        var validator = applicationContext.getBean(Validator)
+        var violations = validator.forExecutables().validateParameters(
+                bookService,
+                BookService.class.getDeclaredMethod("rxSimple", Publisher<String>),
+                [Flowable.just("")] as Object[]
+        )
+
+        then: "No errors because publisher is not executed"
+        violations.size() == 0
+
+        when:
+        Single.fromPublisher(bookService.rxSimple(Single.just("").toFlowable())).blockingGet()
+
+        then:
+        def e = thrown(ConstraintViolationException)
+        Pattern.matches('rxSimple.title\\[]<T [^>]*String>: must not be blank', e.message)
+        def path = e.getConstraintViolations().first().propertyPath.iterator()
+        path.next().getName() == 'rxSimple'
+        path.next().getName() == 'title'
+        path.next().isInIterable()
+
+    }
+
+    void "test reactive validation with valid argument"() {
+        given:
+        BookServiceRxJava2 bookService = applicationContext.getBean(BookServiceRxJava2)
+
+        when:
+        def input = Observable.just(new Book("It"))
+        def book = bookService.rxValid(input).blockingFirst()
+
+        then:
+        book.title == 'It'
+    }
+
+    void "test reactive maybe validation with valid argument"() {
+        given:
+        BookServiceRxJava2 bookService = applicationContext.getBean(BookServiceRxJava2)
+
+        when:
+        def input = Maybe.just(new Book("It"))
+        def book = bookService.rxValidMaybe(input).blockingGet()
+
+        then:
+        book.title == 'It'
+    }
+
+    void "test reactive validation with invalid argument"() {
+        given:
+        BookServiceRxJava2 bookService = applicationContext.getBean(BookServiceRxJava2)
+
+        when:
+        def input = Observable.just(new Book(""))
+        bookService.rxValid(input).blockingFirst()
+
+        then:
+        def e = thrown(ConstraintViolationException)
+        Pattern.matches('rxValid.book\\[]<T .*Book>.title: must not be blank', e.message)
+        e.getConstraintViolations().first().propertyPath.toString().startsWith('rxValid.book')
+    }
+
+    void "test reactive validation with invalid argument type parameter"() {
+        given:
+        BookServiceRxJava2 bookService = applicationContext.getBean(BookServiceRxJava2)
+
+        when:
+        def input = Single.just([new Book("It"), new Book("")])
+        bookService.rxValidWithTypeParameter(input).blockingAwait()
+
+        then:
+        def e = thrown(ConstraintViolationException)
+        Pattern.matches('rxValidWithTypeParameter.books\\[]<T List>\\[1]<E Book>.title: must not be blank', e.message)
+        e.getConstraintViolations().first().propertyPath.toString().startsWith('rxValidWithTypeParameter.books')
+    }
+
+}