Implemented Completable#andThen(Observable)

Author: Aaron Tull

src/main/java/rx/Completable.java

@@ -1080,6 +1080,21 @@ public final Completable compose(CompletableTransformer transformer) {
         return to(transformer);
     }
 
+    /**
+     * Returns an Observable which will subscribe to this Completable and once that is completed then 
+     * will subscribe to the {@code next} Observable. An error event from this Completable will be 
+     * propagated to the downstream subscriber and will result in skipping the subscription of the 
+     * Observable.  
+     * 
+     * @param next the Observable to subscribe after this Completable is completed, not null
+     * @return Observable that composes this Completable and next
+     * @throws NullPointerException if next is null
+     */
+    public final <T> Observable<T> andThen(Observable<T> next) {
+        requireNonNull(next);
+        return next.delaySubscription(toObservable());
+    }
+    
     /**
      * Concatenates this Completable with another Completable.
      * @param other the other Completable, not null

src/test/java/rx/CompletableTest.java

@@ -357,6 +357,56 @@ public void call(Long v) {
         Assert.assertEquals(Arrays.asList(5L, 1L, 1L, 1L, 1L, 1L, 1L, 1L, 1L, 1L, 1L), requested);
     }
 
+    @Test
+    public void andThen() {
+        TestSubscriber<String> ts = new TestSubscriber<>(0);
+        Completable.complete().andThen(Observable.just("foo")).subscribe(ts);
+        ts.requestMore(1);
+        ts.assertValue("foo");
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
+    
+    @Test
+    public void andThenNever() {
+        TestSubscriber<String> ts = new TestSubscriber<>(0);
+        Completable.never().andThen(Observable.just("foo")).subscribe(ts);
+        ts.requestMore(1);
+        ts.assertNoValues();
+        ts.assertNoTerminalEvent();
+    }
+    
+    @Test
+    public void andThenError() {
+        TestSubscriber<String> ts = new TestSubscriber<>(0);
+        AtomicBoolean hasRun = new AtomicBoolean(false);
+        Exception e = new Exception();
+        Completable.create(cs -> cs.onError(e))
+            .andThen(Observable.<String>create(s-> {
+                hasRun.set(true);
+                s.onNext("foo");
+                s.onCompleted();
+            }))
+            .subscribe(ts);
+        ts.assertNoValues();
+        ts.assertError(e);
+        Assert.assertFalse("Should not have subscribed to observable when completable errors", hasRun.get());
+    }
+    
+    @Test
+    public void andThenSubscribeOn() {
+        TestSubscriber<String> ts = new TestSubscriber<>(0);
+        TestScheduler scheduler = new TestScheduler();
+        Completable.complete().andThen(Observable.just("foo").delay(1, TimeUnit.SECONDS, scheduler)).subscribe(ts);
+        ts.requestMore(1);
+        ts.assertNoValues();
+        ts.assertNoTerminalEvent();
+        scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+        ts.assertValue("foo");
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
+    
     @Test(expected = NullPointerException.class)
     public void createNull() {
         Completable.create(null);