Merge pull request #778 from vadims/fix-zip-race

Fix zip race condition

Author: benjchristensen

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);
     }
 
     /**
@@ -203,11 +202,9 @@ private static final class ItemObserver<T> implements Observer<T>, Subscription
             /** Reader-writer lock. */
             protected final ReadWriteLock rwLock;
             /** The queue. */
-            public final Queue<Object> queue = new LinkedList<Object>();
+            public final Queue<T> queue = new LinkedList<T>();
             /** The list of the other observers. */
             public final List<ItemObserver<T>> all;
-            /** The null sentinel value. */
-            protected static final Object NULL_SENTINEL = new Object();
             /** The global cancel. */
             protected final Subscription cancel;
             /** The subscription to the source. */
@@ -246,55 +243,18 @@ public ItemObserver(
                 this.cancel = cancel;
             }
 
-            @SuppressWarnings("unchecked")
             @Override
             public void onNext(T value) {
                 rwLock.readLock().lock();
                 try {
                     if (done) {
                         return;
                     }
-                    queue.add(value != null ? value : NULL_SENTINEL);
+                    queue.add(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 +281,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 +295,41 @@ public void unsubscribe() {
                 toSource.unsubscribe();
             }
 
+            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;
+                                    }
+                                } else {
+                                    T value = io.queue.peek();
+                                    values.add(value);
+                                }
+                            }
+                            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();
+                        }
+                    }
+                }
+            }
         }
     }