codeconsole commented on code in PR #16619: URL: https://github.com/apache/grails-core/pull/16619#discussion_r4329746626
########## grails-doc/src/en/guide/async/rxjava.adoc: ########## @@ -17,23 +17,100 @@ specific language governing permissions and limitations under the License. //// -Since Grails 3.2, you can use https://github.com/ReactiveX/RxJava[RxJava] to write reactive logic in your Grails controllers that leverages the underlying containers asynchronous processing capabilities. +Grails integrates https://github.com/ReactiveX/RxJava[RxJava] 3 at three points. Each is a separate module that an application adds to `build.gradle`; Grails itself does not depend on RxJava. -To get started simply declare a dependency on the plugin in `build.gradle`: +==== Promises on RxJava + +`grails-async-rxjava3` provides a `grails.async.PromiseFactory` backed by RxJava 3, so every `Promise` the application creates is an RxJava subscription: + +[source,groovy] +.build.gradle +---- +runtimeOnly 'org.apache.grails.async:grails-async-rxjava3' +---- + +See <<asyncPromises,Promises>>. + +==== Events on RxJava + +`grails-events-rxjava3` provides an `EventBus` that delivers events through RxJava 3 subjects and schedulers: [source,groovy] .build.gradle ---- -dependencies { - //... - implementation 'org.apache.grails:grails-rxjava3' +runtimeOnly 'org.apache.grails.events:grails-events-rxjava3' +---- + +See <<events,Events>>. + +==== RxJava Data Services + +With `grails-datamapping-rx` on the classpath, a GORM data service can declare methods that return `io.reactivex.rxjava3.core.Observable` or `io.reactivex.rxjava3.core.Single`. GORM implements the method as it does any other data service method: the blocking query runs when the result is subscribed to, on the subscribing thread, and the result is delivered on RxJava's IO scheduler, so the caller consumes it reactively without the datastore itself being non-blocking: + +[source,groovy] +.build.gradle +---- +implementation 'org.apache.grails.data:grails-datamapping-rx' +---- + +[source,groovy] +---- +import grails.gorm.services.Service +import io.reactivex.rxjava3.core.Observable +import io.reactivex.rxjava3.core.Single + +@Service(Book) +interface BookService { + + Single<Book> findOne(String title) + + Observable<Book> findAllByTitle(String title) + + Observable<Book> findAllByAuthor(String author) +} +---- + +A `Single` that finds nothing signals a `java.util.NoSuchElementException`, since an RxJava 3 `Single` cannot complete without a value; an `Observable` that finds nothing completes without emitting. The `grails.gorm.rx.services.RxSchedule` annotation selects the scheduler a method delivers its result on. Neither it nor the default moves the query itself off the subscribing thread, so subscribe from a task, as <<asyncRequests,Asynchronous Request Handling>> describes, to keep a blocking query off a request thread. See the RxJava Support section of the GORM for link:{hibernate7Guide}index.html#dataServicesRxJava[Hibernate 7] or link:{hibernate5Guide}index.html#dataServicesRxJava[Hibernate 5] guide. + +==== RxJava Results from Controllers + +An RxJava 3 `Flowable` is a Reactive Streams `Publisher`, so an action can return one and Grails streams every item it emits to the client as a server-sent event, with no plugin. See <<asyncStreaming,Streaming Responses>>: + +[source,groovy] +---- +import io.reactivex.rxjava3.core.BackpressureStrategy + +class BookController { + + BookService bookService + + def feed(String author) { + bookService.findAllByAuthor(author) + .toFlowable(BackpressureStrategy.BUFFER) + } } ---- -You can then return `rx.Observable` as a return value from any controller and Grails will automatically apply the following steps: +Grails subscribes to the publisher on the request thread, so in this example the data service runs its query there, before the action returns, as the previous section describes; the items are then delivered on the IO scheduler and written to the response from it, with the container thread released while they are. Running the query itself on another thread with `subscribeOn(..)` needs a session on that thread, which the data services do not set up. Review Comment: The last sentence isn't quite what happens. I tried it on Hibernate 7 with an `Observable<Author> findAllByName(String)` data service. With `.subscribeOn(Schedulers.io())`, the query runs on `RxCachedThreadScheduler-2` and succeeds. The generated method is `@ReadOnly`, and its transaction opens a session on that thread. What fails is touching a lazy association on a result afterwards: `LazyInitializationException: Cannot lazily initialize collection of role ... (no session)`. Something like this would be accurate: "With `subscribeOn(..)` the query runs on the IO thread in the data service's own transaction, so its results have no session to load lazy associations from, and anything that reads the request, such as a tenant resolver, doesn't see it there." ########## grails-async/rxjava3/src/main/groovy/org/grails/async/factory/rxjava3/RxPromiseFactory.groovy: ########## @@ -76,24 +76,27 @@ class RxPromiseFactory extends AbstractPromiseFactory { @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) + new RxPromise<T>(this, values(promises)).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>> + new RxPromise(this, values(promises)).onError(callable) as Promise<List<T>> + } + + /** + * Concatenates the values of the given promises into one list, in order, with a {@code null} value restored + * from the sentinel it travels through the stream as. + */ + private static <T> Single<List<T>> values(List<Promise<T>> promises) { + List<Observable<T>> sources = promises.collect { Promise<T> p -> + if (p instanceof BoundPromise) { + return Observable.just(RxPromise.wrap(((BoundPromise) p).value)) as Observable<T> + } + return ((RxPromise<T>) p).toObservable() Review Comment: `toObservable()` returns the promise's source, not its result: - **A task:** the source is the cold `Single.create(..)`, so subscribing to it here runs the task again. - **A `then` or `onComplete` promise:** the source is `subject.map(..)` on the previous promise's `subject`. That subject never completes, because the constructor's observer sets `finished` but doesn't pass `onComplete` on. So `concat` waits forever. I ran the same calls against both modules: | Call | `grails-async-rxjava` | `grails-async-rxjava3` | |---|---|---| | `Promises.onComplete([task]) { }` | task runs once | task runs twice | | `Promises.onError([task]) { }` | task runs once | task runs twice | | `Promises.onComplete([task.then { }]) { }` | completes | never completes (`get(3, SECONDS)` times out), and the `then` callback runs twice | The RxJava 1 factory reads `((RxPromise) p).subject` here, and its subject completes. Two lines bring this module in line: - here: `return ((RxPromise<T>) p).subject as Observable<T>` - in the `RxPromise` constructor's observer: `subject.onComplete()` after `finished = true` With both changes, all three rows match RxJava 1, and the module's existing tests still pass. A feature with a `then`-chained promise in the list and a counter on the task would pin it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
