RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
var source = Observable.Interval(TimeSpan.FromSeconds(0.1)).Take(10);
''var group = source.GroupBy(i => i % 3);
group.SelectMany(
grp =>
grp.Max()
.Select(value => new { grp.Key, value }))
.Dump("group");''
''var group = source.GroupBy(i => i % 3);
group.SelectMany(
grp =>
grp.Max()
.Select(value => new { grp.Key, value }))
.Dump("group");''
::hh::
How do u format
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?
