codeconsole commented on code in PR #16690:
URL: https://github.com/apache/grails-core/pull/16690#discussion_r4332046546
##########
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. It's the bug
from my #16619 review, and it's on `7.0.x` too: before this PR, `onComplete`
and `onError` already read `toObservable()`.
- **A task:** the source is the cold `Single.create(..)`, so subscribing to
it here runs the task a second time, and the list gets the second run's value.
- **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, and `isDone()` on the chained promise stays `false`.
I ran the four features below on this branch with JDK 17:
| Call | At a474a77242 | With the two lines below |
|---|---|---|
| `Promises.onComplete([task]) { it }` | `[2]`, the task ran twice | `[1]`,
the task ran once |
| `Promises.onError([task]) { }` | `[2]`, the task ran twice | `[1]`, the
task ran once |
| `Promises.onComplete([task.then { it + 1 }]) { it }` | `TimeoutException`
after 5 seconds | `[2]`, the `then` callback ran once |
| `task.then { it + 1 }.done` | still `false` after 5 seconds | `true` |
The fix is two lines. They do what the RxJava 1 factory does: it reads
`p.subject` here, and its constructor subscribes the subject itself, so the
subject completes.
- here: `return ((RxPromise<T>) p).subject as Observable<T>`
- in the `RxPromise` constructor's observer: `subject.onComplete()` after
`finished = true`
Both are needed. With only the first, nothing completes the subject, so
every list hangs. The module's own tests didn't finish in 10 minutes. With
both, the four features and the module's 20 existing tests pass.
<details><summary>The four features</summary>
```groovy
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import spock.lang.Specification
import spock.util.concurrent.PollingConditions
import grails.async.Promise
import grails.async.Promises
void 'onComplete over a list runs each task once'() {
given: 'a task that counts its runs'
AtomicInteger runs = new AtomicInteger()
Promise<Integer> task = Promises.createPromise {
runs.incrementAndGet() }
when: 'the list completes'
List result = Promises.onComplete([task]) { it }.get(5,
TimeUnit.SECONDS)
then: 'the task ran once and the callback received its value'
task.get() == 1
result == [1]
runs.get() == 1
}
void 'onError over a list runs each task once'() {
given: 'a task that counts its runs'
AtomicInteger runs = new AtomicInteger()
Promise<Integer> task = Promises.createPromise {
runs.incrementAndGet() }
when: 'the list resolves without an error'
List result = Promises.onError([task]) { }.get(5,
TimeUnit.SECONDS)
then: 'the task ran once and the list holds its value'
task.get() == 1
result == [1]
runs.get() == 1
}
void 'onComplete over a list completes when a promise in it was chained
with then'() {
given: 'a promise chained from a task, with a callback that counts
its runs'
AtomicInteger callbacks = new AtomicInteger()
Promise<Integer> chained = Promises.createPromise { 1 }.then {
callbacks.incrementAndGet(); it + 1 }
when: 'the list completes'
List result = Promises.onComplete([chained]) { it }.get(5,
TimeUnit.SECONDS)
then: 'the callback received the chained value and the then callback
ran once'
result == [2]
callbacks.get() == 1
}
void 'a promise chained with then is done once it resolves'() {
when: 'a task is chained with then'
Promise<Integer> chained = Promises.createPromise { 1 }.then {
it + 1 }
then: 'it resolves and reports done'
chained.get(5, TimeUnit.SECONDS) == 2
new PollingConditions(timeout: 5, delay: 0.2).eventually {
assert chained.done
}
}
```
`task.get()` sits before the `runs` check so that both runs have finished by
then. Otherwise the original run could still be in flight.
</details>
--
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]
