RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
There is an overload of
http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#onBackpressureBuffer(long,%20io.reactivex.functions.Action,%20io.reactivex.BackpressureOverflowStrategy)
onBackpressureBuffer where you can specify the overflow behavior.http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#onBackpressureBuffer(long,%20io.reactivex.functions.Action,%20io.reactivex.BackpressureOverflowStrategy)
and I also think it makes no sense to have any BackpressureStrategy other than MISSING in Flowable.create() if you plan to follow it with an onBackpressureXXX.
Hey! Are there some operators with which I can achieve something described in the following?
For example I have a inheritance heriachy like: Condition as parent and ConditionA, ConditionB, ConditionC are children of Condition.
Than I have a Method which looks like this:
public Observable<Condition> getCondition() {...}which returns indefinitely Condition's.
The goal would be a chaining an operator which caches the Condition's and when there is one available of each (ConditionA, ConditionB, ConditionC),
it should should filter them as described in the filter.
Maybe something like this:
getCondition()
.cacheFilter<ConditionA, ConditionB, ConditionC>((a, b, c) -> a.value == 0 && b.value == 0 && c.value == 0)
.do(somethingA())
.cacheFilter<ConditionA, ConditionB, ConditionC>((a, b, c) -> a.value == 0 && b.value == 1 && c.value == 1)
.do(somethingB())
.cacheFilter<ConditionA, ConditionB, ConditionC>((a, b, c) -> a.value == 1 && b.value == 0 && c.value == 0)
.do(somethingB())
.subscribe();
@Cir0X Instead of emitting all conditions on 1 stream, you should have 3 streams and combine these streams. Could be done with a number of operators, my gut feeling is
Next there are a few options to act on this stream. I wouldn't filter if you want to do all operations on the same stream, as filter takes out elements and you wont be able to act on them after that. Instead you can make a large
If you subscribe multiple times, that means the source stream will also run multiple times. Depending on the source, you might want to multicast instead. You can do that with
Another interesting solution would be to use
combineLatest(conditionA, conditionB, conditionC) would work best. You also seem to be doing a.value everywhere, you could add in a .map(a -> a.value) so you can look at the value directly later.Next there are a few options to act on this stream. I wouldn't filter if you want to do all operations on the same stream, as filter takes out elements and you wont be able to act on them after that. Instead you can make a large
do or subscribe block where you test all conditions one by one, or you can simply subscribe to the condition stream multiple times, and filter out the correct elements. Myself i would then do .filter((arr) -> !arr[0]).filter((arr) -> !arr[1]).filter((arr) -> !arr[2]) instead of !arr[0] && !arr[1] && !arr[2].If you subscribe multiple times, that means the source stream will also run multiple times. Depending on the source, you might want to multicast instead. You can do that with
share().Another interesting solution would be to use
.publish(selector), like this:getCondition().publish(_conditions -> Observable.merge(
_conditions.filter(a -> a[0] == 0).filter(a -> a[1] == 0).filter(a -> a[2] == 0).do(() -> sometihngA()),
_conditions.filter(a -> a[0] == 0).filter(a -> a[1] == 1).filter(a -> a[2] == 1).do(() -> sometihngB()),
_conditions.filter(a -> a[0] == 0).filter(a -> a[1] == 1).filter(a -> a[2] == 0).do(() -> sometihngC())
)).subscribe();
Mmm looking back i think the
&& style would be just as good :)
rxjava 2.* doesn't have that anymore
Thanks!
so if I have a subject and I want to turn it to an observable, what's a good way to do it?
.toFlowable(BackpressureStrategy.MISSING)
or whatever overflow strategy you prefer maybe .toFlowable(BackpressureStrategy.BUFFER)
I just found
hide
can I use that too?
I don't understand, both Subject, Observable, Flowables are some kind of Publishers as they allow subscribing to them. I understand that Observables are kind of deprecated in rxjava2 in favor of Flowables which force you to think about backpressure upfront(you need to specify a backpressure strategy when you create them), so why would you need an Observable. Anyway looks like a Flowable can simply be turned into an Observable by doing flowable.toObservable
also my question was unclear. On a
Subject you can do a onNext
(which is essentially publishing to other
Observers)
and I wanted to hide the
onNext
I didn't know that Observables are in the process of being deprecated?
I'd say not deprecated, but not much different than Flowable as I do not see reasons where I'd want to use Observables instead of Flowables(which have the benefit that you are aware of backpressure and what the backpressure strategy is and you won't be surprised at runtime by a MissingBackpressuException)
which is also unbounded
but at least you know upfront you might expect an OOM
Hi, I have question.
I Have this scenario - I want to make a remote call and get results, I want to do something with the data, and then notify to subscribers that this call has completed.
I dont want that every subscriber will be able to make this remote call, So i want someone to request the call and other to subscribe to its completion.
I Have this scenario - I want to make a remote call and get results, I want to do something with the data, and then notify to subscribers that this call has completed.
I dont want that every subscriber will be able to make this remote call, So i want someone to request the call and other to subscribe to its completion.
Got a ~10k line project , and just received
MissingBackPressureException which isn't picked up by my unit tests. How do I debug this?Caused by: rx.exceptions.MissingBackpressureException
at rx.internal.util.RxRingBuffer.onNext(RxRingBuffer.java:353)
at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onNext(OperatorZip.java:319)
at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:77)
at rx.observers.SerializedObserver.onNext(SerializedObserver.java:91)
at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:94)
at rx.internal.operators.OperatorSampleWithTime$SamplerSubscriber.emitIfNonEmpty(OperatorSampleWithTime.jav
at rx.internal.operators.OperatorSampleWithTime$SamplerSubscriber.call(OperatorSampleWithTime.java:98)
is it safe to issue a dontwarn for it?
Also, while I'm here: In RxJava1 I would occasionally use a
Subject as the Subscriber to an Observable, i. e. "pipe" the output of the Observable into the Subject. I can't find a way to do this in RxJava2 - subscribe(Observer) does accept the Subject, but does not return a Disposable, and since Subject is not a DisposableObserver, I can't use subscribeWith either... Should I just wrap my Subject in a DisposableObserver before subscribing or is there a simpler way I am missing?
PublishSubject.<Message>create().toSerialized()