Merge remote-tracking branch 'origin/master' into docs

Author: DavidMGross

rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/Async.java

@@ -20,10 +20,16 @@
 import java.util.concurrent.FutureTask;
 
 import rx.Observable;
+import rx.Observer;
 import rx.Scheduler;
 import rx.Scheduler.Inner;
+import rx.Subscriber;
+import rx.Subscription;
 import rx.schedulers.Schedulers;
 import rx.subjects.AsyncSubject;
+import rx.subjects.PublishSubject;
+import rx.subjects.Subject;
+import rx.subscriptions.SerialSubscription;
 import rx.util.async.operators.Functionals;
 import rx.util.async.operators.OperationDeferFuture;
 import rx.util.async.operators.OperationForEachFuture;
@@ -1711,4 +1717,54 @@ public static <R> Observable<R> fromCallable(Callable<? extends R> callable, Sch
     public static <R> Observable<R> fromRunnable(final Runnable run, final R result, Scheduler scheduler) {
         return Observable.create(OperationFromFunctionals.fromRunnable(run, result)).subscribeOn(scheduler);
     }
+    /**
+     * Runs the provided action on the given scheduler and allows propagation
+     * of multiple events to the observers of the returned StoppableObservable.
+     * The action is immediately executed and unobserved values will be lost.
+     * @param <T> the output value type
+     * @param scheduler the scheduler where the action is executed
+     * @param action the action to execute, receives an Observer where the events can be pumped
+     *               and a Subscription which lets check for cancellation condition.
+     * @return an Observable which provides a Subscription interface to cancel the action
+     */
+    public static <T> StoppableObservable<T> runAsync(Scheduler scheduler, 
+            final Action2<? super Observer<? super T>, ? super Subscription> action) {
+        return runAsync(scheduler, PublishSubject.<T>create(), action);
+    }
+    /**
+     * Runs the provided action on the given scheduler and allows propagation
+     * of multiple events to the observers of the returned StoppableObservable.
+     * The action is immediately executed and unobserved values might be lost,
+     * depending on the subject type used.
+     * @param <T> the output value of the action
+     * @param <U> the output type of the observable sequence
+     * @param scheduler the scheduler where the action is executed
+     * @param subject the subject to use to distribute values emitted by the action
+     * @param action the action to execute, receives an Observer where the events can be pumped
+     *               and a Subscription which lets check for cancellation condition.
+     * @return an Observable which provides a Subscription interface to cancel the action
+     */
+    public static <T, U> StoppableObservable<U> runAsync(Scheduler scheduler,
+            final Subject<T, U> subject, 
+            final Action2<? super Observer<? super T>, ? super Subscription> action) {
+        final SerialSubscription csub = new SerialSubscription();
+        
+        StoppableObservable<U> co = new StoppableObservable<U>(new Observable.OnSubscribe<U>() {
+            @Override
+            public void call(Subscriber<? super U> t1) {
+                subject.subscribe(t1);
+            }
+        }, csub);
+        
+        csub.set(scheduler.schedule(new Action1<Inner>() {
+            @Override
+            public void call(Inner t1) {
+                if (!csub.isUnsubscribed()) {
+                    action.call(subject, csub);
+                }
+            }
+        }));
+        
+        return co;
+    }
 }

rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/StoppableObservable.java

@@ -0,0 +1,41 @@
+/**
+ * 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.util.async;
+
+import rx.Observable;
+import rx.Subscription;
+
+/**
+ * An Observable which provides a Subscription interface to signal a stop
+ * condition to an asynchronous task.
+ */
+public class StoppableObservable<T> extends Observable<T> implements Subscription {
+    private final Subscription token;
+    public StoppableObservable(Observable.OnSubscribe<T> onSubscribe, Subscription token) {
+        super(onSubscribe);
+        this.token = token;
+    }
+    
+    @Override
+    public boolean isUnsubscribed() {
+        return token.isUnsubscribed();
+    }
+    
+    @Override
+    public void unsubscribe() {
+        token.unsubscribe();
+    }
+}

rxjava-contrib/rxjava-async-util/src/test/java/rx/util/async/AsyncTest.java

@@ -16,8 +16,8 @@
 
 package rx.util.async;
 
+import java.util.concurrent.CountDownLatch;
 import static org.junit.Assert.*;
-import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.concurrent.TimeUnit;
@@ -35,6 +35,7 @@
 
 import rx.Observable;
 import rx.Observer;
+import rx.Subscription;
 import rx.observers.TestObserver;
 import rx.schedulers.Schedulers;
 import rx.schedulers.TestScheduler;
@@ -818,4 +819,47 @@ public String answer(InvocationOnMock invocation) throws Throwable {
         verify(func, times(1)).call();
     }
 
+    @Test
+    public void testRunAsync() throws InterruptedException {
+        final CountDownLatch cdl = new CountDownLatch(1);
+        final CountDownLatch cdl2 = new CountDownLatch(1);
+        Action2<Observer<? super Integer>, Subscription> action = new Action2<Observer<? super Integer>, Subscription>() {
+            @Override
+            public void call(Observer<? super Integer> t1, Subscription t2) {
+                try {
+                    cdl.await();
+                } catch (InterruptedException ex) {
+                    Thread.currentThread().interrupt();
+                    return;
+                }
+                for (int i = 0; i < 10 && !t2.isUnsubscribed(); i++) {
+                    t1.onNext(i);
+                }
+                t1.onCompleted();
+                cdl2.countDown();
+            }
+        };
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        InOrder inOrder = inOrder(o);
+        
+        StoppableObservable<Integer> so = Async.<Integer>runAsync(Schedulers.io(), action);
+        
+        so.subscribe(o);
+        
+        cdl.countDown();
+        
+        if (!cdl2.await(2, TimeUnit.SECONDS)) {
+            fail("Didn't complete");
+        }
+        
+        for (int i = 0; i < 10; i++) {
+            inOrder.verify(o).onNext(i);
+        }
+        inOrder.verify(o).onCompleted();
+        inOrder.verifyNoMoreInteractions();
+        verify(o, never()).onError(any(Throwable.class));
+        
+    }
 }

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;
@@ -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;
@@ -1611,11 +1612,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));
     }
 
     /**
@@ -3024,7 +3023,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));
     }
 
     /**
@@ -3053,12 +3056,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));
     }
 
     /**
@@ -3084,8 +3089,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));
     }
 
     /**
@@ -3115,7 +3120,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));
     }
 
     /**
@@ -3147,7 +3152,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));
     }
 
     /**
@@ -3181,7 +3186,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));
     }
 
     /**
@@ -3217,7 +3222,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));
     }
 
     /**
@@ -3255,7 +3260,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));
     }
 
     /**
@@ -3295,7 +3300,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));
     }
 
     /**
@@ -3337,7 +3342,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));
     }
 
     /**
@@ -8372,7 +8377,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(OperationZip.zipIterable(this, other, zipFunction));
+        return lift(new OperatorZipIterable<T, T2, R>(other, zipFunction));
     }
 
     /**

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

@@ -47,27 +47,6 @@ protected Subscriber(Subscriber<?> op) {
         this(op.cs);
     }
 
-    public static <T> Subscriber<T> from(final Observer<? super T> o) {
-        return new Subscriber<T>() {
-
-            @Override
-            public void onCompleted() {
-                o.onCompleted();
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                o.onError(e);
-            }
-
-            @Override
-            public void onNext(T t) {
-                o.onNext(t);
-            }
-
-        };
-    }
-
     /**
      * Used to register an unsubscribe callback.
      */