Playing with RxJava now and stumbled upon the following problem:
I have 2 different streams:
So essentially I have stream of items and I want all those items to be combined with that single item from 2nd stream:
----a1----a2----a3----a4----a5----|--------------->
-------------b1--|----------------------------------->
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
------------a1b1-a2b1-a3b1-a4b1-a5b1-------->
It looks really similar to combileLatest operator, but combineLatest will ignore all items from the first stream except the closest to the item from the second stream. It means that I will not receive a1b1 - the first resulting item emitted is gonna be a2b1.
I also looked at delay operator, but it doesn't allow me to specify close stream like it is done with buffer operatior
Is there any fancy operator which solves the problem above?
AFAIK, there is no a built-in operator to achieve the behavior you've described. You can always implement a custom operator or build it on top of existing operators. I think the second option is easier to implement and here is the code:
public static <L, R, T> Observable<T> zipper(final Observable<? extends L> left, final Observable<? extends R> right, final Func2<? super L, ? super R, ? extends T> function) {
return Observable.defer(new Func0<Observable<T>>() {
@Override
public Observable<T> call() {
final SerialSubscription subscription = new SerialSubscription();
final ConnectableObservable<? extends R> cached = right.replay();
return left.flatMap(new Func1<L, Observable<T>>() {
@Override
public Observable<T> call(final L valueLeft) {
return cached.map(new Func1<R, T>() {
@Override
public T call(final R valueRight) {
return function.call(valueLeft, valueRight);
}
});
}
}).doOnSubscribe(new Action0() {
@Override
public void call() {
subscription.set(cached.connect());
}
}).doOnUnsubscribe(new Action0() {
@Override
public void call() {
subscription.unsubscribe();
}
});
}
});
}
If you have any questions regarding the code, I can explain it in details.
UPDATE
Regarding the questing how my solution is different from the following one:
left.flatMap(valueLeft -> right.map(valueRight -> together(valueLeft, valueRight)));
left and right observables are executing in parallel. right observable doesn't have to wait for a left one to emit its first item.Caching - my solution subscribes only once to the right observables and caches its result. Thats why b1 will always be the same for all aXXX items. The solution provided by akarnokd subscribes to the rightobservable every time the left one emits an item. That means:
There is no guarantee that b1 won't change its value. For example for the following observable you will get a different b for each a.
final Observable<Double> right = Observable.defer(new Func0<Observable<Double>>() {
@Override
public Observable<Double> call() {
return Observable.just(Math.random());
}
});
If the right observable is a time consuming operation (e.g. network call), you will have to wait for its completion every time the left observable emits a new item.
Collected from the Internet
Please contact [email protected] to delete if infringement.
Comments