Merge pull request #767 from akarnokd/ZipFixes

Zip fix for multiple onCompleted and moved unsubscribe outside the lock.
ReactiveX · Jan 21, 2014 · f17e934 · f17e934
2 parents db0440a + 51b8acf
commit f17e934
Show file tree

Hide file tree

Showing 2 changed files with 71 additions and 17 deletions.
diff --git a/rxjava-core/src/main/java/rx/operators/OperationZip.java b/rxjava-core/src/main/java/rx/operators/OperationZip.java
@@ -152,19 +152,28 @@ public Subscription onSubscribe(final Observer<? super U> observer) {
             final List<ItemObserver<T>> all = new ArrayList<ItemObserver<T>>();
 
             Observer<List<T>> o2 = new Observer<List<T>>() {
+                boolean done;
                 @Override
                 public void onCompleted() {
-                    observer.onCompleted();
+                    if (!done) {
+                        done = true;
+                        observer.onCompleted();
+                    }
                 }
 
                 @Override
                 public void onError(Throwable t) {
-                    observer.onError(t);
+                    if (!done) {
+                        done = true;
+                        observer.onError(t);
+                    }
                 }
 
                 @Override
                 public void onNext(List<T> value) {
-                    observer.onNext(selector.call(value.toArray(new Object[value.size()])));
+                    if (!done) {
+                        observer.onNext(selector.call(value.toArray(new Object[value.size()])));
+                    }
                 }
             };
 
@@ -251,14 +260,15 @@ public void onNext(T value) {
                 }
                 // run collector
                 if (rwLock.writeLock().tryLock()) {
+                    boolean cu = false;
                     try {
                         while (true) {
                             List<T> values = new ArrayList<T>(all.size());
                             for (ItemObserver<T> io : all) {
                                 if (io.queue.isEmpty()) {
                                     if (io.done) {
                                         observer.onCompleted();
-                                        cancel.unsubscribe();
+                                        cu = true;
                                         return;
                                     }
                                     continue;
@@ -280,61 +290,60 @@ public void onNext(T value) {
                         }
                     } finally {
                         rwLock.writeLock().unlock();
+                        if (cu) {
+                            cancel.unsubscribe();
+                        }
                     }
                 }
             }
 
             @Override
             public void onError(Throwable ex) {
-                boolean c = false;
                 rwLock.writeLock().lock();
                 try {
                     if (done) {
                         return;
                     }
                     done = true;
-                    c = true;
                     observer.onError(ex);
-                    cancel.unsubscribe();
                 } finally {
                     rwLock.writeLock().unlock();
                 }
-                if (c) {
-                    unsubscribe();
-                }
+                cancel.unsubscribe();
+                unsubscribe();
             }
 
             @Override
             public void onCompleted() {
-                boolean c = false;
                 rwLock.readLock().lock();
                 try {
                     done = true;
-                    c = true;
                 } finally {
                     rwLock.readLock().unlock();
                 }
                 if (rwLock.writeLock().tryLock()) {
+                    boolean cu = false;
                     try {
                         for (ItemObserver<T> io : all) {
                             if (io.queue.isEmpty() && io.done) {
                                 observer.onCompleted();
-                                cancel.unsubscribe();
+                                cu = true;
                                 return;
                             }
                         }
                     } finally {
                         rwLock.writeLock().unlock();
+                        if (cu) {
+                            cancel.unsubscribe();
+                        }
                     }
                 }
-                if (c) {
-                    unsubscribe();
-                }
+                unsubscribe();
             }
 
             /** Connect to the source observable. */
             public void connect() {
-                toSource.setSubscription(source.subscribe(this));
+                toSource.set(source.subscribe(this));
             }
 
             @Override

diff --git a/rxjava-core/src/test/java/rx/operators/OperationZipTest.java b/rxjava-core/src/test/java/rx/operators/OperationZipTest.java
@@ -1018,4 +1018,49 @@ public void remove() {
         verify(o, never()).onCompleted();
 
     }
+
+    @Test
+    public void testZipWithOnCompletedTwice() {
+        // issue: https://groups.google.com/forum/#!topic/rxjava/79cWTv3TFp0
+        // The problem is the original "zip" implementation does not wrap
+        // an internal observer with a SafeObserver. However, in the "zip",
+        // it may calls "onCompleted" twice. That breaks the Rx contract.
+
+        // This test tries to emulate this case.
+        // As "mock(Observer.class)" will create an instance in the package "rx",
+        // we need to wrap "mock(Observer.class)" with an observer instance
+        // which is in the package "rx.operators".
+        @SuppressWarnings("unchecked")
+        final Observer<Integer> observer = mock(Observer.class);
+
+        Observable.zip(Observable.from(1),
+                Observable.from(1), new Func2<Integer, Integer, Integer>() {
+                    @Override
+                    public Integer call(Integer a, Integer b) {
+                        return a + b;
+                    }
+                }).subscribe(new Observer<Integer>() {
+
+            @Override
+            public void onCompleted() {
+                observer.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                observer.onError(e);
+            }
+
+            @Override
+            public void onNext(Integer args) {
+                observer.onNext(args);
+            }
+
+        });
+
+        InOrder inOrder = inOrder(observer);
+        inOrder.verify(observer, times(1)).onNext(2);
+        inOrder.verify(observer, times(1)).onCompleted();
+        inOrder.verifyNoMoreInteractions();
+    }
 }