2.x: move DisposableObserver to public area, add some javadocs (#4337)

akarnokd · web-flow · commit 2c3ec3802ef2 · 2016-08-10T19:31:11.000+02:00
diff --git a/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableLatest.java b/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableLatest.java
@@ -21,7 +21,7 @@
 import io.reactivex.Observable;
 import io.reactivex.Optional;
 import io.reactivex.exceptions.Exceptions;
-import io.reactivex.internal.subscribers.observable.DisposableObserver;
+import io.reactivex.observers.DisposableObserver;
 
 /**
  * Wait for and iterate over the latest values of the source observable. If the source works faster than the
diff --git a/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableNext.java b/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableNext.java
@@ -21,7 +21,7 @@
 import io.reactivex.Optional;
 import io.reactivex.Try;
 import io.reactivex.exceptions.Exceptions;
-import io.reactivex.internal.subscribers.observable.DisposableObserver;
+import io.reactivex.observers.DisposableObserver;
 
 /**
  * Returns an Iterable that blocks until the Observable emits another item, then returns that item.
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableBufferBoundary.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableBufferBoundary.java
@@ -26,7 +26,7 @@
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.observable.*;
 import io.reactivex.internal.util.QueueDrainHelper;
-import io.reactivex.observers.SerializedObserver;
+import io.reactivex.observers.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableBufferBoundary<T, U extends Collection<? super T>, Open, Close> 
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableBufferBoundarySupplier.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableBufferBoundarySupplier.java
@@ -23,7 +23,7 @@
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.observable.*;
 import io.reactivex.internal.util.QueueDrainHelper;
-import io.reactivex.observers.SerializedObserver;
+import io.reactivex.observers.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableBufferBoundarySupplier<T, U extends Collection<? super T>, B> 
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableBufferExactBoundary.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableBufferExactBoundary.java
@@ -22,7 +22,7 @@
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.observable.*;
 import io.reactivex.internal.util.QueueDrainHelper;
-import io.reactivex.observers.SerializedObserver;
+import io.reactivex.observers.*;
 
 public final class ObservableBufferExactBoundary<T, U extends Collection<? super T>, B> 
 extends AbstractObservableWithUpstream<T, U> {
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableDebounce.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableDebounce.java
@@ -19,8 +19,7 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.disposables.DisposableHelper;
-import io.reactivex.internal.subscribers.observable.DisposableObserver;
-import io.reactivex.observers.SerializedObserver;
+import io.reactivex.observers.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableDebounce<T, U> extends AbstractObservableWithUpstream<T, T> {
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableTimeout.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableTimeout.java
@@ -21,7 +21,7 @@
 import io.reactivex.functions.Function;
 import io.reactivex.internal.disposables.*;
 import io.reactivex.internal.subscribers.observable.*;
-import io.reactivex.observers.SerializedObserver;
+import io.reactivex.observers.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableTimeout<T, U, V> extends AbstractObservableWithUpstream<T, T> {
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundary.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundary.java
@@ -23,7 +23,7 @@
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.observable.*;
 import io.reactivex.internal.util.NotificationLite;
-import io.reactivex.observers.SerializedObserver;
+import io.reactivex.observers.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.UnicastSubject;
 
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundarySelector.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundarySelector.java
@@ -27,7 +27,7 @@
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.observable.*;
 import io.reactivex.internal.util.NotificationLite;
-import io.reactivex.observers.SerializedObserver;
+import io.reactivex.observers.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.UnicastSubject;
 
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundarySupplier.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundarySupplier.java
@@ -23,7 +23,7 @@
 import io.reactivex.internal.queue.MpscLinkedQueue;
 import io.reactivex.internal.subscribers.observable.*;
 import io.reactivex.internal.util.NotificationLite;
-import io.reactivex.observers.SerializedObserver;
+import io.reactivex.observers.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.UnicastSubject;
 
diff --git a/src/main/java/io/reactivex/observers/DisposableObserver.java b/src/main/java/io/reactivex/observers/DisposableObserver.java
@@ -11,7 +11,7 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.internal.subscribers.observable;
+package io.reactivex.observers;
 
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -20,9 +20,9 @@
 import io.reactivex.internal.disposables.DisposableHelper;
 
 /**
- * An abstract subscription that allows asynchronous cancellation.
+ * An abstract Observer that allows asynchronous cancellation by implementing Disposable.
  * 
- * @param <T>
+ * @param <T> the received value type
  */
 public abstract class DisposableObserver<T> implements Observer<T>, Disposable {
     final AtomicReference<Disposable> s = new AtomicReference<Disposable>();
@@ -34,6 +34,9 @@ public final void onSubscribe(Disposable s) {
         }
     }
     
+    /**
+     * Called once the single upstream Disposable is set via onSubscribe.
+     */
     protected void onStart() {
     }
     
diff --git a/src/main/java/io/reactivex/observers/Observers.java b/src/main/java/io/reactivex/observers/Observers.java
@@ -18,7 +18,6 @@
 import io.reactivex.functions.Consumer;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.functions.*;
-import io.reactivex.internal.subscribers.observable.DisposableObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
diff --git a/src/main/java/io/reactivex/subscribers/DisposableSubscriber.java b/src/main/java/io/reactivex/subscribers/DisposableSubscriber.java
@@ -21,9 +21,9 @@
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 
 /**
- * An abstract subscription that allows asynchronous cancellation.
+ * An abstract Subscriber that allows asynchronous cancellation by implementing Disposable.
  * 
- * @param <T>
+ * @param <T> the received value type.
  */
 public abstract class DisposableSubscriber<T> implements Subscriber<T>, Disposable {
     final AtomicReference<Subscription> s = new AtomicReference<Subscription>();
@@ -35,18 +35,38 @@ public final void onSubscribe(Subscription s) {
         }
     }
     
+    /**
+     * Returns the current Subscription sent to this Subscriber via onSubscribe().
+     * @return the current Subscription, may be null
+     */
     protected final Subscription subscription() {
         return s.get();
     }
     
+    /**
+     * Called once the single upstream Subscription is set via onSubscribe.
+     */
     protected void onStart() {
         s.get().request(Long.MAX_VALUE);
     }
     
+    /**
+     * Requests the specified amount from the upstream if its Subscription is set via
+     * onSubscribe already.
+     * <p>Note that calling this method before a Subscription is set via onSubscribe
+     * leads to NullPointerException and meant to be called from inside onStart or
+     * onNext.
+     * @param n the request amount, positive
+     */
     protected final void request(long n) {
         s.get().request(n);
     }
     
+    /**
+     * Cancels the Subscription set via onSubscribe or makes sure a
+     * Subscription set asynchronously (later) is cancelled immediately.
+     * <p>This method is thread-safe and can be exposed as a public API.
+     */
     protected final void cancel() {
         dispose();
     }
