RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
expensiveGeneration is not blocking, i.e. it returns a Publisher, then you can do this:Flux.range(0, 10)
.flatMap(x -> expensiveGeneration(), 4);
in the process of evaluating RxJava, i've run into strange behavior while attempting recursion. first, i noticed that RxJava is not stacksafe (please correct me if i'm wrong). second, while attempting to get around the stackoverflow issues, i've gotten into a situation where a very simple program seemingly hangs. here's the bit of code (in Kotlin) i'm working with:
package demos
import io.reactivex.rxjava3.core.Flowable
import io.reactivex.rxjava3.core.Single
import io.reactivex.rxjava3.schedulers.Schedulers
import java.util.concurrent.TimeUnit.SECONDS
fun main() {
fun incr(n: Int): Single<Int> = Single.just(n + 1)
fun numbers(n: Int, max: Int): Flowable<Int> = Flowable.just(n).concatWith(
if (n < max)
incr(n).observeOn(Schedulers.single()).toFlowable().concatMap { next -> numbers(next, max) }
else
Flowable.empty()
)
numbers(1, 1_000_000).sample(5, SECONDS).blockingForEach(::println)
}without the call to .observeOn(Schedulers.single()), the stack explodes... with it, the first few sampled numbers get printed and then... just... nothing, e.g.:
14737
19765
23398
23802would really appreciate any help/advice.
public String save(UserRequestDTO requestDTO) {
UserInfoEntity userInfoEntity = mapper.mapToUserInfoEntity(requestDTO);
return repository.save(userInfoEntity)
.flatMap((x) -> getUser(requestDTO.getUsername()).flatMap(y -> Mono.just(getUrl(y.getUsername())))).block();
}
Hi All,
I am using io.reactivex.rxjava2:rxjava:2.2.19 with Java-11 Ref: ReactiveX/RxJava#6701, but when I have observed Thread dump, Following is the content
"RxCachedThreadScheduler-512" #1715 daemon prio=5 os_prio=0 cpu=14871.41ms elapsed=3810.70s tid=0x00007fdfc8051800 nid=0x5b37 waiting on condition [0x00007fdefa1f6000]
java.lang.Thread.State: WAITING (parking)
at jdk.internal.misc.Unsafe.park(java.base@11.0.6/Native Method)
- parking to wait for <0x000000068f374f98> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(java.base@11.0.6/LockSupport.java:194)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(java.base@11.0.6/AbstractQueuedSynchronizer.java:2081)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(java.base@11.0.6/ScheduledThreadPoolExecutor.java:1170)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(java.base@11.0.6/ScheduledThreadPoolExecutor.java:899)
at java.util.concurrent.ThreadPoolExecutor.getTask(java.base@11.0.6/ThreadPoolExecutor.java:1054)
at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@11.0.6/ThreadPoolExecutor.java:1114)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@11.0.6/ThreadPoolExecutor.java:628)
at java.lang.Thread.run(java.base@11.0.6/Thread.java:834)Reason to check stacktrace is, I have around 1k threads are in waiting state and it happens only in some of the cases and its difficult to reproduce.
Please suggest if I am missing anything.
P.S. Upgraded RxJava1 to RxJava2 and replaced Observables with Flowable with backpressure.Buffer strategy.
Hi, I have the follow situation and need some help here.
A method that find if the payment exists, and if it exists I need some processes that returns a Single. But this does not works since maybe.flatMap requires a MaybeSource. There are any other pattern to do this?
In the end I need return the payment (and need that the flow be executed as well).
public Maybe<Payment> process(String id) {
return getPayment(id).flatMap(payment -> {
// do something that returns a Single source
});
}Hi @melston, thanks for your reply. I will try to explain...
I have a database with a payment entities that represents the payment requests. When I call the process(String id), method it will find a Payment on database (that could not be found, so getPayment(id) is a Maybe. Right?
Ok, if the payment exists on database, I need to process this payment on a gateway (some actions that process this payment on a payment gateway). This actions are network calls to register this payment to a gateway. After all, I need that this method process returns the current (maybe) payment.
All network calls on the gateway return a Single<?> response.
I more elaborated code to try to explain it better. I trying to chain some invocations that depends on a first invocation (getPayment)
public Maybe<Payment> getPayment(String id) {
// lookup a payment in database and return it as a maybe
}
public Single<?> registerPayment(Payment payment) {
// register this payment on a payment gateway.
}
// This method contains a error cause the flatMap only accetps a MaybeSource and registerPayment returns a Single.
public Maybe<Payment> process(String id) {
return getPayment(id).flatMap(payment -> {
return registerPayment(payment);
});
}Remembering that at the end, I need to return the Maybe<Payment>.
1 reply
Hi all, I would like to ask you for an advice about retryWhen() operator. Here is an example code:
fun testRetry() {
val publisher = PublishSubject.create<ChangedEntity>()
val onNext: (ChangedEntity) -> Unit = { println("onNext") }
val onError: (Throwable) -> Unit = { println("onError") }
publisher
.map {
if (Random().nextInt(10) % 3 == 0) {
println("Throwing error")
throw RuntimeException(TimeoutException())
}
else {
println(it)
it
}
}
.retryWhen { errors ->
errors.flatMap {
if (it.cause is TimeoutException) {
println(">>>>>>>>>>>>>>>>> TimeoutException, retry")
Observable.just(null)
} else {
Observable.error<Throwable>(it)
}
}
}
.subscribe(onNext, onError)
publisher.onNext(createChangedEntity())
publisher.onNext(createChangedEntity())
publisher.onNext(createChangedEntity())
publisher.onNext(createChangedEntity())
publisher.onNext(createChangedEntity())
publisher.onNext(createChangedEntity())
publisher.onNext(createChangedEntity())
publisher.onNext(createChangedEntity())
}Basically map() operator can have a code which can trow an exception (contains a call to another service) and I would like to always retry it for certain exceptions, but I can't figure out how to use it. In this example the line with the TimeoutException will be printed out but the executions is stopped, there is no re-subscription.
Any idea what I'm doing wrong?
If you have any related experience, we highly appreciate your input!
Thanks in advance and the survey is available at https://usi.eu.qualtrics.com/jfe/form/SV_cO6Ayah0D6q4eSF