RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
https://stackoverflow.com/questions/61814106/choice-between-groupby-and-filter-with-infinite-streams-in-reactor
Hi everybody,
In some cases, I need to apply a filter operation on the mainstream which afterward I need to process both results branched, not only the true one. So I can avoid filtering one stream twice, once to take the true branch and the other to take the false branch.
Do we have any such operator in RxJava2?
Thanks
1 reply
Hi, I have two unrelated Maybes that I need to chain in a single stream (to keep the order of execution). The current solution is:
Redis.rxHget(hashName, "value")
.doOnSuccess(to do in success case ...)
.flatMap(response -> Redis.rxHset(args)
.subscribe()The issue with the above approach is that if the first Maybe resulted in empty, the second one will not be triggered. So I was wondering is there any standard workaround for such a scenario, other than using the toSingle(default value) method after the first maybe?
Thanks
3 replies
Hi everybody,
Suppose that we have a list of items that we need to do the same action on each.
Regarding the performance optimization, I want to know is there any differences between the following approaches in the RxJava:
Flowable.fromIterable(itemList)
.flatMap(item -> client.rxQuery(item))
.map(...)
....
.subscribe()Or
for (Item item : itemList){
.client.rxQuery(item)
.map(...)
.subscribe()
}In the first approach, we create a flowable of the list's items and then apply the logic but in the second one, we use a for loop and will have an individual subscribe method call for each item.
Definitely, the first approach fits better into the RxJava pattern, but technically speaking, having an individual subscribe call as in the second approach will bring us more overloads?
Thanks
Hello ! Question regarding Flowable.using(): Flowable.using() is cancel() upstream on onComplete() and onError() (see: https://github.com/ReactiveX/RxJava/blob/79f8e6dde6446b1aa33c146eaedbb958086daf56/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableUsing.java#L146). I was wondering whether this could be considered a violation of 2.3 (see: https://github.com/reactive-streams/reactive-streams-jvm#2-subscriber-code)
Subscriber.onComplete() and Subscriber.onError(Throwable t) MUST NOT call any methods on the Subscription or the Publisher.
Still related to cancel()ation, with adoOnCancel(action), it seems a late cancel() received after a onComplete() or onError() would trigger the action while 2.4 mentions:
Subscriber.onComplete() and Subscriber.onError(Throwable t) MUST consider the Subscription cancelled after having received the signal.
And 3.7 mentions:
After the Subscription is cancelled, additional Subscription.cancel() MUST be NOPs.
doOnCancel is somewhat deliberate. It will execute the lambda upon cancellation independent if the sequence terminates or not.
upstream.cancel was a remnant from an early transition to Reactive Streams apparently and has no unit tests associated with the behavior. The side-effect is that nested usings would trigger the outer's cancel path before its onComplete chain has finished, releasing the outer source while the inner is still in the completion phase.
Flowable.just(string)
.doOnNext(this::save)
.map(val -> Flowable.zip(async1(val), async2(val), async3(val), this::collector)
.parallel().runOn(Schedulers.computation()))
.doOnComplete(() -> sendAck(channel, tag))
.subscribe(System.out::println);
i am learning rxjava and would like to see the possibility if someone could review my code and improve it.
specifically i want to see how to program nested observables
http://ix.io/2Hly
@blackdev1l How are those async methods implemented? map and parallel won't work in that setup and when you find yourself trying to map to some Flowable, that's an indication you most likely need flatMap.
Flowable.just(string)
.doOnNext(this::save)
.flatMap(val -> Flowable.zip(async1(val), async2(val), async3(val), this::collector))
.doOnComplete(() -> sendAck(channel, tag))
.subscribe(System.out::println);For this to go parallel with the asyncs, they could look something like this:
public Flowable<T> async1(String val) {
return Flowable.fromCallable(() -> process(val)).subscribeOn(Schedulers.computation());
}create(). Is it an Iterable, or Stream? fromXXX.
An advantage of this approach is that when you have a bunch of tasks that are not dependent on each other, you can start them all at the same time rather than waiting for each one to finish before starting the next one — that way, your entire bundle of tasks only takes as long to complete as the longest task in the bundle.
Yeah sure seldomly, but what happens when you try to do too many things at once, when you try to multi-task?!?! I know that no matter how intelligent you are, and no matter how MUCH YOU THINK you got your sh_t together, you are bound to falter and miss a few detailed major things when you MULTI-TASKS. So what do you think is going to happen when you WIRE A COMPUTER THAT WAY.
Hi everybody,
I'm seeking a solution for error handling of the time a connection exception happens. In my case, the connection has been established to RMQ:
rabbitMQClient.rxStart()
.doOnComplete(() -> logger.info("RMQ Client connected successfully"))
.doOnError(error -> logger.error("RMQ connection error: ", error))
.retryWhen(throwableFlowable -> throwableFlowable.delay(5, TimeUnit.SECONDS))
....The above chain is of the type Completable. My desire logic is that in time of the connection exception it should be trying to reconnect with a customized delay. For example, at first, it waits 1 second and then tries to connect, the second time it waits 2 seconds and then tries to reconnect, and so on until a max number of reties.
The current version will try to reconnect every 5 seconds, but can someone give me a hint on how I should implement it to work as I described?
Thanks