Observable { @NonNull public static <T, R> Observable<R> zip(@NonNull final Observable<T> lh, @NonNull final Observable<T> rh, @NonNull final BiFunction<T, T, R> func) { return create(new Function<Observer<R>, Subscription>() { @NonNull @Override public Subscription apply(@NonNull final Observer<R> observer) { final CompoundSubscription compoundSubscription = new CompoundSubscription(); final ArrayList<T> lhValues = new ArrayList<>(); final ArrayList<T> rhValues = new ArrayList<>(); final Holder<Boolean> lhCompleted = new Holder<>(false); final Holder<Boolean> rhCompleted = new Holder<>(false); final Holder<Boolean> completed = new Holder<>(false); final Runnable completeIfNeeded = new Runnable() { @Override public void run() { synchronized (observer) { if (completed.getValue()) { return; } if ((lhCompleted.getValue() && lhValues.size() == 0) && rhCompleted.getValue() && rhValues.size() == 0) { completed.setValue(true); compoundSubscription.cancel(); observer.onCompleted(); } } } }; final Runnable emitNextIfNeeded = new Runnable() { @Override public void run() { synchronized (observer) { if (lhValues.size() > 0 && rhValues.size() > 0) { R zipped = func.apply(lhValues.get(0), rhValues.get(0)); lhValues.remove(0); rhValues.remove(0); observer.onNext(zipped); completeIfNeeded.run(); } } } }; compoundSubscription.add(lh.subscribe(new Subscriber<T>() { @Override public void onNext(@NonNull T value) { synchronized (observer) { lhValues.add(value); emitNextIfNeeded.run(); } } @Override public void onCompleted() { synchronized (observer) { lhCompleted.setValue(true); completeIfNeeded.run(); } } @Override public void onError(@NonNull Exception e) { synchronized (observer) { compoundSubscription.cancel(); observer.onError(e); } } })); compoundSubscription.add(rh.subscribe(new Subscriber<T>() { @Override public void onNext(@NonNull T value) { synchronized (observer) { rhValues.add(value); emitNextIfNeeded.run(); } } @Override public void onCompleted() { synchronized (observer) { rhCompleted.setValue(true); completeIfNeeded.run(); } } @Override public void onError(@NonNull Exception e) { synchronized (observer) { compoundSubscription.cancel(); observer.onError(e); } } })); return compoundSubscription; } }); } protected  Observable(); protected  Observable(@Nullable Function<Observer<T>, Subscription> onSubscribe); @NonNull static Observable<T> create(@NonNull Function<Observer<T>, Subscription> func); @NonNull static Observable<T> just(@NonNull final T value); @NonNull static Observable<T> empty(); @NonNull static Observable<T> never(); @NonNull static Observable<T> error(@NonNull final Exception e); @NonNull static Observable<T> from(@NonNull final Collection<T> collection); @NonNull Subscription subscribe(@NonNull Observer<T> observer); @NonNull Observable<R> flatMap(@NonNull final Function<T, Observable<R>> func); @NonNull Observable<R> map(@NonNull final Function<T, R> func); @NonNull Observable<T> filter(@NonNull final Predicate<T> pred); @NonNull Observable<T> distinctUntilChanged(); @NonNull Observable<T> defaultIfEmpty(@NonNull final T defaultValue); @NonNull Observable<T> observeOn(@NonNull final Scheduler scheduler); @NonNull Observable<T> subscribeOn(@NonNull final Scheduler scheduler); @NonNull static Observable<T> merge(@NonNull final Observable<T> lh, @NonNull final Observable<T> rh); @NonNull static Observable<T> merge(@NonNull Collection<Observable<T>> observables); @NonNull static Observable<T> concat(@NonNull final Observable<T> lh, @NonNull final Observable<T> rh); @NonNull static Observable<T> defer(@NonNull final Supplier<Observable<T>> func); @NonNull static Observable<R> zip(@NonNull final Observable<T> lh, @NonNull final Observable<T> rh, @NonNull final BiFunction<T, T, R> func);  }
@Test public void testZip() throws Exception { Subject<Integer> first = Subject.create(); Subject<Integer> second = Subject.create(); Observable<Pair<Integer, Integer>> zipped = Observable.zip(first, second, new BiFunction<Integer, Integer, Pair<Integer, Integer>>() { @NonNull @Override public Pair<Integer, Integer> apply(@NonNull Integer lh, @NonNull Integer rh) { return new Pair<>(lh, rh); } }); subscribeObservable(zipped); first.onNext(1); second.onNext(4); first.onNext(2); second.onNext(5); first.onNext(3); second.onNext(6); first.onCompleted(); second.onCompleted(); List<Pair<Integer, Integer>> expected = Arrays.asList(new Pair<>(1, 4), new Pair<>(2, 5), new Pair<>(3, 6)); performAsserts(null, null, expected, 3, 1, 0); }