2.x: CombineLatest removed leftover debug field + better cleanup

akarnokd · akarnokd · commit ca20a2bf090a · 2015-09-27T21:02:15.000+02:00
diff --git a/src/main/java/io/reactivex/internal/operators/PublisherCombineLatest.java b/src/main/java/io/reactivex/internal/operators/PublisherCombineLatest.java
@@ -152,13 +152,18 @@ public void cancel() {
         }
         
         void cancel(Queue<?> q) {
-            q.clear();
+            clear(q);
             for (CombinerSubscriber<T, R> s : subscribers) {
                 s.cancel();
             }
         }
-        
-        Object last = "1";
+
+        void clear(Queue<?> q) {
+            synchronized (this) {
+                Arrays.fill(latest, null);
+            }
+            q.clear();
+        }
         
         void combine(T value, int index) {
             CombinerSubscriber<T, R> cs = subscribers[index];
@@ -169,6 +174,9 @@ void combine(T value, int index) {
             boolean empty;
             boolean f;
             synchronized (this) {
+                if (cancelled) {
+                    return;
+                }
                 len = latest.length;
                 Object o = latest[index];
                 a = active;
@@ -287,6 +295,7 @@ boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a, Queue<?> q, b
             if (d) {
                 if (delayError) {
                     if (empty) {
+                        clear(queue);
                         Throwable e = error;
                         if (e != null) {
                             a.onError(e);
@@ -303,6 +312,7 @@ boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a, Queue<?> q, b
                         return true;
                     } else
                     if (empty) {
+                        clear(queue);
                         a.onComplete();
                         return true;
                     }
diff --git a/src/main/java/io/reactivex/internal/operators/nbp/NbpOnSubscribeCombineLatest.java b/src/main/java/io/reactivex/internal/operators/nbp/NbpOnSubscribeCombineLatest.java
@@ -13,7 +13,7 @@
 
 package io.reactivex.internal.operators.nbp;
 
-import java.util.Queue;
+import java.util.*;
 import java.util.concurrent.atomic.*;
 import java.util.function.Function;
 
@@ -139,13 +139,18 @@ public void dispose() {
         }
         
         void cancel(Queue<?> q) {
-            q.clear();
+            clear(q);
             for (CombinerSubscriber<T, R> s : subscribers) {
                 s.dispose();
             }
         }
         
-        Object last = "1";
+        void clear(Queue<?> q) {
+            synchronized (this) {
+                Arrays.fill(latest, null);
+            }
+            q.clear();
+        }
         
         void combine(T value, int index) {
             CombinerSubscriber<T, R> cs = subscribers[index];
@@ -156,6 +161,9 @@ void combine(T value, int index) {
             boolean empty;
             boolean f;
             synchronized (this) {
+                if (cancelled) {
+                    return;
+                }
                 len = latest.length;
                 Object o = latest[index];
                 a = active;
@@ -258,6 +266,7 @@ boolean checkTerminated(boolean d, boolean empty, NbpSubscriber<?> a, Queue<?> q
             if (d) {
                 if (delayError) {
                     if (empty) {
+                        clear(queue);
                         Throwable e = error;
                         if (e != null) {
                             a.onError(e);
@@ -274,6 +283,7 @@ boolean checkTerminated(boolean d, boolean empty, NbpSubscriber<?> a, Queue<?> q
                         return true;
                     } else
                     if (empty) {
+                        clear(queue);
                         a.onComplete();
                         return true;
                     }
