3.x: Allow Single.zip and Maybe.zip result to be garbage collected (#7196)

* #7195 Allow zip result to be garbage collected

* Also `null` collected values on error and complete

* Update MaybeZipArray.java

* Update SingleZipArray.java

* Update MaybeZipArray.java

* Update SingleZipArray.java

* Update MaybeZipArray.java

* Update MaybeZipArray.java

Co-authored-by: Alex <[email protected]>
Co-authored-by: David Karnok <[email protected]>

Author: 3 people

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

@@ -13,6 +13,7 @@
 
 package io.reactivex.rxjava3.internal.operators.maybe;
 
+import java.util.Arrays;
 import java.util.Objects;
 import java.util.concurrent.atomic.*;
 
@@ -73,7 +74,7 @@ static final class ZipCoordinator<T, R> extends AtomicInteger implements Disposa
 
         final ZipMaybeObserver<T>[] observers;
 
-        final Object[] values;
+        Object[] values;
 
         @SuppressWarnings("unchecked")
         ZipCoordinator(MaybeObserver<? super R> observer, int n, Function<? super Object[], ? extends R> zipper) {
@@ -99,22 +100,29 @@ public void dispose() {
                 for (ZipMaybeObserver<?> d : observers) {
                     d.dispose();
                 }
+
+                values = null;
             }
         }
 
         void innerSuccess(T value, int index) {
-            values[index] = value;
+            Object[] values = this.values;
+            if (values != null) {
+                values[index] = value;
+            }
             if (decrementAndGet() == 0) {
                 R v;
 
                 try {
                     v = Objects.requireNonNull(zipper.apply(values), "The zipper returned a null value");
                 } catch (Throwable ex) {
                     Exceptions.throwIfFatal(ex);
+                    this.values = null;
                     downstream.onError(ex);
                     return;
                 }
 
+                this.values = null;
                 downstream.onSuccess(v);
             }
         }
@@ -133,6 +141,7 @@ void disposeExcept(int index) {
         void innerError(Throwable ex, int index) {
             if (getAndSet(0) > 0) {
                 disposeExcept(index);
+                values = null;
                 downstream.onError(ex);
             } else {
                 RxJavaPlugins.onError(ex);
@@ -142,6 +151,7 @@ void innerError(Throwable ex, int index) {
         void innerComplete(int index) {
             if (getAndSet(0) > 0) {
                 disposeExcept(index);
+                values = null;
                 downstream.onComplete();
             }
         }

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

@@ -13,6 +13,7 @@
 
 package io.reactivex.rxjava3.internal.operators.single;
 
+import java.util.Arrays;
 import java.util.Objects;
 import java.util.concurrent.atomic.*;
 
@@ -74,7 +75,7 @@ static final class ZipCoordinator<T, R> extends AtomicInteger implements Disposa
 
         final ZipSingleObserver<T>[] observers;
 
-        final Object[] values;
+        Object[] values;
 
         @SuppressWarnings("unchecked")
         ZipCoordinator(SingleObserver<? super R> observer, int n, Function<? super Object[], ? extends R> zipper) {
@@ -100,22 +101,29 @@ public void dispose() {
                 for (ZipSingleObserver<?> d : observers) {
                     d.dispose();
                 }
+
+                values = null;
             }
         }
 
         void innerSuccess(T value, int index) {
-            values[index] = value;
+            Object[] values = this.values;
+            if (values != null) {
+                values[index] = value;
+            }
             if (decrementAndGet() == 0) {
                 R v;
 
                 try {
                     v = Objects.requireNonNull(zipper.apply(values), "The zipper returned a null value");
                 } catch (Throwable ex) {
                     Exceptions.throwIfFatal(ex);
+                    this.values = null;
                     downstream.onError(ex);
                     return;
                 }
 
+                this.values = null;
                 downstream.onSuccess(v);
             }
         }
@@ -134,6 +142,7 @@ void disposeExcept(int index) {
         void innerError(Throwable ex, int index) {
             if (getAndSet(0) > 0) {
                 disposeExcept(index);
+                values = null;
                 downstream.onError(ex);
             } else {
                 RxJavaPlugins.onError(ex);