RunAsync method for outputting multiple values

Author: akarnokd

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));
+        
+    }
 }