RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
let me try it
Still missing
sb 0
Oh right, there is no
sb 0, as that's (0,1) -> 1
@Override
public Observable<R> call(Observable<T> source) {
Observable<T> source2 = source.share();
Observable<T> offsetSource = source2.skip(1);
return source2.zipWith(offsetSource, mergeFunc);
}
updated
I need to build a regular Java project
do 0
do 1
sb 0 1
do 2
sb 1 2
do 3
sb 2 3
Thats what i see with share()
do 1
sb 0 1
do 2
sb 1 2
do 3
sb 2 3
Thats what i see with share()
Perhaps with observable.create?
@Override
public Observable<R> call(Observable<T> source2) {
return Observable.create(new OnSubscribe<R>() {
@Override
public void call(Subscriber<? super R> t) {
Observable<T> source = source2.share();
Observable<T> offsetSource = source.skip(1);
source.zipWith(offsetSource, mergeFunc).subscribe(t);
}
});
}
That works
Feels a little funky
static <T, R> Transformer<T, R> lookback(Func2<T, T, R> mergeFunc) {
return source -> Observable.create(result -> {
Observable<T> left = source.share();
Observable<T> right = left.skip(1);
left.zipWith(right, mergeFunc).subscribe(result);
});
}
if you're curious, what I'm using this for is to take a stream of video positions, [0, 60, 120, 150...] and turn it into a stream of video intervals, [ (0,60), (60, 120), (120, 150) ... ], and then log the start and end time of each interval
well, technically, the stream is completed on pause, and a new stream is created when playback begins again
I really didn't want to do an awful stateful thing where I was caching the previous timestamp in a mapping function or something
but then you need to make a tuple each time that keeps the last two values
The mergfunction is much more elegant
but it felt weird
stuffing a lot of state into an accumulator function is a code smell to me
indeed
i want a scan that emit two values, one for the next iteration and one to send downstream
you can wrap it in a tuple, but then you are constantly creating new tuples, as well as keeping the previous element until the next element arrives.