2.x: added missing ops, cleanup 8/19-1 (ReactiveX#4375)

Author: akarnokd

src/main/java/io/reactivex/Completable.java

@@ -28,6 +28,7 @@
 import io.reactivex.internal.subscribers.completable.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subscribers.TestSubscriber;
 
 /**
  * Represents a deferred computation without any value but only indication for completion or exception.
@@ -165,33 +166,44 @@ public static Completable concat(Publisher<? extends CompletableSource> sources,
     }
 
     /**
-     * Constructs a Completable instance by wrapping the given source callback.
+     * Provides an API (via a cold Completable) that bridges the reactive world with the callback-style world.
+     * <p>
+     * Example:
+     * <pre><code>
+     * Completable.create(emitter -&gt; {
+     *     Callback listener = new Callback() {
+     *         &#64;Override
+     *         public void onEvent(Event e) {
+     *             emitter.onComplete();
+     *         }
+     *         
+     *         &#64;Override
+     *         public void onFailure(Exception e) {
+     *             emitter.onError(e);
+     *         }
+     *     };
+     *     
+     *     AutoCloseable c = api.someMethod(listener);
+     *     
+     *     emitter.setCancellable(c::close);
+     *     
+     * });
+     * </code></pre>
+     * <p>
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code create} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
-     * @param source the callback which will receive the CompletableObserver instances
-     * when the Completable is subscribed to.
-     * @return the created Completable instance
-     * @throws NullPointerException if source is null
+     * @param source the emitter that is called when a Subscriber subscribes to the returned {@code Flowable}
+     * @return the new Completable instance
+     * @see FlowableOnSubscribe
+     * @see FlowableEmitter.BackpressureMode
+     * @see Cancellable
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    public static Completable create(CompletableSource source) {
+    public static Completable create(CompletableOnSubscribe source) {
         Objects.requireNonNull(source, "source is null");
-        if (source instanceof Completable) {
-            throw new IllegalArgumentException("Use of create(Completable)!");
-        }
-        try {
-            // TODO plugin wrapping source
-
-            return RxJavaPlugins.onAssembly(new CompletableFromSource(source));
-        } catch (NullPointerException ex) { // NOPMD
-            throw ex;
-        } catch (Throwable ex) {
-            Exceptions.throwIfFatal(ex);
-            RxJavaPlugins.onError(ex);
-            throw toNpe(ex);
-        }
+        return RxJavaPlugins.onAssembly(new CompletableCreate(source));
     }
 
     /**
@@ -214,8 +226,6 @@ public static Completable unsafeCreate(CompletableSource source) {
             throw new IllegalArgumentException("Use of unsafeCreate(Completable)!");
         }
         try {
-            // TODO plugin wrapping source
-            
             return RxJavaPlugins.onAssembly(new CompletableFromUnsafeSource(source));
         } catch (NullPointerException ex) { // NOPMD
             throw ex;
@@ -1696,4 +1706,33 @@ public final Completable unsubscribeOn(final Scheduler scheduler) {
         Objects.requireNonNull(scheduler, "scheduler is null");
         return new CompletableUnsubscribeOn(this, scheduler);
     }
+    // -------------------------------------------------------------------------
+    // Fluent test support, super handy and reduces test preparation boilerplate
+    // -------------------------------------------------------------------------
+    
+    /**
+     * Creates a TestSubscriber and subscribes
+     * it to this Completable.
+     * @return the new TestSubscriber instance
+     * @since 2.0
+     */
+    public final TestSubscriber<Void> test() {
+        TestSubscriber<Void> ts = new TestSubscriber<Void>();
+        subscribe(ts);
+        return ts;
+    }
+
+    /**
+     * Creates a TestSubscriber optionally in cancelled state, then subscribes it to this Completable.
+     * @param cancelled if true, the TestSubscriber will be cancelled before subscribing to this
+     * Completable.
+     * @return the new TestSubscriber instance
+     * @since 2.0
+     */
+    public final TestSubscriber<Void> test(boolean cancelled) {
+        TestSubscriber<Void> ts = new TestSubscriber<Void>();
+        ts.dispose();
+        subscribe(ts);
+        return ts;
+    }
 }

src/main/java/io/reactivex/CompletableEmitter.java

@@ -0,0 +1,59 @@
+/**
+ * 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;
+
+import io.reactivex.disposables.Disposable;
+import io.reactivex.functions.Cancellable;
+
+/**
+ * Abstraction over a RxJava CompletableObserver that allows associating
+ * a resource with it.
+ * <p>
+ * All methods are safe to call from multiple threads.
+ * <p>
+ * Calling onComplete or onError multiple times has no effect.
+ */
+public interface CompletableEmitter {
+
+    /**
+     * Signal the completion.
+     */
+    void onComplete();
+    
+    /**
+     * Signal an exception.
+     * @param t the exception, not null
+     */
+    void onError(Throwable t);
+    
+    /**
+     * Sets a Disposable on this emitter; any previous Disposable
+     * or Cancellation will be unsubscribed/cancelled.
+     * @param d the disposable, null is allowed
+     */
+    void setDisposable(Disposable d);
+    
+    /**
+     * Sets a Cancellable on this emitter; any previous Disposable
+     * or Cancellation will be unsubscribed/cancelled.
+     * @param c the cancellable resource, null is allowed
+     */
+    void setCancellable(Cancellable c);
+    
+    /**
+     * Returns true if the downstream cancelled the sequence.
+     * @return true if the downstream cancelled the sequence
+     */
+    boolean isCancelled();
+}

src/main/java/io/reactivex/CompletableOnSubscribe.java

@@ -0,0 +1,29 @@
+/**
+ * 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;
+
+/**
+ * A functional interface that has a {@code subscribe()} method that receives
+ * an instance of a {@link CompletableEmitter} instance that allows pushing
+ * an event in a cancellation-safe manner.
+ */
+public interface CompletableOnSubscribe {
+    
+    /**
+     * Called for each CompletableObserver that subscribes.
+     * @param e the safe emitter instance, never null
+     * @throws Exception on error
+     */
+    void subscribe(CompletableEmitter e) throws Exception;
+}
+

src/main/java/io/reactivex/CompletableSource.java

@@ -15,9 +15,6 @@
 /**
  * Represents a basic {@link Completable} source base interface,
  * consumable via an {@link CompletableObserver}.
- * <p>
- * This class also serves the base type for custom operators wrapped into
- * Completable via {@link Completable#create(CompletableSource)}.
  * 
  * @since 2.0
  */

src/main/java/io/reactivex/Flowable.java

@@ -27,7 +27,6 @@
 import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.operators.completable.CompletableFromPublisher;
 import io.reactivex.internal.operators.flowable.*;
-import io.reactivex.internal.operators.flowable.FlowableConcatMap.ErrorMode;
 import io.reactivex.internal.operators.observable.ObservableFromPublisher;
 import io.reactivex.internal.operators.single.SingleFromPublisher;
 import io.reactivex.internal.schedulers.ImmediateThinScheduler;
@@ -1638,14 +1637,14 @@ public static <T> Flowable<T> concatEager(Iterable<? extends Publisher<? extends
      * @param source the emitter that is called when a Subscriber subscribes to the returned {@code Flowable}
      * @param mode the backpressure mode to apply if the downstream Subscriber doesn't request (fast) enough
      * @return the new Flowable instance
-     * @see FlowableSource
+     * @see FlowableOnSubscribe
      * @see FlowableEmitter.BackpressureMode
      * @see Cancellable
      */
     @BackpressureSupport(BackpressureKind.SPECIAL)
     @SchedulerSupport(SchedulerSupport.NONE)
-    public static <T> Flowable<T> create(FlowableSource<T> source, FlowableEmitter.BackpressureMode mode) {
-        return new FlowableFromSource<T>(source, mode);
+    public static <T> Flowable<T> create(FlowableOnSubscribe<T> source, FlowableEmitter.BackpressureMode mode) {
+        return new FlowableCreate<T>(source, mode);
     }
 
     /**
@@ -15454,8 +15453,8 @@ public final TestSubscriber<T> test(long initialRequest) { // NoPMD
      * @param cancelled if true, the TestSubscriber will be cancelled before subscribing to this
      * Flowable.
      * @return the new TestSubscriber instance
-      * @since 2.0
-    */
+     * @since 2.0
+     */
     public final TestSubscriber<T> test(long initialRequest, int fusionMode, boolean cancelled) { // NoPMD
         TestSubscriber<T> ts = new TestSubscriber<T>(initialRequest);
         ts.setInitialFusionMode(fusionMode);

src/main/java/io/reactivex/FlowableEmitter.java

@@ -59,7 +59,7 @@ public interface FlowableEmitter<T> {
      * or Cancellation will be unsubscribed/cancelled.
      * @param c the cancellable resource, null is allowed
      */
-    void setCancellation(Cancellable c);
+    void setCancellable(Cancellable c);
     /**
      * The current outstanding request amount.
      * <p>This method it threadsafe.

src/main/java/io/reactivex/FlowableOnSubscribe.java

@@ -0,0 +1,31 @@
+/**
+ * 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;
+
+/**
+ * A functional interface that has a {@code subscribe()} method that receives
+ * an instance of a {@link FlowableEmitter} instance that allows pushing
+ * events in a backpressure-safe and cancellation-safe manner.
+ *
+ * @param <T> the value type pushed
+ */
+public interface FlowableOnSubscribe<T> {
+    
+    /**
+     * Called for each Subscriber that subscribes.
+     * @param e the safe emitter instance, never null
+     * @throws Exception on error
+     */
+    void subscribe(FlowableEmitter<T> e) throws Exception;
+}
+