Merge pull request #13962 from jdaugherty/rxjava3

Author: jdaugherty

gradle.properties

@@ -17,6 +17,7 @@ plexusSisuInjectVersion=2.6.0
 slf4jVersion=2.0.16
 rxJava1Version=1.3.8
 rxJava2Version=2.2.21
+rxJava3Version=3.1.10
 gparsVersion=1.2.1
 
 org.gradle.caching=true

grails-async/rxjava3/build.gradle

@@ -0,0 +1,13 @@
+dependencies {
+    implementation platform(project(':grails-bom'))
+
+    api project(':grails-async-core')
+    api "io.reactivex.rxjava3:rxjava:$rxJava3Version"
+
+    implementation 'org.apache.groovy:groovy'
+    implementation 'org.slf4j:slf4j-api'
+
+    testImplementation 'org.spockframework:spock-core'
+
+    testRuntimeOnly 'org.slf4j:slf4j-nop' // Get rid of warning about missing slf4j implementation during test task
+}

grails-async/rxjava3/src/main/groovy/org/grails/async/factory/rxjava3/RxPromise.groovy

@@ -0,0 +1,159 @@
+package org.grails.async.factory.rxjava3
+
+import grails.async.Promise
+import groovy.transform.AutoFinal
+import groovy.transform.CompileStatic
+import groovy.transform.PackageScope
+import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.core.Observer
+import io.reactivex.rxjava3.core.Scheduler
+import io.reactivex.rxjava3.core.Single
+import io.reactivex.rxjava3.core.SingleEmitter
+import io.reactivex.rxjava3.core.SingleOnSubscribe
+import io.reactivex.rxjava3.disposables.Disposable
+import io.reactivex.rxjava3.functions.Consumer
+import io.reactivex.rxjava3.functions.Function
+import io.reactivex.rxjava3.subjects.ReplaySubject
+import io.reactivex.rxjava3.subjects.Subject
+import org.grails.async.factory.BoundPromise
+
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
+
+/**
+ * Promise based on RxJava 3.x
+ *
+ * @since 7.0
+ */
+@AutoFinal
+@CompileStatic
+@PackageScope
+class RxPromise<T>  implements Promise<T> {
+
+    protected final Subject<T> subject
+    protected final RxPromiseFactory promiseFactory
+    protected final Observable<T> observable
+
+    protected Disposable subscription
+    protected boolean finished = false
+
+    RxPromise(RxPromiseFactory promiseFactory, Closure callable, Scheduler scheduler) {
+        this(promiseFactory, Single.create( { SingleEmitter<? super T> singleSubscriber ->
+            try {
+                singleSubscriber.onSuccess((T)callable.call())
+            } catch (Throwable t) {
+                singleSubscriber.onError(t)
+            }
+        } as SingleOnSubscribe<T>)
+                .subscribeOn(scheduler))
+    }
+
+    RxPromise(RxPromiseFactory promiseFactory, Observable single) {
+        this(promiseFactory, single, ReplaySubject.create(1))
+    }
+
+    RxPromise(RxPromiseFactory promiseFactory, Single single) {
+        this(promiseFactory, single, ReplaySubject.create(1))
+    }
+
+    RxPromise(RxPromiseFactory promiseFactory, Single single, Subject subject) {
+        this(promiseFactory, single.toObservable(), subject)
+    }
+
+    RxPromise(RxPromiseFactory promiseFactory, Observable observable, Subject subject) {
+        this.observable = observable
+        this.promiseFactory = promiseFactory
+        observable.subscribe(new Observer<T>() {
+            @Override
+            void onSubscribe(Disposable d) {
+                subscription = d
+            }
+
+            @Override
+            void onNext(T t) {
+                subject.onNext(t)
+            }
+
+            @Override
+            void onError(Throwable e) {
+                subject.onError(e)
+            }
+
+            @Override
+            void onComplete() {
+                finished = true
+            }
+        })
+        this.subject = subject
+    }
+
+    Observable<T> toObservable() {
+        return this.observable
+    }
+
+    @Override
+    Promise<T> accept(T value) {
+        return new BoundPromise<T>(value)
+    }
+
+    @Override
+    Promise<T> onComplete(Closure<T> callable) {
+        def decoratedCallable = promiseFactory.applyDecorators(callable, null)
+        return new RxPromise<T>(promiseFactory, subject.map(decoratedCallable as Function<T, T>))
+    }
+
+    @Override
+    Promise<T> onError(Closure<T> callable) {
+        def decoratedCallable = promiseFactory.applyDecorators(callable, null)
+        return new RxPromise<T>(promiseFactory, subject.doOnError(decoratedCallable as Consumer<Throwable>))
+    }
+
+    @Override
+    Promise<T> then(Closure<T> callable) {
+        return onComplete(callable)
+    }
+
+    @Override
+    boolean cancel(boolean mayInterruptIfRunning) {
+        if(subscription != null) {
+            subscription.dispose()
+            return subscription.isDisposed()
+        }
+        return false
+    }
+
+    @Override
+    boolean isCancelled() {
+        if(subscription == null) {
+            return false
+        }
+        else {
+            return subscription.isDisposed()
+        }
+    }
+
+    @Override
+    boolean isDone() {
+        return finished
+    }
+
+    @Override
+    T get() throws InterruptedException, ExecutionException {
+        return subject.blockingFirst()
+    }
+
+    @Override
+    T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+        try {
+            return subject.timeout(timeout, unit).blockingFirst()
+        } catch (Throwable e) {
+            if(e.cause instanceof TimeoutException) {
+                throw e.cause
+            }
+            else {
+                throw e
+            }
+        }
+    }
+}

grails-async/rxjava3/src/main/groovy/org/grails/async/factory/rxjava3/RxPromiseFactory.groovy

@@ -0,0 +1,78 @@
+package org.grails.async.factory.rxjava3
+
+import grails.async.Promise
+import grails.async.PromiseList
+import grails.async.factory.AbstractPromiseFactory
+import groovy.transform.AutoFinal
+import groovy.transform.CompileStatic
+import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.core.Single
+import io.reactivex.rxjava3.schedulers.Schedulers
+import org.grails.async.factory.BoundPromise
+
+import java.util.concurrent.TimeUnit
+
+/**
+ * An RxJava {@link grails.async.PromiseFactory} implementation
+ */
+@AutoFinal
+@CompileStatic
+class RxPromiseFactory extends AbstractPromiseFactory {
+
+    @Override
+    <T> Promise<T> createPromise(Class<T> returnType) {
+        new RxPromise<T>(this, Single.just(null))
+    }
+
+    @Override
+    Promise<Object> createPromise() {
+        new RxPromise<Object>(this, Single.just(null))
+    }
+
+    @Override
+    <T> Promise<T> createPromise(Closure<T>[] closures) {
+        if(closures.length == 1) {
+            return new RxPromise<T>(this, closures[0], Schedulers.io())
+        }
+        else {
+            def promiseList = new PromiseList()
+            for (Closure closure : closures) {
+                promiseList.add(closure)
+            }
+            return promiseList
+        }
+    }
+
+    @Override
+    <T> List<T> waitAll(List<Promise<T>> promises) {
+        return promises.collect { Promise<T> p -> p.get() }
+    }
+
+    @Override
+    <T> List<T> waitAll(List<Promise<T>> promises, long timeout, TimeUnit units) {
+        promises.collect { Promise<T> p -> p.get(timeout, units) }
+    }
+
+    @Override
+    <T> Promise<T> onComplete(List<Promise<T>> promises, Closure<T> callable) {
+        new RxPromise<T>(this, Observable.concat(
+            promises.collect { Promise p ->
+                if(p instanceof BoundPromise) {
+                    return Observable.just(((BoundPromise)p).value) as Observable<T>
+                }
+                else {
+                    return ((RxPromise)p).toObservable() as Observable<T>
+                }
+            }
+        ).toList())
+         .onComplete(callable)
+    }
+
+    @Override
+    <T> Promise<List<T>> onError(List<Promise<T>> promises, Closure<?> callable) {
+        new RxPromise(this, Observable.concat(
+                promises.collect { ((RxPromise<T>)it).toObservable() }
+        ).toList())
+        .onError(callable) as Promise<List<T>>
+    }
+}

grails-async/rxjava3/src/main/resources/META-INF/services/grails.async.PromiseFactory

@@ -0,0 +1 @@
+org.grails.async.factory.rxjava3.RxPromiseFactory

grails-async/rxjava3/src/test/groovy/org/grails/async/factory/rxjava3/RxPromiseListSpec.groovy

@@ -0,0 +1,80 @@
+package org.grails.async.factory.rxjava3
+
+import grails.async.PromiseList
+import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+class RxPromiseListSpec extends Specification {
+
+    void 'Test promise list handling'() {
+
+        when: 'a list of promises is created'
+            def list = new PromiseList()
+            list << { 1 }
+            list << { 2 }
+            list << { 3 }
+            def result = null
+            list.onComplete { result = it }
+
+        then: 'then the result from onComplete is correct'
+            new PollingConditions(timeout: 5).eventually {
+                result == [1,2,3]
+            }
+    }
+
+    void 'Test promise list handling with some async operations and some values'() {
+
+        when: 'a list of promises is created'
+            def list = new PromiseList()
+            list << { 1 }
+            list <<  2
+            list << { 3 }
+            def result = null
+            list.onComplete { result = it }
+
+        then: 'then the result from onComplete is correct'
+            new PollingConditions(timeout: 5).eventually {
+                result == [1,2,3]
+            }
+    }
+
+    void 'Test promise list with then chaining'() {
+
+        when: 'a promise list is used with then chaining'
+            def list = new PromiseList()
+            list << { 1 }
+            def promise = list
+                .then { it << 2; it }
+                .then {
+                    Thread.dumpStack()
+                    it << 3; it
+                }
+            def result = promise.get()
+
+        then: 'An appropriately populated list is produced'
+            result == [1,2,3]
+
+    }
+
+    void 'Test promise list with an exception'() {
+
+        given: 'a promise list with a promise that throws an exception'
+            def list = new PromiseList()
+            list << { 1 }
+            list << { throw new RuntimeException('bad') }
+            list << { 3 }
+
+        when: 'the list is completed'
+            def result = null
+            Throwable error = null
+            list.onComplete { result = it }
+            list.onError { error = it }.get()
+            list.get()
+
+        then: 'the onError handler is invoked with the exception'
+            thrown(RuntimeException)
+            !result
+            error
+            error.message == 'bad'
+    }
+}