2.x: dedicated Single.zip implementation, no dispose on all-success

src/main/java/io/reactivex/Single.java

@@ -13,6 +13,7 @@
 
 package io.reactivex;
 
+import java.util.NoSuchElementException;
 import java.util.concurrent.*;
 
 import org.reactivestreams.Publisher;
@@ -1035,8 +1036,9 @@ public static <T> Single<T> wrap(SingleSource<T> source) {
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     public static <T, R> Single<R> zip(final Iterable<? extends SingleSource<? extends T>> sources, Function<? super Object[], ? extends R> zipper) {
+        ObjectHelper.requireNonNull(zipper, "zipper is null");
         ObjectHelper.requireNonNull(sources, "sources is null");
-        return toSingle(Flowable.zipIterable(SingleInternalHelper.iterableToFlowable(sources), zipper, false, 1));
+        return RxJavaPlugins.onAssembly(new SingleZipIterable<T, R>(sources, zipper));
     }
 
     /**
@@ -1475,17 +1477,13 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Single<R> zip(
      */
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
-    @SuppressWarnings({"rawtypes", "unchecked"})
     public static <T, R> Single<R> zipArray(Function<? super Object[], ? extends R> zipper, SingleSource<? extends T>... sources) {
+        ObjectHelper.requireNonNull(zipper, "zipper is null");
         ObjectHelper.requireNonNull(sources, "sources is null");
-        Publisher[] sourcePublishers = new Publisher[sources.length];
-        int i = 0;
-        for (SingleSource<? extends T> s : sources) {
-            ObjectHelper.requireNonNull(s, "The " + i + "th source is null");
-            sourcePublishers[i] = RxJavaPlugins.onAssembly(new SingleToFlowable<T>(s));
-            i++;
+        if (sources.length == 0) {
+            return error(new NoSuchElementException());
         }
-        return toSingle(Flowable.zipArray(zipper, false, 1, sourcePublishers));
+        return RxJavaPlugins.onAssembly(new SingleZipArray<T, R>(sources, zipper));
     }
 
     /**

src/main/java/io/reactivex/internal/operators/maybe/MaybeZipArray.java

@@ -59,7 +59,13 @@ public R apply(T t) throws Exception {
                 return;
             }
 
-            sources[i].subscribe(parent.observers[i]);
+            MaybeSource<? extends T> source = sources[i];
+
+            if (source == null) {
+                parent.innerError(new NullPointerException("One of the sources is null"), i);
+                return;
+            }
+            source.subscribe(parent.observers[i]);
         }
     }
 

src/main/java/io/reactivex/internal/operators/maybe/MaybeZipIterable.java

@@ -40,6 +40,10 @@ protected void subscribeActual(MaybeObserver<? super R> observer) {
 
         try {
             for (MaybeSource<? extends T> source : sources) {
+                if (source == null) {
+                    EmptyDisposable.error(new NullPointerException("One of the sources is null"), observer);
+                    return;
+                }
                 if (n == a.length) {
                     a = Arrays.copyOf(a, n + (n >> 2));
                 }

src/main/java/io/reactivex/internal/operators/single/SingleMap.java

@@ -30,30 +30,42 @@ public SingleMap(SingleSource<? extends T> source, Function<? super T, ? extends
 
     @Override
     protected void subscribeActual(final SingleObserver<? super R> t) {
-        source.subscribe(new SingleObserver<T>() {
-            @Override
-            public void onSubscribe(Disposable d) {
-                t.onSubscribe(d);
-            }
+        source.subscribe(new MapSingleObserver<T, R>(t, mapper));
+    }
 
-            @Override
-            public void onSuccess(T value) {
-                R v;
-                try {
-                    v = mapper.apply(value);
-                } catch (Throwable e) {
-                    Exceptions.throwIfFatal(e);
-                    onError(e);
-                    return;
-                }
-
-                t.onSuccess(v);
-            }
+    static final class MapSingleObserver<T, R> implements SingleObserver<T> {
+
+        final SingleObserver<? super R> t;
+
+        final Function<? super T, ? extends R> mapper;
 
-            @Override
-            public void onError(Throwable e) {
-                t.onError(e);
+        MapSingleObserver(SingleObserver<? super R> t, Function<? super T, ? extends R> mapper) {
+            this.t = t;
+            this.mapper = mapper;
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            t.onSubscribe(d);
+        }
+
+        @Override
+        public void onSuccess(T value) {
+            R v;
+            try {
+                v = mapper.apply(value);
+            } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
+                onError(e);
+                return;
             }
-        });
+
+            t.onSuccess(v);
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            t.onError(e);
+        }
     }
 }

src/main/java/io/reactivex/internal/operators/single/SingleZipArray.java

@@ -0,0 +1,185 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.single;
+
+import java.util.concurrent.atomic.*;
+
+import io.reactivex.*;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.Function;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.functions.ObjectHelper;
+import io.reactivex.plugins.RxJavaPlugins;
+
+public final class SingleZipArray<T, R> extends Single<R> {
+
+    final SingleSource<? extends T>[] sources;
+
+    final Function<? super Object[], ? extends R> zipper;
+
+    public SingleZipArray(SingleSource<? extends T>[] sources, Function<? super Object[], ? extends R> zipper) {
+        this.sources = sources;
+        this.zipper = zipper;
+    }
+
+    @Override
+    protected void subscribeActual(SingleObserver<? super R> observer) {
+        SingleSource<? extends T>[] sources = this.sources;
+        int n = sources.length;
+
+
+        if (n == 1) {
+            sources[0].subscribe(new SingleMap.MapSingleObserver<T, R>(observer, new Function<T, R>() {
+                @Override
+                public R apply(T t) throws Exception {
+                    return zipper.apply(new Object[] { t });
+                }
+            }));
+            return;
+        }
+
+        ZipCoordinator<T, R> parent = new ZipCoordinator<T, R>(observer, n, zipper);
+
+        observer.onSubscribe(parent);
+
+        for (int i = 0; i < n; i++) {
+            if (parent.isDisposed()) {
+                return;
+            }
+
+            SingleSource<? extends T> source = sources[i];
+
+            if (source == null) {
+                parent.innerError(new NullPointerException("One of the sources is null"), i);
+                return;
+            }
+
+            source.subscribe(parent.observers[i]);
+        }
+    }
+
+    static final class ZipCoordinator<T, R> extends AtomicInteger implements Disposable {
+
+
+        private static final long serialVersionUID = -5556924161382950569L;
+
+        final SingleObserver<? super R> actual;
+
+        final Function<? super Object[], ? extends R> zipper;
+
+        final ZipSingleObserver<T>[] observers;
+
+        final Object[] values;
+
+        @SuppressWarnings("unchecked")
+        ZipCoordinator(SingleObserver<? super R> observer, int n, Function<? super Object[], ? extends R> zipper) {
+            super(n);
+            this.actual = observer;
+            this.zipper = zipper;
+            ZipSingleObserver<T>[] o = new ZipSingleObserver[n];
+            for (int i = 0; i < n; i++) {
+                o[i] = new ZipSingleObserver<T>(this, i);
+            }
+            this.observers = o;
+            this.values = new Object[n];
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return get() <= 0;
+        }
+
+        @Override
+        public void dispose() {
+            if (getAndSet(0) > 0) {
+                for (ZipSingleObserver<?> d : observers) {
+                    d.dispose();
+                }
+            }
+        }
+
+        void innerSuccess(T value, int index) {
+            values[index] = value;
+            if (decrementAndGet() == 0) {
+                R v;
+
+                try {
+                    v = ObjectHelper.requireNonNull(zipper.apply(values), "The zipper returned a null value");
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    actual.onError(ex);
+                    return;
+                }
+
+                actual.onSuccess(v);
+            }
+        }
+
+        void disposeExcept(int index) {
+            ZipSingleObserver<T>[] observers = this.observers;
+            int n = observers.length;
+            for (int i = 0; i < index; i++) {
+                observers[i].dispose();
+            }
+            for (int i = index + 1; i < n; i++) {
+                observers[i].dispose();
+            }
+        }
+
+        void innerError(Throwable ex, int index) {
+            if (getAndSet(0) > 0) {
+                disposeExcept(index);
+                actual.onError(ex);
+            } else {
+                RxJavaPlugins.onError(ex);
+            }
+        }
+    }
+
+    static final class ZipSingleObserver<T>
+    extends AtomicReference<Disposable>
+    implements SingleObserver<T> {
+
+        private static final long serialVersionUID = 3323743579927613702L;
+
+        final ZipCoordinator<T, ?> parent;
+
+        final int index;
+
+        ZipSingleObserver(ZipCoordinator<T, ?> parent, int index) {
+            this.parent = parent;
+            this.index = index;
+        }
+
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            DisposableHelper.setOnce(this, d);
+        }
+
+        @Override
+        public void onSuccess(T value) {
+            parent.innerSuccess(value, index);
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            parent.innerError(e, index);
+        }
+    }
+}