jdaugherty commented on code in PR #16619:
URL: https://github.com/apache/grails-core/pull/16619#discussion_r4337336872
##########
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:
Fixed in d006362dd6. Rather than reading `subject` in `values()`,
`toObservable()` itself returns the subject now, so nothing but the constructor
subscribes to the source; the class is package-scoped and `values()` is the
method's only caller. The constructor's observer calls `subject.onComplete()`
after `finished = true`.
Your four features are in: three in `RxPromiseListSpec` and the `then` one
in `RxPromiseSpec`. All four failed before the change (`[2]`, `[2]`, a
`TimeoutException`, and `done` still `false`), and `:grails-async-rxjava3:test`
(24 tests) and `codeStyle` pass with it. The same commit is on #16690
(592e8d95fc), so the merge-up stays clean.
##########
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:
Agreed, and it matches the transform: the `@Transactional` body is woven
inside the deferred callable, so `subscribeOn(..)` moves the transaction along
with the query. Reworded close to your suggestion in d006362dd6.
--
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]
