OperatorZipIterable

Author: benjchristensen

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

@@ -108,6 +108,7 @@
 import rx.operators.OperatorTimestamp;
 import rx.operators.OperatorToObservableList;
 import rx.operators.OperatorToObservableSortedList;
+import rx.operators.OperatorZipIterable;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
 import rx.schedulers.Schedulers;
@@ -8407,9 +8408,7 @@ 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(OperatorZip.zipIterable(this, other, zipFunction));
-        // TODO fix this
-        return null;
+        return lift(new OperatorZipIterable<T, T2, R>(other, zipFunction));
     }
 
     /**

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

@@ -0,0 +1,72 @@
+/**
+ * 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.Iterator;
+
+import rx.Subscriber;
+import rx.observers.Subscribers;
+import rx.util.functions.Func2;
+
+public final class OperatorZipIterable<T1, T2, R> implements Operator<R, T1> {
+
+    final Iterable<? extends T2> iterable;
+    final Func2<? super T1, ? super T2, ? extends R> zipFunction;
+
+    public OperatorZipIterable(Iterable<? extends T2> iterable, Func2<? super T1, ? super T2, ? extends R> zipFunction) {
+        this.iterable = iterable;
+        this.zipFunction = zipFunction;
+    }
+
+    @Override
+    public Subscriber<? super T1> call(final Subscriber<? super R> subscriber) {
+        final Iterator<? extends T2> iterator = iterable.iterator();
+        try {
+            if (!iterator.hasNext()) {
+                subscriber.onCompleted();
+                return Subscribers.create();
+            }
+        } catch (Throwable e) {
+            subscriber.onError(e);
+        }
+        return new Subscriber<T1>() {
+
+            @Override
+            public void onCompleted() {
+                subscriber.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                subscriber.onError(e);
+            }
+
+            @Override
+            public void onNext(T1 t) {
+                try {
+                    subscriber.onNext(zipFunction.call(t, iterator.next()));
+                    if (!iterator.hasNext()) {
+                        onCompleted();
+                    }
+                } catch (Throwable e) {
+                    onError(e);
+                }
+            }
+
+        };
+    }
+
+}