RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
public class LookbackTransformer<T, R> implements Observable.Transformer<T, R> {
private final Func2<T, T, R> mergeFunc;
public LookbackTransformer(Func2<T, T, R> mergeFunc) {
this.mergeFunc = mergeFunc;
}
@Override
public Observable<R> call(Observable<T> source) {
return source.window(2, 1).flatMap(pair -> {
Observable<T> left = pair.elementAt(1);
Observable<T> right = pair.elementAt(2);
return left.zipWith(right, mergeFunc);
}
};
}
that way, if there were only one element in
source, right would never resolve and the final merge would never occur
dammit, that wouldn't work either,
elementAt throws an IndexOutOfBoundsException :D
public class LookbackTransformer<T, R> implements Observable.Transformer<T, R> {
private final Func2<T, T, R> mergeFunc;
public LookbackTransformer(Func2<T, T, R> mergeFunc) {
this.mergeFunc = mergeFunc;
}
@Override
public Observable<R> call(Observable<T> source) {
return source.window(2, 1).flatMap(pair -> {
try {
Observable<T> left = pair.elementAt(1);
Observable<T> right = pair.elementAt(2);
return left.zipWith(right, mergeFunc);
} catch (IndexOutOfBoundsException e) {
return Observable.never();
}
}
};
not very elegant
and I'm not sure it would work
Not even sure if you really need window here. Might as well go for
left = publish source; right = left.skip(1); result = left.zipWith(right, mergFunc); left.connect(); return result;(in speudo code). Should probably use observable.create to be able to track the disposable from connect() there.
Mm, i got it this far, but now i cant figure out how to connect the subscription to the observable.create like you would do in C#.
@Override
public Observable<R> call(Observable<T> source) {
return Observable.create(new Observable.OnSubscribe<R>() {
@Override
public void call(Subscriber<? super R> observer) {
ConnectableObservable<T> left = source.publish();
Observable<T> right = left.skip(1);
Subscription sub = new CompositeSubscription(
left.zipWith(right, mergeFunc).subscribe(observer),
left.connect());
// return sub??
}
}) ;
}
which is an alias for
publish().refcount()
iirc
yes share sounds perfect
are we sure it is impossible for elements to emit during subscription?
i mean, is it possibel for the source to emit a element between
source.share() and left.skip(1)
because that would be a problem
public class LookbackTransformer<T, R> implements Observable.Transformer<T, R> {
private final Func2<T, T, R> mergeFunc;
public LookbackTransformer(Func2<T, T, R> mergeFunc) {
this.mergeFunc = mergeFunc;
}
@Override
public Observable<R> call(Observable<T> source) {
Observable<T> offsetSource = source.skip(1);
return source.zipWith(offsetSource, mergeFunc);
}
}
I just tested it and all mapped functions prior to the
compose(lookbackTransformer) are getting called once
I'm looking at the internals of the Zip operator
wow, it's pretty hairy in there, I think I'd need to step through with the debugger to figure out what's going on :)
today is not that day though
The output looks like this:
do 0
do 0
do 1
do 1
sb 1
do 2
sb 2
do 0
do 0
do 1
do 1
sb 1
do 2
sb 2
sb 0 is missing and the do is seen twice per element.
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);
}