rename AsyncEmitter to Emitter (#4580)

Author: davidmoten

src/main/java/rx/AsyncEmitter.java

@@ -29,8 +29,10 @@
  *
  * @param <T> the value type to emit
  * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+ * @deprecated as of 1.2.1 because Async prefix of this class potentially misleading. Use {@link Emitter} instead.
  */
 @Experimental
+@Deprecated
 public interface AsyncEmitter<T> extends Observer<T> {
 
     /**

src/main/java/rx/CompletableEmitter.java

@@ -25,8 +25,10 @@
  * All methods are thread-safe; calling onCompleted or onError twice or one after the other has
  * no effect.
  * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+ * @deprecated as of 1.2.1 because `setCancellation` method will change signature (not name) for 1.2.2.
  */
 @Experimental
+@Deprecated
 public interface CompletableEmitter {
 
     /**

src/main/java/rx/Emitter.java

@@ -0,0 +1,85 @@
+/**
+ * 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 rx;
+
+import rx.annotations.Experimental;
+import rx.functions.Cancellable;
+
+/**
+ * Abstraction over a RxJava Subscriber that allows associating
+ * a resource with it and exposes the current number of downstream
+ * requested amount.
+ * <p>
+ * The onNext, onError and onCompleted methods should be called
+ * in a sequential manner, just like the Observer's methods. The
+ * other methods are thread-safe.
+ *
+ * @param <T> the value type to emit
+ * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+ */
+@Experimental
+public interface Emitter<T> extends Observer<T> {
+
+    /**
+     * Sets a Subscription on this emitter; any previous Subscription
+     * or Cancellation will be unsubscribed/cancelled.
+     * @param s the subscription, null is allowed
+     */
+    void setSubscription(Subscription s);
+
+    /**
+     * Sets a Cancellable on this emitter; any previous Subscription
+     * or Cancellation will be unsubscribed/cancelled.
+     * @param c the cancellable resource, null is allowed
+     */
+    void setCancellation(Cancellable c);
+    /**
+     * The current outstanding request amount.
+     * <p>This method it thread-safe.
+     * @return the current outstanding request amount
+     */
+    long requested();
+
+    /**
+     * Options to handle backpressure in the emitter.
+     */
+    enum BackpressureMode {
+        /**
+         * No backpressure is applied as the onNext calls pass through the Emitter;
+         * note that this may cause {@link rx.exceptions.MissingBackpressureException} or {@link IllegalStateException}
+         * somewhere downstream.
+         */
+        NONE,
+        /**
+         * Signals a {@link rx.exceptions.MissingBackpressureException} if the downstream can't keep up.
+         */
+        ERROR,
+        /**
+         * Buffers (unbounded) all onNext calls until the downstream can consume them.
+         */
+        BUFFER,
+        /**
+         * Drops the incoming onNext value if the downstream can't keep up.
+         */
+        DROP,
+        /**
+         * Keeps the latest onNext value and overwrites it with newer ones until the downstream
+         * can consume it.
+         */
+        LATEST
+    }
+}

src/main/java/rx/Observable.java

@@ -2016,9 +2016,58 @@ public static <T> Observable<T> from(T[] array) {
      * @see AsyncEmitter.BackpressureMode
      * @see AsyncEmitter.Cancellable
      * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     * @deprecated since 1.2.1 because Async prefix of AsyncEmitter class is potentially misleading. Use 
+     *            {@link #fromEmitter(Action1, Emitter.BackpressureMode)} instead.
      */
     @Experimental
+    @Deprecated
     public static <T> Observable<T> fromEmitter(Action1<AsyncEmitter<T>> emitter, AsyncEmitter.BackpressureMode backpressure) {
+        return create(new OnSubscribeFromAsyncEmitter<T>(emitter, backpressure));
+    }
+    
+    /**
+     * Provides an API (via a cold Observable) that bridges the reactive world with the callback-style,
+     * generally non-backpressured world.
+     * <p>
+     * Example:
+     * <pre><code>
+     * Observable.&lt;Event&gt;fromEmitter(emitter -&gt; {
+     *     Callback listener = new Callback() {
+     *         &#64;Override
+     *         public void onEvent(Event e) {
+     *             emitter.onNext(e);
+     *             if (e.isLast()) {
+     *                 emitter.onCompleted();
+     *             }
+     *         }
+     *
+     *         &#64;Override
+     *         public void onFailure(Exception e) {
+     *             emitter.onError(e);
+     *         }
+     *     };
+     *
+     *     AutoCloseable c = api.someMethod(listener);
+     *
+     *     emitter.setCancellation(c::close);
+     *
+     * }, BackpressureMode.BUFFER);
+     * </code></pre>
+     * <p>
+     * You should call the Emitter's onNext, onError and onCompleted methods in a serialized fashion. The
+     * rest of its methods are thread-safe.
+     *
+     * @param <T> the element type
+     * @param emitter the emitter that is called when a Subscriber subscribes to the returned {@code Observable}
+     * @param backpressure the backpressure mode to apply if the downstream Subscriber doesn't request (fast) enough
+     * @return the new Observable instance
+     * @see Emitter
+     * @see Emitter.BackpressureMode
+     * @see Emitter.Cancellable
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public static <T> Observable<T> fromEmitter(Action1<Emitter<T>> emitter, Emitter.BackpressureMode backpressure) {
         return create(new OnSubscribeFromEmitter<T>(emitter, backpressure));
     }
 

src/main/java/rx/functions/Cancellable.java

@@ -0,0 +1,34 @@
+/**
+ * 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 rx.functions;
+
+import rx.annotations.Experimental;
+
+/**
+ * A functional interface that has a single close method that can throw.
+ */
+@Experimental
+public interface Cancellable {
+
+    /**
+     * Cancel the action or free a resource.
+     * 
+     * @throws Exception
+     *             on error
+     */
+    void cancel() throws Exception;
+}

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

@@ -21,7 +21,7 @@
 import rx.AsyncEmitter.Cancellable;
 import rx.exceptions.Exceptions;
 import rx.functions.Action1;
-import rx.internal.operators.OnSubscribeFromEmitter.CancellableSubscription;
+import rx.internal.operators.OnSubscribeFromAsyncEmitter.CancellableSubscription;
 import rx.internal.subscriptions.SequentialSubscription;
 import rx.plugins.RxJavaHooks;