Merge pull request #832 from duncani/fixes

[Issue #831] Fix for OperationJoin race condition
ReactiveX · Feb 7, 2014 · a1df031 · a1df031
2 parents 9520d63 + a225944
commit a1df031
Showing 1 changed file with 29 additions and 20 deletions.
diff --git a/rxjava-core/src/main/java/rx/operators/OperationJoin.java b/rxjava-core/src/main/java/rx/operators/OperationJoin.java
@@ -109,10 +109,12 @@ protected void expire(int id, Subscription resource) {
 
             @Override
             public void onNext(TLeft args) {
-                int id;
+                int id, highRightId;
+
                 synchronized (gate) {
                     id = leftId++;
                     leftMap.put(id, args);
+                    highRightId = rightId;
                 }
                 SerialSubscription md = new SerialSubscription();
                 group.add(md);
@@ -129,16 +131,19 @@ public void onNext(TLeft args) {
                 md.setSubscription(duration.subscribe(new LeftDurationObserver(id, md)));
 
                 synchronized (gate) {
-                    for (TRight r : rightMap.values()) {
-                        R result;
-                        try {
-                            result = resultSelector.call(args, r);
-                        } catch (Throwable t) {
-                            observer.onError(t);
-                            cancel.unsubscribe();
-                            return;
+                    for (Map.Entry<Integer, TRight> entry : rightMap.entrySet()) {
+                        if (entry.getKey() < highRightId) {
+                            TRight r = entry.getValue();
+                            R result;
+                            try {
+                                result = resultSelector.call(args, r);
+                            } catch (Throwable t) {
+                                observer.onError(t);
+                                cancel.unsubscribe();
+                                return;
+                            }
+                            observer.onNext(result);
                         }
-                        observer.onNext(result);
                     }
                 }
             }
@@ -212,10 +217,11 @@ void expire(int id, Subscription resource) {
 
             @Override
             public void onNext(TRight args) {
-                int id = 0;
+                int id = 0, highLeftId;
                 synchronized (gate) {
                     id = rightId++;
                     rightMap.put(id, args);
+                    highLeftId = leftId;
                 }
                 SerialSubscription md = new SerialSubscription();
                 group.add(md);
@@ -232,16 +238,19 @@ public void onNext(TRight args) {
                 md.setSubscription(duration.subscribe(new RightDurationObserver(id, md)));
 
                 synchronized (gate) {
-                    for (TLeft lv : leftMap.values()) {
-                        R result;
-                        try {
-                            result = resultSelector.call(lv, args);
-                        } catch (Throwable t) {
-                            observer.onError(t);
-                            cancel.unsubscribe();
-                            return;
+                    for (Map.Entry<Integer, TLeft> entry : leftMap.entrySet()) {
+                        if (entry.getKey() < highLeftId) {
+                            TLeft lv = entry.getValue();
+                            R result;
+                            try {
+                                result = resultSelector.call(lv, args);
+                            } catch (Throwable t) {
+                                observer.onError(t);
+                                cancel.unsubscribe();
+                                return;
+                            }
+                            observer.onNext(result);
                         }
-                        observer.onNext(result);
                     }
                 }
             }