Add Single.zip() for Iterable of Singles

[artem-zinnatullin]

No description provided.

[akarnokd]

Although not required, I prefer iterating such sources when the child subscribes and not when the sequence is assembled.

[artem-zinnatullin]

Hm, I am afraid that iterating over sequence when child subscribes may lead to data inconsistency if user will pass the collection of Singles and then change it before subscribe().

[akarnokd]

Did you consider adding zip(Single[], FuncN) overload as well?

Otherwise, looks good. 👍

[artem-zinnatullin]

I'll add zin(Single[], FuncN) and zipWith() in separate PRs if you don't mind.

Btw, idk if @benjchristensen remembers how I was arguing against adding Single to the RxJava. My apologies 😅. I'm starting loving Single in our codebase!

[akarnokd]

Here is an implementation that avoids conversion to Observable:

@SuppressWarnings("unchecked")
public static <T, R> Single<R> zip(
        Iterable<? extends Single<? extends T>> singles, 
                FuncN<? extends R> zipper) {
    return Single.create(f -> {
        Single<? extends T>[] sa;
        int count;
        if (singles instanceof List) {
            List<? extends Single<? extends T>> list = 
                    (List<? extends Single<? extends T>>)singles;
            count = list.size();
            sa = list.toArray(new Single[count]);
        } else {
            sa = new Single[8];
            count = 0;
            for (Single<? extends T> s : singles) {
                if (count == sa.length) {
                    Single<? extends T>[] sb = new Single[count + (count >> 2)];
                    System.arraycopy(sa, 0, sb, 0, count);
                    sa = sb;
                }
                sa[count] = s;
                count++;
            }
        }

        final AtomicInteger wip = new AtomicInteger(count);
        final AtomicBoolean once = new AtomicBoolean();
        final Object[] values = new Object[count];

        CompositeSubscription csub = new CompositeSubscription();
        f.add(csub);

        for (int i = 0; i < count; i++) {
            if (csub.isUnsubscribed() || once.get()) {
                break;
            }

            final int j = i;
            SingleSubscriber<T> te = new SingleSubscriber<T>() {
                @Override
                public void onSuccess(T value) {
                    values[j] = value;
                    if (wip.decrementAndGet() == 0) {
                        R r;

                        try {
                            r = zipper.call(values);
                        } catch (Throwable e) {
                            Exceptions.throwIfFatal(e);
                            f.onError(e);
                            return;
                        }

                        f.onSuccess(r);
                    }
                }

                @Override
                public void onError(Throwable error) {
                    if (once.compareAndSet(false, true)) {
                        f.onError(error);
                    } else {
                        RxJavaPlugins.getInstance()
                        .getErrorHandler().handleError(error);
                    }
                }
            };
            csub.add(te);
            if (csub.isUnsubscribed() || once.get()) {
                break;
            }
            sa[i].subscribe(te);
        }
    });
}

[artem-zinnatullin]

@akarnokd I've used your code (rewrote some parts and fixed one bug) and now we don't have to convert Singles to Observables!

Later we can switch other Single.zip overloads to SingleOperatorZip.

[akarnokd]

This should be just this.onError to avoid error races.

[artem-zinnatullin]

fixed

[artem-zinnatullin]

@akarnokd changed instanceof List to instanceof Collection so we are now really efficient for all collections and fall back to slightly less fast code for other Iterables, rebased.

PTAL

[akarnokd]

Iterables and Observables are mainly for deferred data streaming. This will iterate at assembly time instead on Subscription time, losing the duality aspects. For example, an Observable source turned into an Iterable via toBlocking().getIterable() is no longer lazy when this zip is applied.

[karlicoss]

I think it is compatible with the behavior of Observable.zip.

[stealthcode]

Yep currently Observable serializes all values in the iterable into an array before zipping them so you can't append after lifting. IMO this is probably intentional since appending to some iterables while iterating results in ConcurrentModificationExceptions. Today the Observable<Observable<T>> overload can receive new observables to zip but the zip function won't be called until the outer observable calls onCompleted(). This is slightly later than the Iterable overload.

[artem-zinnatullin]

So, @akarnokd, @stealthcode let's use current implementation from this PR to be consistent with Observable.zip(Iterable<Observable>)?

[stealthcode]

Sounds good to me.

[akarnokd]

Okay.

[artem-zinnatullin]

Rebased!

[artem-zinnatullin]

Blocked by #3569.

[akarnokd]

This needs to be rebased again.

[akarnokd]

👍

[stealthcode]

👍

[stevegury]

👍 (for posterity)