RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
like.. I understand that Flowable.timer( delay, unit) produces an observable that returns 0L after the specified delay
ahh.. and now I figure out why my next change didn't work either..
It might be worth reading an RxJava 2 book: https://www.packtpub.com/application-development/learning-rxjava
I'm playing with WebFlux, RxJava and Lagom all at the same time..
not in the same app, thankfully
but in parallel, trying to do the same thing in each
when response.cancel() is invoked i want to send a CANCEL request. But surely i don't want to send a CANCEL request when the response is completed.
i manage to solve that using .doOnCancel(), but i found the method naming setCancelled() to be a bit misleading
and it is named setCancellable
Why don't want you to send a CANCEL to whatever source when your generator decides to end the sequence?
I have a RequestState which can be shared amongst multiple request (see it like a transaction). this request state manage the cancellation of the whole transaction.
If RequestState.isCancelled() return true, subsequent request sharing the same RequestState will signal onError(CancelException) on the response Flowable.
I've used setCancellable() to plug the RxJava cancel() to this RequestState.
If RequestState.isCancelled() return true, subsequent request sharing the same RequestState will signal onError(CancelException) on the response Flowable.
I've used setCancellable() to plug the RxJava cancel() to this RequestState.
Not sure this is clear. But my bug was that on the first request completed, the RequestState.cancel() was invoked. The subsequent request were always cancelled without "reason".
the FlowableEmitter.setCancellable() looks more like a Flowable.doOnTerminate() to me
FloableEmitter.doOnTerminate()
RequestStare are designed to be shared. Everything works fine if i bridge the RequestState using the doOnCancel()
Flowable.create(...).doOnCancel(requestState::cancel) worked while Flowable.create(emitter -> emitter.setCancellable(requestState::cancel)) didn't
My point is just that setCancellable() is a misleading method name given that it is invoked when flowable is cancel() but also when onComplete() or onError().
because each requests is producing a Flowable<Response>
once a request is onComplete(), a Flowable.concat(), subscribe to the next one. If the RequestSTate is cancelled, then the processing will throw and stop subsequent request to be subscribed. Otherwise, the request will produce result, then onComplete(), then the Flowable.concat() will subscribe to the next one. And so on
So is
setDisposable. The behavior then is provided by the Emitter. I don't see any case for misleading naming, just misunderstanding. The main use case for setCancellable is to release resources created within the emitter callback which are scoped for the duration of the emission. If you have external resources that are shared, you'll need other means to release them considering how a solution would interact with it.
But of course YMMV, this is just a remark from a RxJava user :)
I should not have used setCancellable() for my use case, this is an error on my side. BUT i would probably not used it at the first place if it was named "setFinalizer" or sthg like that.
The governing rule here is 1.6:
"If a Publisher signals either onError or onComplete on a Subscriber, that Subscriber’s Subscription MUST be considered cancelled.
The intent of this rule is to make sure that a Subscription is treated the same no matter if it was cancelled, the Publisher signalled onError or onComplete."
People coming from different backgrounds tend to get confused by different naming. There is often discussion how to name components so that it doesn't imply too much but doesn't go under the radar either. setCancellable follows the typical bean-setter naming indicating you set some object on this emitter.
I agree that the naming makes sense, i just fall in a trap :)
(and girls)
I'm new here and in the Rx(Java/Swift) world, and I have a very dummy question :)
I want to create a PublishSubject, but how can I, upon declaration, specify actions that should be done when an observer subscribes, or when the subject gets disposed?
I want to create a PublishSubject, but how can I, upon declaration, specify actions that should be done when an observer subscribes, or when the subject gets disposed?
Thanks!
Mostly the disposing part
I tried this:
BehaviorSubject<T> valueSubject = (BehaviorSubject<T>) BehaviorSubject.create(new ObservableOnSubscribe<T>() {
@Override
public void subscribe(ObservableEmitter<T> emitter) throws Exception {
// do something here
}
});