RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
observe how i kept the subject as hidden as possible in my snipped
This case we had here is pretty much the only one where i dont know how to avoid the subject.
@mattblang I haven't used IRC in years I guess I'll find a new client :D
This is really good stuff .. covers almost everything in Rx with examples on where to use https://github.com/Froussios/Intro-To-RxJava
Hello, everyone.
Can someone please explain why do the following two snippets of code behave differently?
Observable.<Boolean>create(subscriber -> Observable.just(true).subscribe(subscriber))
.flatMap(
b -> Observable.<Integer>create(subscriber -> {
subscriber.onNext(1);
subscriber.onCompleted();
}).subscribeOn(Schedulers.io()) // to imitate async request
)
.subscribe(System.out::println);Observable.<Boolean>create(subscriber -> Observable.just(true).subscribe(subscriber::onNext, subscriber::onError, subscriber::onCompleted))
.flatMap(
b -> Observable.<Integer>create(subscriber -> {
subscriber.onNext(1);
subscriber.onCompleted();
}).subscribeOn(Schedulers.io()) // to imitate async request
)
.subscribe(System.out::println);When I run the first one it doesn’t print anything to the console, but the second one prints 1 as expected.
@yesenarman your first example has a race condition. When your observable chain subscribes to the subscribeOn operator a new thread on the io scheduler subscribes upward to the flatmap but your main thread returns from the subscribe and terminates the program. hence why you don't see it print. The other thread hasn't onNexted yet.
you can fix the race condition by using a TestSubscriber and calling awaitTerminalEventAndUnsubscribeOnTimeout(int, TimeUnit) or simply adding a countdown latch and awaiting after your subscribe
Hi. Is there an equivalent to
delaySubscription but for unsubscription? I'd like to be able to have the subscriber be effectively unsubscribed, but the observable should not be unsubscribed immediately (and sent data should be sent to oblivion). Use case: a GPS observable that takes time to acquire a fix initially that is share()d. A delayed unsubscription would let the GPS active for a few seconds in case another subscriber gets interested fast enough. I guess that could be a variant of refCount() too, that would take a delay before unsubscribing.
(I cooked up an operator for doing this if anyone is interested: https://gist.github.com/samueltardieu/12db44dc01340f22e9da)
(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?