2.x: Test static from methods and add Maybe.fromSingle & fromCompletable

src/main/java/io/reactivex/Maybe.java

@@ -563,6 +563,42 @@ public static <T> Maybe<T> fromAction(final Action run) {
         return RxJavaPlugins.onAssembly(new MaybeFromAction<T>(run));
     }
 
+    /**
+     * Wraps a CompletableSource into a Maybe.
+     *
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code fromCompletable} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param <T> the target type
+     * @param completableSource the CompletableSource to convert from
+     * @return the new Maybe instance
+     * @throws NullPointerException if completable is null
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public static <T> Maybe<T> fromCompletable(CompletableSource completableSource) {
+        ObjectHelper.requireNonNull(completableSource, "completableSource is null");
+        return RxJavaPlugins.onAssembly(new MaybeFromCompletable<T>(completableSource));
+    }
+
+    /**
+     * Wraps a SingleSource into a Maybe.
+     *
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code fromSingle} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param <T> the target type
+     * @param singleSource the SingleSource to convert from
+     * @return the new Maybe instance
+     * @throws NullPointerException if single is null
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public static <T> Maybe<T> fromSingle(SingleSource singleSource) {
+        ObjectHelper.requireNonNull(singleSource, "singleSource is null");
+        return RxJavaPlugins.onAssembly(new MaybeFromSingle<T>(singleSource));
+    }
+
     /**
      * Returns a {@link Maybe} that invokes passed function and emits its result for each new MaybeObserver that subscribes
      * while considering {@code null} value from the callable as indication for valueless completion.

src/main/java/io/reactivex/internal/operators/completable/CompletableFromObservable.java

@@ -26,28 +26,34 @@ public CompletableFromObservable(ObservableSource<T> observable) {
 
     @Override
     protected void subscribeActual(final CompletableObserver s) {
-        observable.subscribe(new Observer<T>() {
+        observable.subscribe(new CompletableFromObservableObserver<T>(s));
+    }
+
+    static final class CompletableFromObservableObserver<T> implements Observer<T> {
+        final CompletableObserver co;
 
-            @Override
-            public void onComplete() {
-                s.onComplete();
-            }
+        CompletableFromObservableObserver(CompletableObserver co) {
+            this.co = co;
+        }
 
-            @Override
-            public void onError(Throwable e) {
-                s.onError(e);
-            }
+        @Override
+        public void onSubscribe(Disposable d) {
+            co.onSubscribe(d);
+        }
 
-            @Override
-            public void onNext(T value) {
-                // ignored
-            }
+        @Override
+        public void onNext(T value) {
+            // Deliberately ignored.
+        }
 
-            @Override
-            public void onSubscribe(Disposable d) {
-                s.onSubscribe(d);
-            }
+        @Override
+        public void onError(Throwable e) {
+            co.onError(e);
+        }
 
-        });
+        @Override
+        public void onComplete() {
+            co.onComplete();
+        }
     }
 }

src/main/java/io/reactivex/internal/operators/completable/CompletableFromSingle.java

@@ -26,24 +26,29 @@ public CompletableFromSingle(SingleSource<T> single) {
 
     @Override
     protected void subscribeActual(final CompletableObserver s) {
-        single.subscribe(new SingleObserver<T>() {
+        single.subscribe(new CompletableFromSingleObserver<T>(s));
+    }
 
-            @Override
-            public void onError(Throwable e) {
-                s.onError(e);
-            }
+    static final class CompletableFromSingleObserver<T> implements SingleObserver<T> {
+        final CompletableObserver co;
 
-            @Override
-            public void onSubscribe(Disposable d) {
-                s.onSubscribe(d);
-            }
+        CompletableFromSingleObserver(CompletableObserver co) {
+            this.co = co;
+        }
 
-            @Override
-            public void onSuccess(T value) {
-                s.onComplete();
-            }
+        @Override
+        public void onError(Throwable e) {
+            co.onError(e);
+        }
 
-        });
-    }
+        @Override
+        public void onSubscribe(Disposable d) {
+            co.onSubscribe(d);
+        }
 
+        @Override
+        public void onSuccess(T value) {
+            co.onComplete();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/completable/CompletableFromActionTest.java

@@ -0,0 +1,100 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.operators.completable;
+
+import io.reactivex.Completable;
+import io.reactivex.functions.Action;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class CompletableFromActionTest {
+    @Test(expected = NullPointerException.class)
+    public void fromActionNull() {
+        Completable.fromAction(null);
+    }
+
+    @Test
+    public void fromAction() {
+        final AtomicInteger atomicInteger = new AtomicInteger();
+
+        Completable.fromAction(new Action() {
+            @Override
+            public void run() throws Exception {
+                atomicInteger.incrementAndGet();
+            }
+        })
+            .test()
+            .assertResult();
+
+        assertEquals(1, atomicInteger.get());
+    }
+
+    @Test
+    public void fromActionTwice() {
+        final AtomicInteger atomicInteger = new AtomicInteger();
+
+        Action run = new Action() {
+            @Override
+            public void run() throws Exception {
+                atomicInteger.incrementAndGet();
+            }
+        };
+
+        Completable.fromAction(run)
+            .test()
+            .assertResult();
+
+        assertEquals(1, atomicInteger.get());
+
+        Completable.fromAction(run)
+            .test()
+            .assertResult();
+
+        assertEquals(2, atomicInteger.get());
+    }
+
+    @Test
+    public void fromActionInvokesLazy() {
+        final AtomicInteger atomicInteger = new AtomicInteger();
+
+        Completable completable = Completable.fromAction(new Action() {
+            @Override
+            public void run() throws Exception {
+                atomicInteger.incrementAndGet();
+            }
+        });
+
+        assertEquals(0, atomicInteger.get());
+
+        completable
+            .test()
+            .assertResult();
+
+        assertEquals(1, atomicInteger.get());
+    }
+
+    @Test
+    public void fromActionThrows() {
+        Completable.fromAction(new Action() {
+            @Override
+            public void run() throws Exception {
+                throw new UnsupportedOperationException();
+            }
+        })
+            .test()
+            .assertFailure(UnsupportedOperationException.class);
+    }
+}

src/test/java/io/reactivex/internal/operators/completable/CompletableFromCallableTest.java

@@ -0,0 +1,103 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.operators.completable;
+
+import io.reactivex.Completable;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class CompletableFromCallableTest {
+    @Test(expected = NullPointerException.class)
+    public void fromCallableNull() {
+        Completable.fromCallable(null);
+    }
+
+    @Test
+    public void fromCallable() {
+        final AtomicInteger atomicInteger = new AtomicInteger();
+
+        Completable.fromCallable(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                atomicInteger.incrementAndGet();
+                return null;
+            }
+        })
+            .test()
+            .assertResult();
+
+        assertEquals(1, atomicInteger.get());
+    }
+
+    @Test
+    public void fromCallableTwice() {
+        final AtomicInteger atomicInteger = new AtomicInteger();
+
+        Callable<Object> callable = new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                atomicInteger.incrementAndGet();
+                return null;
+            }
+        };
+
+        Completable.fromCallable(callable)
+            .test()
+            .assertResult();
+
+        assertEquals(1, atomicInteger.get());
+
+        Completable.fromCallable(callable)
+            .test()
+            .assertResult();
+
+        assertEquals(2, atomicInteger.get());
+    }
+
+    @Test
+    public void fromCallableInvokesLazy() {
+        final AtomicInteger atomicInteger = new AtomicInteger();
+
+        Completable completable = Completable.fromCallable(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                atomicInteger.incrementAndGet();
+                return null;
+            }
+        });
+
+        assertEquals(0, atomicInteger.get());
+
+        completable
+            .test()
+            .assertResult();
+
+        assertEquals(1, atomicInteger.get());
+    }
+
+    @Test
+    public void fromCallableThrows() {
+        Completable.fromCallable(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                throw new UnsupportedOperationException();
+            }
+        })
+            .test()
+            .assertFailure(UnsupportedOperationException.class);
+    }
+}

src/test/java/io/reactivex/internal/operators/completable/CompletableFromObservableTest.java

@@ -0,0 +1,46 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.operators.completable;
+
+import io.reactivex.Completable;
+import io.reactivex.Observable;
+import org.junit.Test;
+
+public class CompletableFromObservableTest {
+    @Test(expected = NullPointerException.class)
+    public void fromObservableNull() {
+        Completable.fromObservable(null);
+    }
+
+    @Test
+    public void fromObservable() {
+        Completable.fromObservable(Observable.just(1))
+            .test()
+            .assertResult();
+    }
+
+    @Test
+    public void fromObservableEmpty() {
+        Completable.fromObservable(Observable.empty())
+            .test()
+            .assertResult();
+    }
+
+    @Test
+    public void fromObservableError() {
+        Completable.fromObservable(Observable.error(new UnsupportedOperationException()))
+            .test()
+            .assertFailure(UnsupportedOperationException.class);
+    }
+}