Fix Zip race condition

ItemObserver onNext might not acquire the write lock due to an
onCompleted being handled by another thread. When handling
onCompleted, the ItemObserver does not check for any values that are
ready to be emitted, which might cause OperationZip to never emit
OnNext or OnCompleted.

Author: vadims

rxjava-core/src/main/java/rx/operators/OperationZip.java

@@ -103,8 +103,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> OnSubscribeFunc<R> zip(Obs
     }
 
     public static <R> OnSubscribeFunc<R> zip(Iterable<? extends Observable<?>> ws, final FuncN<? extends R> zipFunction) {
-        ManyObservables<?, R> a = new ManyObservables<Object, R>(ws, zipFunction);
-        return a;
+        return new ManyObservables<Object, R>(ws, zipFunction);
     }
 
     /**
@@ -246,7 +245,6 @@ public ItemObserver(
                 this.cancel = cancel;
             }
 
-            @SuppressWarnings("unchecked")
             @Override
             public void onNext(T value) {
                 rwLock.readLock().lock();
@@ -258,43 +256,7 @@ public void onNext(T value) {
                 } finally {
                     rwLock.readLock().unlock();
                 }
-                // 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();
-                                        cu = true;
-                                        return;
-                                    }
-                                    continue;
-                                }
-                                Object v = io.queue.peek();
-                                if (v == NULL_SENTINEL) {
-                                    v = null;
-                                }
-                                values.add((T) v);
-                            }
-                            if (values.size() == all.size()) {
-                                for (ItemObserver<T> io : all) {
-                                    io.queue.poll();
-                                }
-                                observer.onNext(values);
-                            } else {
-                                break;
-                            }
-                        }
-                    } finally {
-                        rwLock.writeLock().unlock();
-                        if (cu) {
-                            cancel.unsubscribe();
-                        }
-                    }
-                }
+                runCollector();
             }
 
             @Override
@@ -321,23 +283,7 @@ public void onCompleted() {
                 } 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();
-                                cu = true;
-                                return;
-                            }
-                        }
-                    } finally {
-                        rwLock.writeLock().unlock();
-                        if (cu) {
-                            cancel.unsubscribe();
-                        }
-                    }
-                }
+                runCollector();
                 unsubscribe();
             }
 
@@ -351,6 +297,45 @@ public void unsubscribe() {
                 toSource.unsubscribe();
             }
 
+            @SuppressWarnings("unchecked")
+            private void runCollector() {
+                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();
+                                        cu = true;
+                                        return;
+                                    }
+                                    continue;
+                                }
+                                Object v = io.queue.peek();
+                                if (v == NULL_SENTINEL) {
+                                    v = null;
+                                }
+                                values.add((T) v);
+                            }
+                            if (values.size() == all.size()) {
+                                for (ItemObserver<T> io : all) {
+                                    io.queue.poll();
+                                }
+                                observer.onNext(values);
+                            } else {
+                                break;
+                            }
+                        }
+                    } finally {
+                        rwLock.writeLock().unlock();
+                        if (cu) {
+                            cancel.unsubscribe();
+                        }
+                    }
+                }
+            }
         }
     }