Fix PublishSubject from skipping newly added items

Ugly hack, this would better be fixed in the Map implementation instead.
ReactiveX · May 30, 2013 · a7fc861 · a7fc861
1 parent 52640f5
commit a7fc861
Showing 1 changed file with 35 additions and 9 deletions.
diff --git a/rxjava-core/src/main/java/rx/subjects/PublishSubject.java b/rxjava-core/src/main/java/rx/subjects/PublishSubject.java
@@ -20,6 +20,8 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
+import java.util.HashSet;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -94,23 +96,47 @@ protected PublishSubject(Func1<Observer<T>, Subscription> onSubscribe, Concurren
 
     @Override
     public void onCompleted() {
-        for (Observer<T> observer : observers.values()) {
-            observer.onCompleted();
-        }
+        int observersSize;
+        Set<Observer<T>> notifiedObservers = new HashSet<Observer<T>>();
+        do {
+            observersSize = observers.size();
+            for (Observer<T> observer : observers.values()) {
+                if (!notifiedObservers.contains(observer)) {
+                    observer.onCompleted();
+                    notifiedObservers.add(observer);
+                }
+            }
+        } while (observers.size() > observersSize);
     }
 
     @Override
     public void onError(Exception e) {
-        for (Observer<T> observer : observers.values()) {
-            observer.onError(e);
-        }
+        int observersSize;
+        Set<Observer<T>> notifiedObservers = new HashSet<Observer<T>>();
+        do {
+            observersSize = observers.size();
+            for (Observer<T> observer : observers.values()) {
+                if (!notifiedObservers.contains(observer)) {
+                    observer.onError(e);
+                    notifiedObservers.add(observer);
+                }
+            }
+        } while (observers.size() > observersSize);
     }
 
     @Override
     public void onNext(T args) {
-        for (Observer<T> observer : observers.values()) {
-            observer.onNext(args);
-        }
+        int observersSize;
+        Set<Observer<T>> notifiedObservers = new HashSet<Observer<T>>();
+        do {
+            observersSize = observers.size();
+            for (Observer<T> observer : observers.values()) {
+                if (!notifiedObservers.contains(observer)) {
+                    observer.onNext(args);
+                    notifiedObservers.add(observer);
+                }
+            }
+        } while (observers.size() > observersSize);
     }
 
     public static class UnitTest {