RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
(I can use it as
newObservable = observable.share().lift(new DelayedUnsubscription(5, TimeUnit.SECONDS)), and I then subscribe to newObservable)
@wKovacs64 I answered that on stack overflow a while back. See http://stackoverflow.com/questions/31733455/rxjava-observable-cache-invalidate/31738646#31738646
does anybody know what effect
onNext will have on a PublishSubject after PublishSubject#onCompleted has been called?
will the
onNext item be emitted, will it fail to be emitted silently, or will the call fail with an exception?
Try it? If the subject is implemented properly, it should silently ignore the onNext().
http://reactivex.io/RxJava/javadoc/rx/subjects/PublishSubject.html#onNext%28T%29
http://reactivex.io/RxJava/javadoc/rx/subjects/PublishSubject.html#onNext%28T%29
onNext [...] The Observable will not call this method again after it calls either Observer.onCompleted() or Observer.onError(java.lang.Throwable).
Hi, I have some flow which I can't resolve with provided operations. Or I just don't see something important.
I have sequence A which produces sequence B. Then later I combine them with combineLatest operator, and get two notifies on each A update. But I want one and latest.
Short illustration:
Observable<Long> o1 = Observable.interval(1, TimeUnit.SECONDS).map(x -> x + 1);
Observable<Long> o2 = o1.map(v -> -v);
Observable.combineLatest(o1, o2, (v1, v2) -> "o1: " + v1 + ", o2: " + v2)
.subscribe(System.out::println);Provides output like this:
o1: 1, o2: -1
o1: 1, o2: -2
o1: 2, o2: -2
o1: 3, o2: -2
o1: 3, o2: -3
o1: 3, o2: -4
o1: 4, o2: -4
o1: 4, o2: -5
o1: 5, o2: -5Ok. I missed one detail. B can be merged with C
Observable<Long> a = Observable.interval(1, TimeUnit.SECONDS).map(x -> x + 1);
Observable<Long> b = a.map(v -> -v);
Observable<Long> c = Observable.interval(500, TimeUnit.MILLISECONDS).map(x -> x + 1000);
Observable.zip(a, Observable.merge(b, c), (v1, v2) -> "o1: " + v1 + ", o2: " + v2).subscribe(System.out::println);Then zip does wrong job too
Alternative, you could use debounce to filter out the double results when a and b both yield, at the risk of losing c when it yields too soon after b.
oh, and i just realize, you need something like
combineLatestRight. Does that exists in RxJava?
thanks a lot!
Maybe Scan?
Thats what i can come up with in 15 seconds ;)
I'm not 100% sure if this is possible, but i suppose it should be, but you can also use window or groupBy to combine all elements within 1 minute and then only take the first N elements from the resulting sequences. Doing that with live data is easy, but i'm sure it's possible with pre-existing timestamp data too.
It does depend on what state you had in mind
Subscriptions only receive events, they do not update the source.
PublishSubject<Integer> seq = PublishSubject.create();
Observable<Integer> seq2= seq.map(x ->{
timesCalled += 1;
return x * 2;
});
seq2.subscribe(e -> System.out.println("Hello 1: " + e));
seq2.subscribe(e -> System.out.println("Hello 2: " + e));
seq.onNext(1);
System.out.println("Times map called: " + timesCalled);map in this case called twice. (Is that a error or not?). And I get 2 different objects in each subscription, expected same.
If it helps to understand my problem
I assume timesCalled is declared outside the scope of map, that's where the problem lies.
Unless that was intended
Or publish seq2 so that side effects are shared between the observables instead.
It works. But as far as I understand doc,
publish creates ConnectableObservable. Which waits connect before emitting somethig. Sounds like not really aimed at current problem, right? Or I get it wrong? Or I need publish just to create new observable which keeps needed objects.
Until I
map it again by the way