Implemented Observable#toCompletable

Author: Aaron Tull

src/main/java/rx/Observable.java

@@ -232,6 +232,26 @@ public Single<T> toSingle() {
         return new Single<T>(OnSubscribeSingle.create(this));
     }
 
+    /**
+     * Returns a Completable that emits the single item emitted by the source Observable, if that Observable
+     * emits only a single item. This operator discards all onNext emissions. When onCompleted is called on the 
+     * source observable then the Completable will call onCompleted. Error terminal events are propagated.
+     * <p>
+     * <img width="640" height="295" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Completable.toCompletable.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code toCompletable} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @return a Completable that calls onCompleted on it's subscriber when the source Observable calls onCompleted
+     * @see <a href="http://reactivex.io/documentation/completable.html">ReactiveX documentation: Completable</a>
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public Completable toCompletable() {
+        return new Completable(OnSubscribeCompletable.create(this));
+    }
+    
 
     /* *********************************************************************************************************
      * Operators Below Here

src/main/java/rx/internal/operators/OnSubscribeCompletable.java

@@ -0,0 +1,75 @@
+/**
+ * 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.internal.operators;
+
+import rx.Completable.CompletableOnSubscribe;
+import rx.Completable.CompletableSubscriber;
+import rx.Observable;
+import rx.Single;
+import rx.SingleSubscriber;
+import rx.Subscriber;
+
+import java.util.NoSuchElementException;
+
+/**
+ * Allows conversion of an Observable to a Completable that will call onCompleted after an
+ * onCompleted event is emitted by the source observable. Also forwards errors as appropriate.
+ */
+public class OnSubscribeCompletable<T> implements CompletableOnSubscribe {
+
+    private final Observable<T> observable;
+
+    public OnSubscribeCompletable(Observable<T> observable) {
+        this.observable = observable;
+    }
+
+    @Override
+    public void call(final CompletableSubscriber child) {
+        Subscriber<T> parent = new Subscriber<T>() {
+            private boolean alreadyCompleted = false;
+            
+            @Override
+            public void onStart() {
+                request(Long.MAX_VALUE);
+            }
+
+            @Override
+            public void onCompleted() {
+                if (!alreadyCompleted) {
+                    alreadyCompleted = true;
+                    child.onCompleted();
+                }
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                child.onError(e);
+                unsubscribe();
+            }
+
+            @Override
+            public void onNext(T t) {
+                // do nothing at all
+            }
+        };
+        child.onSubscribe(parent);
+        observable.unsafeSubscribe(parent);
+    }
+
+    public static <T> OnSubscribeCompletable<T> create(Observable<T> observable) {
+        return new OnSubscribeCompletable<T>(observable);
+    }
+}

src/test/java/rx/internal/operators/OnSubscribeCompletableTest.java

@@ -0,0 +1,98 @@
+/**
+ * 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.internal.operators;
+
+import static org.junit.Assert.assertFalse;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Test;
+
+import rx.Completable;
+import rx.Observable;
+import rx.functions.Action0;
+import rx.observers.TestSubscriber;
+
+public class OnSubscribeCompletableTest {
+
+    @Test
+    public void testJustSingleItemObservable() {
+        TestSubscriber<String> subscriber = TestSubscriber.create();
+        Completable cmp = Observable.just("Hello World!").toCompletable();
+        cmp.subscribe(subscriber);
+
+        subscriber.assertNoValues();
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+    }
+
+    @Test
+    public void testErrorObservable() {
+        TestSubscriber<String> subscriber = TestSubscriber.create();
+        IllegalArgumentException error = new IllegalArgumentException("Error");
+        Completable cmp = Observable.<String>error(error).toCompletable();
+        cmp.subscribe(subscriber);
+
+        subscriber.assertError(error);
+        subscriber.assertNoValues();
+    }
+
+    @Test
+    public void testJustTwoEmissionsObservableThrowsError() {
+        TestSubscriber<String> subscriber = TestSubscriber.create();
+        Completable cmp = Observable.just("First", "Second").toCompletable();
+        cmp.subscribe(subscriber);
+
+        subscriber.assertNoErrors();
+        subscriber.assertNoValues();
+    }
+
+    @Test
+    public void testEmptyObservable() {
+        TestSubscriber<String> subscriber = TestSubscriber.create();
+        Completable cmp = Observable.<String>empty().toCompletable();
+        cmp.subscribe(subscriber);
+
+        subscriber.assertNoErrors();
+        subscriber.assertNoValues();
+        subscriber.assertCompleted();
+    }
+
+    @Test
+    public void testNeverObservable() {
+        TestSubscriber<String> subscriber = TestSubscriber.create();
+        Completable cmp = Observable.<String>never().toCompletable();
+        cmp.subscribe(subscriber);
+
+        subscriber.assertNoTerminalEvent();
+        subscriber.assertNoValues();
+    }
+
+    @Test
+    public void testShouldUseUnsafeSubscribeInternallyNotSubscribe() {
+        TestSubscriber<String> subscriber = TestSubscriber.create();
+        final AtomicBoolean unsubscribed = new AtomicBoolean(false);
+        Completable cmp = Observable.just("Hello World!").doOnUnsubscribe(new Action0() {
+
+            @Override
+            public void call() {
+                unsubscribed.set(true);
+            }}).toCompletable();
+        cmp.subscribe(subscriber);
+        subscriber.assertCompleted();
+        assertFalse(unsubscribed.get());
+    }
+}