RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
Flowable.range(2, 200_000_000)
.groupBy(e -> {
// First 100 000 000 in different group
if (e < 100_000_000) {
return e;
}
// Last 100 000 000 in the same group
return 1;
})
.subscribe(g -> {
if (g.getKey() == 1) {
g.subscribe(TestSubscriber.create(0));
} else {
g.subscribe(new DefaultSubscriber<Integer>() {
@Override
public void onNext(Integer t) {
if (t.intValue() % 100_000 == 0) {
System.out.println("Received: " + t);
}
Schedulers.computation().scheduleDirect(this::cancel);
}
@Override
public void onError(Throwable t) {
}
@Override
public void onComplete() {
}
});
}
});Hope that helps.
I need it for a scenegraph implementation
@ylecaillez
'''
'''
'''code'''
Ohhh
var code
I'm on mobile how do I line break
Hmm idk
That might work
Code
Hey
No boots
\help
Gitterbot helps kinda spammy but fun
I wish I could get the one from gamedev
Josh from undead labs knows the guy
I have a need to do something once a stream is subscribed to for the first time. I have written the following transformer to do it:
public static <T> Observable.Transformer<T, T> doOnFirstSubscribe(final Action0 func) {
return new Observable.Transformer<T, T>() {
boolean firstSubscription = true;
@Override
public Observable<T> call(Observable<T> input) {
return input.doOnSubscribe(() -> {
if (firstSubscription) {
func.call();
firstSubscription = false;
}
});
}
};
}i've written tests for it and they pass however during usage in my application it appears to call func when the stream is resubscribed to.
does my transformer seem correct?
why are
doOnComplete, doOnSubscribe and doFinally never called?// val obs = createObservable()
val obs = createcompletable()
@JvmStatic fun main(args: Array<String>) {
this.obs.doOnSubscribe {
println("on subscribe!")
}
this.obs.doOnComplete {
println("complete!")
}
this.obs.doFinally{
println("dofinally!")
}
this.obs.subscribe {
println("subscribe!")
}
}
fun createcompletable(): Completable =
Completable.create{ emitter ->
println("calling oncomplete")
emitter.onComplete()
}
fun createObservable(): Observable<Void> =
// Observable.defer {
Observable.create<Void> { emitter ->
println("calling oncomplete")
emitter.onComplete()
}
// }
+ also, the
subscribe block is called only when using Completable, with Observable<Void> not even that is called
Hi everyone, I am using Rxjava 2 and its hook mechanism to pass threadlocal value arround. It's work. However when calling another non blocking api (Couchbase api which uses Rxjava 1), threadlocal was lost after receiving results because couchbase receive couchbase response on its own thread. Anyone have the solution for this problem?
Hello guys !
While working on a custom operator, i was wondering about the correct behavior to implement when my operator deals with a Subscriber throwing an exception on onNext().
Reactive-streams 2#13 (https://github.com/reactive-streams/reactive-streams-jvm#2.13) says that the caller "MUST raise this error condition in a fashion that is adequate for the runtime environment."
Reading at RxJava code, looks like onNext() invocations are never (? i admit i didn't check all operators) surrounded by try/catch.
So i wondered if it is something planned to be included in a future RxJava or if you guys actually had a reason to not do so ?
While working on a custom operator, i was wondering about the correct behavior to implement when my operator deals with a Subscriber throwing an exception on onNext().
Reactive-streams 2#13 (https://github.com/reactive-streams/reactive-streams-jvm#2.13) says that the caller "MUST raise this error condition in a fashion that is adequate for the runtime environment."
Reading at RxJava code, looks like onNext() invocations are never (? i admit i didn't check all operators) surrounded by try/catch.
So i wondered if it is something planned to be included in a future RxJava or if you guys actually had a reason to not do so ?
There is no need to wrap
onNext calls into try-catches. In fact, it is discouraged in RxJava 2 because it can misdirect exceptions from bugs. If you have something that can throw inside your onNext, wrap that into try-catch and don't let it escape. Btw, are you learning RxJava 2?
The
Function can throw so it is wrapped and managed. onNext should not throw as the specification mandates thus is not wrapped.