* -
*
- Backpressure: - *
- This operator does not support backpressure as it uses time. It requests {@code Long.MAX_VALUE}
+ *
- This operator does not support backpressure as it uses time. It requests {@link Long#MAX_VALUE} * upstream and does not obey downstream requests.
* - Scheduler: *
- This version of {@code buffer} operates by default on the {@code computation} {@link Scheduler}. @@ -6293,7 +6293,7 @@ public final Flowable
- Backpressure: - *
- This operator does not support backpressure as it uses time. It requests {@code Long.MAX_VALUE}
+ *
- This operator does not support backpressure as it uses time. It requests {@link Long#MAX_VALUE} * upstream and does not obey downstream requests.
* - Scheduler: *
- You specify which {@link Scheduler} this operator will use. @@ -6330,7 +6330,7 @@ public final Flowable
- Backpressure: - *
- This operator does not support backpressure as it uses time. It requests {@code Long.MAX_VALUE}
+ *
- This operator does not support backpressure as it uses time. It requests {@link Long#MAX_VALUE} * upstream and does not obey downstream requests.
* - Scheduler: *
- You specify which {@link Scheduler} this operator will use. @@ -6375,7 +6375,7 @@ public final > Flowable buffer(long timespan, *
- Backpressure: - *
- This operator does not support backpressure as it uses time. It requests {@code Long.MAX_VALUE}
+ *
- This operator does not support backpressure as it uses time. It requests {@link Long#MAX_VALUE} * upstream and does not obey downstream requests.
* - Scheduler: *
- This version of {@code buffer} operates by default on the {@code computation} {@link Scheduler}. @@ -6408,7 +6408,7 @@ public final Flowable
- Backpressure: - *
- This operator does not support backpressure as it uses time. It requests {@code Long.MAX_VALUE}
+ *
- This operator does not support backpressure as it uses time. It requests {@link Long#MAX_VALUE} * upstream and does not obey downstream requests.
* - Scheduler: *
- This version of {@code buffer} operates by default on the {@code computation} {@link Scheduler}. @@ -6445,7 +6445,7 @@ public final Flowable
- Backpressure: - *
- This operator does not support backpressure as it uses time. It requests {@code Long.MAX_VALUE}
+ *
- This operator does not support backpressure as it uses time. It requests {@link Long#MAX_VALUE} * upstream and does not obey downstream requests.
* - Scheduler: *
- You specify which {@link Scheduler} this operator will use. @@ -6484,7 +6484,7 @@ public final Flowable
- Backpressure: - *
- This operator does not support backpressure as it uses time. It requests {@code Long.MAX_VALUE}
+ *
- This operator does not support backpressure as it uses time. It requests {@link Long#MAX_VALUE} * upstream and does not obey downstream requests.
* - Scheduler: *
- You specify which {@link Scheduler} this operator will use. @@ -6536,7 +6536,7 @@ public final > Flowable buffer( *
- Backpressure: - *
- This operator does not support backpressure as it uses time. It requests {@code Long.MAX_VALUE}
+ *
- This operator does not support backpressure as it uses time. It requests {@link Long#MAX_VALUE} * upstream and does not obey downstream requests.
* - Scheduler: *
- You specify which {@link Scheduler} this operator will use. @@ -6571,7 +6571,7 @@ public final Flowable
- Backpressure: *
- This operator does not support backpressure as it is instead controlled by the given Publishers and - * buffers data. It requests {@code Long.MAX_VALUE} upstream and does not obey downstream requests. + * buffers data. It requests {@link Long#MAX_VALUE} upstream and does not obey downstream requests. *
- Scheduler: *
- This version of {@code buffer} does not operate by default on a particular {@link Scheduler}. *
- Backpressure: *
- This operator does not support backpressure as it is instead controlled by the given Publishers and - * buffers data. It requests {@code Long.MAX_VALUE} upstream and does not obey downstream requests. + * buffers data. It requests {@link Long#MAX_VALUE} upstream and does not obey downstream requests. *
- Scheduler: *
- This version of {@code buffer} does not operate by default on a particular {@link Scheduler}. *
- Backpressure: *
- This operator does not support backpressure as it is instead controlled by the {@code Publisher} - * {@code boundary} and buffers data. It requests {@code Long.MAX_VALUE} upstream and does not obey + * {@code boundary} and buffers data. It requests {@link Long#MAX_VALUE} upstream and does not obey * downstream requests. *
- Scheduler: *
- This version of {@code buffer} does not operate by default on a particular {@link Scheduler}. @@ -6686,7 +6686,7 @@ public final Flowable
- Backpressure: *
- This operator does not support backpressure as it is instead controlled by the {@code Publisher} - * {@code boundary} and buffers data. It requests {@code Long.MAX_VALUE} upstream and does not obey + * {@code boundary} and buffers data. It requests {@link Long#MAX_VALUE} upstream and does not obey * downstream requests. *
- Scheduler: *
- This version of {@code buffer} does not operate by default on a particular {@link Scheduler}. @@ -6723,7 +6723,7 @@ public final Flowable
- Backpressure: *
- This operator does not support backpressure as it is instead controlled by the {@code Publisher} - * {@code boundary} and buffers data. It requests {@code Long.MAX_VALUE} upstream and does not obey + * {@code boundary} and buffers data. It requests {@link Long#MAX_VALUE} upstream and does not obey * downstream requests. *
- Scheduler: *
- This version of {@code buffer} does not operate by default on a particular {@link Scheduler}. @@ -8503,7 +8503,7 @@ public final
- Backpressure: *
- The operator forwards the backpressure requests to this Publisher once - * the subscription happens and requests Long.MAX_VALUE from the other Publisher + * the subscription happens and requests {@link Long#MAX_VALUE} from the other Publisher *
- Scheduler: *
- This method does not operate by default on a particular {@link Scheduler}. *
- Backpressure: - *
- If {@code maxConcurrency == Integer.MAX_VALUE} the operator consumes the upstream in an unbounded manner.
+ *
- If {@code maxConcurrency == }{@link Integer#MAX_VALUE} the operator consumes the upstream in an unbounded manner. * Otherwise, the operator expects the upstream to honor backpressure. If the upstream doesn't support backpressure - * the operator behaves as if {@code maxConcurrency == Integer.MAX_VALUE} was used.
+ * the operator behaves as if {@code maxConcurrency == }{@link Integer#MAX_VALUE} was used.
* - Scheduler: *
- {@code flatMapCompletable} does not operate by default on a particular {@link Scheduler}. *
- Backpressure: - *
- If {@code maxConcurrency == Integer.MAX_VALUE} the operator consumes the upstream in an unbounded manner.
+ *
- If {@code maxConcurrency == }{@link Integer#MAX_VALUE} the operator consumes the upstream in an unbounded manner. * Otherwise, the operator expects the upstream to honor backpressure. If the upstream doesn't support backpressure - * the operator behaves as if {@code maxConcurrency == Integer.MAX_VALUE} was used.
+ * the operator behaves as if {@code maxConcurrency == }{@link Integer#MAX_VALUE} was used.
* - Scheduler: *
- {@code flatMapMaybe} does not operate by default on a particular {@link Scheduler}. *
- Backpressure: - *
- If {@code maxConcurrency == Integer.MAX_VALUE} the operator consumes the upstream in an unbounded manner.
+ *
- If {@code maxConcurrency == }{@link Integer#MAX_VALUE} the operator consumes the upstream in an unbounded manner. * Otherwise, the operator expects the upstream to honor backpressure. If the upstream doesn't support backpressure - * the operator behaves as if {@code maxConcurrency == Integer.MAX_VALUE} was used.
+ * the operator behaves as if {@code maxConcurrency == }{@link Integer#MAX_VALUE} was used.
* - Scheduler: *
- {@code flatMapSingle} does not operate by default on a particular {@link Scheduler}. *
- Backpressure: *
- Observables don't support backpressure thus the current Flowable is consumed in an unbounded - * manner (by requesting Long.MAX_VALUE). + * manner (by requesting {@link Long#MAX_VALUE}). *
- Scheduler: *
- {@code toObservable} does not operate by default on a particular {@link Scheduler}. *
- Backpressure: *
- The returned TestSubscriber consumes this Flowable in an unbounded fashion. diff --git a/src/main/java/io/reactivex/rxjava3/core/FlowableConverter.java b/src/main/java/io/reactivex/rxjava3/core/FlowableConverter.java index abc89eec4a..92733c3c1e 100644 --- a/src/main/java/io/reactivex/rxjava3/core/FlowableConverter.java +++ b/src/main/java/io/reactivex/rxjava3/core/FlowableConverter.java @@ -16,7 +16,7 @@ import io.reactivex.rxjava3.annotations.NonNull; /** - * Convenience interface and callback used by the {@link Flowable#to} operator to turn a Flowable into another + * Convenience interface and callback used by the {@link Flowable#to} operator to turn a {@link Flowable} into another * value fluently. *
- Backpressure: *
- The {@code other} publisher is consumed in an unbounded fashion but will be
diff --git a/src/main/java/io/reactivex/rxjava3/core/SingleConverter.java b/src/main/java/io/reactivex/rxjava3/core/SingleConverter.java
index 4adc22b8c5..eb100b6c87 100644
--- a/src/main/java/io/reactivex/rxjava3/core/SingleConverter.java
+++ b/src/main/java/io/reactivex/rxjava3/core/SingleConverter.java
@@ -16,7 +16,7 @@
import io.reactivex.rxjava3.annotations.NonNull;
/**
- * Convenience interface and callback used by the {@link Single#to} operator to turn a Single into another
+ * Convenience interface and callback used by the {@link Single#to} operator to turn a {@link Single} into another
* value fluently.
*
History: 2.1.7 - experimental * @param
the upstream type @@ -26,9 +26,9 @@ @FunctionalInterface public interface SingleConverter { /** * Signal an exception. - * @param t the exception, not null + * @param t the exception, not {@code null} */ void onError(@NonNull Throwable t); /** - * Sets a Disposable on this emitter; any previous Disposable - * or Cancellable will be disposed/cancelled. - * @param d the disposable, null is allowed + * Sets a {@link Disposable} on this emitter; any previous {@code Disposable} + * or {@link Cancellable} will be disposed/cancelled. + * This method is thread-safe. + * @param d the {@code Disposable}, {@code null} is allowed */ void setDisposable(@Nullable Disposable d); /** * Sets a Cancellable on this emitter; any previous {@link Disposable} * or {@link Cancellable} will be disposed/cancelled. - * @param c the cancellable resource, null is allowed + *
This method is thread-safe. + * @param c the {@code Cancellable} resource, {@code null} is allowed */ void setCancellable(@Nullable Cancellable c); @@ -85,12 +87,12 @@ public interface SingleEmitter
{ boolean isDisposed(); /** - * Attempts to emit the specified {@code Throwable} error if the downstream + * Attempts to emit the specified {@link Throwable} error if the downstream * hasn't cancelled the sequence or is otherwise terminated, returning false * if the emission is not allowed to happen due to lifecycle restrictions. * - * Unlike {@link #onError(Throwable)}, the {@code RxJavaPlugins.onError} is not called - * if the error could not be delivered. + * Unlike {@link #onError(Throwable)}, the {@link io.reactivex.rxjava3.plugins.RxJavaPlugins#onError(Throwable) RxjavaPlugins.onError} + * is not called if the error could not be delivered. *
History: 2.1.1 - experimental * @param t the throwable error to signal if possible * @return true if successful, false if the downstream is not able to accept further diff --git a/src/main/java/io/reactivex/rxjava3/core/SingleObserver.java b/src/main/java/io/reactivex/rxjava3/core/SingleObserver.java index 778d6feabc..d2d788d38f 100644 --- a/src/main/java/io/reactivex/rxjava3/core/SingleObserver.java +++ b/src/main/java/io/reactivex/rxjava3/core/SingleObserver.java @@ -53,7 +53,7 @@ public interface SingleObserver
{ /** - * Provides the SingleObserver with the means of cancelling (disposing) the + * Provides the {@link SingleObserver} with the means of cancelling (disposing) the * connection (channel) with the Single in both * synchronous (from within {@code onSubscribe(Disposable)} itself) and asynchronous manner. * @param d the Disposable instance whose {@link Disposable#dispose()} can @@ -63,23 +63,23 @@ public interface SingleObserver { void onSubscribe(@NonNull Disposable d); /** - * Notifies the SingleObserver with a single item and that the {@link Single} has finished sending + * Notifies the {@link SingleObserver} with a single item and that the {@link Single} has finished sending * push-based notifications. * - * The {@link Single} will not call this method if it calls {@link #onError}. + * The {@code Single} will not call this method if it calls {@link #onError}. * * @param t - * the item emitted by the Single + * the item emitted by the {@code Single} */ void onSuccess(@NonNull T t); /** - * Notifies the SingleObserver that the {@link Single} has experienced an error condition. + * Notifies the {@link SingleObserver} that the {@link Single} has experienced an error condition. *
- * If the {@link Single} calls this method, it will not thereafter call {@link #onSuccess}. + * If the {@code Single} calls this method, it will not thereafter call {@link #onSuccess}. * * @param e - * the exception encountered by the Single + * the exception encountered by the {@code Single} */ void onError(@NonNull Throwable e); } diff --git a/src/main/java/io/reactivex/rxjava3/core/SingleOnSubscribe.java b/src/main/java/io/reactivex/rxjava3/core/SingleOnSubscribe.java index 46cc6a6f56..7a6a5adf68 100644 --- a/src/main/java/io/reactivex/rxjava3/core/SingleOnSubscribe.java +++ b/src/main/java/io/reactivex/rxjava3/core/SingleOnSubscribe.java @@ -25,8 +25,8 @@ public interface SingleOnSubscribe
{ /** - * Called for each SingleObserver that subscribes. - * @param emitter the safe emitter instance, never null + * Called for each {@link SingleObserver} that subscribes. + * @param emitter the safe emitter instance, never {@code null} * @throws Throwable on error */ void subscribe(@NonNull SingleEmitter emitter) throws Throwable; diff --git a/src/main/java/io/reactivex/rxjava3/core/SingleOperator.java b/src/main/java/io/reactivex/rxjava3/core/SingleOperator.java index 2fc2d80ab6..de2f38f23d 100644 --- a/src/main/java/io/reactivex/rxjava3/core/SingleOperator.java +++ b/src/main/java/io/reactivex/rxjava3/core/SingleOperator.java @@ -16,7 +16,7 @@ import io.reactivex.rxjava3.annotations.NonNull; /** - * Interface to map/wrap a downstream observer to an upstream observer. + * Interface to map/wrap a downstream {@link SingleObserver} to an upstream {@code SingleObserver}. * * @param the value type of the downstream * @param the value type of the upstream @@ -24,11 +24,11 @@ @FunctionalInterface public interface SingleOperator { /** - * Subscribes the given SingleObserver to this SingleSource instance. - * @param observer the SingleObserver, not null - * @throws NullPointerException if {@code observer} is null + * Subscribes the given {@link SingleObserver} to this {@link SingleSource} instance. + * @param observer the {@code SingleObserver}, not {@code null} + * @throws NullPointerException if {@code observer} is {@code null} */ void subscribe(@NonNull SingleObserver observer); } diff --git a/src/main/java/io/reactivex/rxjava3/core/SingleTransformer.java b/src/main/java/io/reactivex/rxjava3/core/SingleTransformer.java index 08f9408cdb..1f0d478a0e 100644 --- a/src/main/java/io/reactivex/rxjava3/core/SingleTransformer.java +++ b/src/main/java/io/reactivex/rxjava3/core/SingleTransformer.java @@ -16,7 +16,7 @@ import io.reactivex.rxjava3.annotations.NonNull; /** - * Interface to compose Singles. + * Interface to compose {@link Single}s. * * @param the upstream value type * @param the downstream value type @@ -24,10 +24,10 @@ @FunctionalInterface public interface SingleTransformer apply(@NonNull Single upstream); diff --git a/src/main/java/io/reactivex/rxjava3/disposables/CompositeDisposable.java b/src/main/java/io/reactivex/rxjava3/disposables/CompositeDisposable.java index c9c0a8988c..3a75ac5f3e 100644 --- a/src/main/java/io/reactivex/rxjava3/disposables/CompositeDisposable.java +++ b/src/main/java/io/reactivex/rxjava3/disposables/CompositeDisposable.java @@ -37,7 +37,7 @@ public CompositeDisposable() { /** * Creates a CompositeDisposables with the given array of initial elements. * @param disposables the array of Disposables to start with - * @throws NullPointerException if {@code disposables} or any of its array items is null + * @throws NullPointerException if {@code disposables} or any of its array items is {@code null} */ public CompositeDisposable(@NonNull Disposable... disposables) { Objects.requireNonNull(disposables, "disposables is null"); @@ -51,7 +51,7 @@ public CompositeDisposable(@NonNull Disposable... disposables) { /** * Creates a CompositeDisposables with the given Iterable sequence of initial elements. * @param disposables the Iterable sequence of Disposables to start with - * @throws NullPointerException if {@code disposables} or any of its items is null + * @throws NullPointerException if {@code disposables} or any of its items is {@code null} */ public CompositeDisposable(@NonNull Iterable disposables) { Objects.requireNonNull(disposables, "disposables is null"); @@ -90,7 +90,7 @@ public boolean isDisposed() { * container has been disposed. * @param disposable the disposable to add, not null * @return true if successful, false if this container has been disposed - * @throws NullPointerException if {@code disposable} is null + * @throws NullPointerException if {@code disposable} is {@code null} */ @Override public boolean add(@NonNull Disposable disposable) { @@ -117,7 +117,7 @@ public boolean add(@NonNull Disposable disposable) { * disposes them all if the container has been disposed. * @param disposables the array of Disposables * @return true if the operation was successful, false if the container has been disposed - * @throws NullPointerException if {@code disposables} or any of its array items is null + * @throws NullPointerException if {@code disposables} or any of its array items is {@code null} */ public boolean addAll(@NonNull Disposable... disposables) { Objects.requireNonNull(disposables, "disposables is null"); @@ -163,7 +163,7 @@ public boolean remove(@NonNull Disposable disposable) { * container. * @param disposable the disposable to remove, not null * @return true if the operation was successful - * @throws NullPointerException if {@code disposable} is null + * @throws NullPointerException if {@code disposable} is {@code null} */ @Override public boolean delete(@NonNull Disposable disposable) { diff --git a/src/main/java/io/reactivex/rxjava3/exceptions/UndeliverableException.java b/src/main/java/io/reactivex/rxjava3/exceptions/UndeliverableException.java index 6c84920e74..08f926e9b3 100644 --- a/src/main/java/io/reactivex/rxjava3/exceptions/UndeliverableException.java +++ b/src/main/java/io/reactivex/rxjava3/exceptions/UndeliverableException.java @@ -14,7 +14,7 @@ package io.reactivex.rxjava3.exceptions; /** - * Wrapper for Throwable errors that are sent to `RxJavaPlugins.onError`. + * Wrapper for Throwable errors that are sent to {@link io.reactivex.rxjava3.plugins.RxJavaPlugins#onError(Throwable) RxJavaPlugins.onError}. * History: 2.0.6 - experimental; 2.1 - beta * @since 2.2 */ diff --git a/src/main/java/io/reactivex/rxjava3/internal/util/BackpressureHelper.java b/src/main/java/io/reactivex/rxjava3/internal/util/BackpressureHelper.java index 9237c11a18..4ccfe67845 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/util/BackpressureHelper.java +++ b/src/main/java/io/reactivex/rxjava3/internal/util/BackpressureHelper.java @@ -14,6 +14,7 @@ import java.util.concurrent.atomic.AtomicLong; +import io.reactivex.rxjava3.annotations.NonNull; import io.reactivex.rxjava3.plugins.RxJavaPlugins; /** @@ -26,10 +27,10 @@ private BackpressureHelper() { } /** - * Adds two long values and caps the sum at Long.MAX_VALUE. + * Adds two long values and caps the sum at {@link Long#MAX_VALUE}. * @param a the first value * @param b the second value - * @return the sum capped at Long.MAX_VALUE + * @return the sum capped at {@link Long#MAX_VALUE} */ public static long addCap(long a, long b) { long u = a + b; @@ -40,10 +41,10 @@ public static long addCap(long a, long b) { } /** - * Multiplies two long values and caps the product at Long.MAX_VALUE. + * Multiplies two long values and caps the product at {@link Long#MAX_VALUE}. * @param a the first value * @param b the second value - * @return the product capped at Long.MAX_VALUE + * @return the product capped at {@link Long#MAX_VALUE} */ public static long multiplyCap(long a, long b) { long u = a * b; @@ -56,13 +57,13 @@ public static long multiplyCap(long a, long b) { } /** - * Atomically adds the positive value n to the requested value in the AtomicLong and - * caps the result at Long.MAX_VALUE and returns the previous value. - * @param requested the AtomicLong holding the current requested value + * Atomically adds the positive value n to the requested value in the {@link AtomicLong} and + * caps the result at {@link Long#MAX_VALUE} and returns the previous value. + * @param requested the {@code AtomicLong} holding the current requested value * @param n the value to add, must be positive (not verified) * @return the original value before the add */ - public static long add(AtomicLong requested, long n) { + public static long add(@NonNull AtomicLong requested, long n) { for (;;) { long r = requested.get(); if (r == Long.MAX_VALUE) { @@ -76,14 +77,14 @@ public static long add(AtomicLong requested, long n) { } /** - * Atomically adds the positive value n to the requested value in the AtomicLong and - * caps the result at Long.MAX_VALUE and returns the previous value and - * considers Long.MIN_VALUE as a cancel indication (no addition then). - * @param requested the AtomicLong holding the current requested value + * Atomically adds the positive value n to the requested value in the {@link AtomicLong} and + * caps the result at {@link Long#MAX_VALUE} and returns the previous value and + * considers {@link Long#MIN_VALUE} as a cancel indication (no addition then). + * @param requested the {@code AtomicLong} holding the current requested value * @param n the value to add, must be positive (not verified) * @return the original value before the add */ - public static long addCancel(AtomicLong requested, long n) { + public static long addCancel(@NonNull AtomicLong requested, long n) { for (;;) { long r = requested.get(); if (r == Long.MIN_VALUE) { @@ -100,12 +101,12 @@ public static long addCancel(AtomicLong requested, long n) { } /** - * Atomically subtract the given number (positive, not validated) from the target field unless it contains Long.MAX_VALUE. + * Atomically subtract the given number (positive, not validated) from the target field unless it contains {@link Long#MAX_VALUE}. * @param requested the target field holding the current requested amount * @param n the produced element count, positive (not validated) * @return the new amount */ - public static long produced(AtomicLong requested, long n) { + public static long produced(@NonNull AtomicLong requested, long n) { for (;;) { long current = requested.get(); if (current == Long.MAX_VALUE) { @@ -124,12 +125,12 @@ public static long produced(AtomicLong requested, long n) { /** * Atomically subtract the given number (positive, not validated) from the target field if - * it doesn't contain Long.MIN_VALUE (indicating some cancelled state) or Long.MAX_VALUE (unbounded mode). + * it doesn't contain {@link Long#MIN_VALUE} (indicating some cancelled state) or {@link Long#MAX_VALUE} (unbounded mode). * @param requested the target field holding the current requested amount * @param n the produced element count, positive (not validated) * @return the new amount */ - public static long producedCancel(AtomicLong requested, long n) { + public static long producedCancel(@NonNull AtomicLong requested, long n) { for (;;) { long current = requested.get(); if (current == Long.MIN_VALUE) { diff --git a/src/main/java/io/reactivex/rxjava3/internal/util/QueueDrainHelper.java b/src/main/java/io/reactivex/rxjava3/internal/util/QueueDrainHelper.java index 41906b9cf7..9cdfab07ce 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/util/QueueDrainHelper.java +++ b/src/main/java/io/reactivex/rxjava3/internal/util/QueueDrainHelper.java @@ -218,7 +218,7 @@ public static
SimpleQueue createQueue(int capacityHint) { } /** - * Requests Long.MAX_VALUE if prefetch is negative or the exact + * Requests {@link Long#MAX_VALUE} if prefetch is negative or the exact * amount if prefetch is positive. * @param s the Subscription to request from * @param prefetch the prefetch value @@ -383,7 +383,7 @@ static boolean postCompleteDrain(long n, * in completed mode, requests no-longer reach the upstream but help in draining the queue. * * The algorithm utilizes the most significant bit (bit 63) of a long value (AtomicLong) since - * request amount only goes up to Long.MAX_VALUE (bits 0-62) and negative values aren't + * request amount only goes up to {@link Long#MAX_VALUE} (bits 0-62) and negative values aren't * allowed. * * @param
the value type emitted diff --git a/src/main/java/io/reactivex/rxjava3/observers/BaseTestConsumer.java b/src/main/java/io/reactivex/rxjava3/observers/BaseTestConsumer.java index f8a0356a75..458325c280 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/BaseTestConsumer.java +++ b/src/main/java/io/reactivex/rxjava3/observers/BaseTestConsumer.java @@ -16,15 +16,17 @@ import java.util.*; import java.util.concurrent.*; +import io.reactivex.rxjava3.annotations.*; import io.reactivex.rxjava3.exceptions.CompositeException; import io.reactivex.rxjava3.functions.Predicate; -import io.reactivex.rxjava3.internal.functions.*; +import io.reactivex.rxjava3.internal.functions.Functions; import io.reactivex.rxjava3.internal.util.*; /** - * Base class with shared infrastructure to support TestSubscriber and TestObserver. + * Base class with shared infrastructure to support + * {@link io.reactivex.rxjava3.subscribers.TestSubscriber TestSubscriber} and {@link TestObserver}. * @param the value type consumed - * @param the subclass of this BaseTestConsumer + * @param the subclass of this {@code BaseTestConsumer} */ public abstract class BaseTestConsumer (); - this.errors = new VolatileSizeArrayList (); + this.values = new VolatileSizeArrayList<>(); + this.errors = new VolatileSizeArrayList<>(); this.done = new CountDownLatch(1); } /** - * Returns a shared list of received onNext values. + * Returns a shared list of received {@code onNext} values or the single {@code onSuccess} value. * * Note that accessing the items via certain methods of the {@link List} * interface while the upstream is still actively emitting * more items may result in a {@code ConcurrentModificationException}. *
* The {@link List#size()} method will return the number of items - * already received by this TestObserver/TestSubscriber in a thread-safe + * already received by this {@code TestObserver}/{@code TestSubscriber} in a thread-safe * manner that can be read via {@link List#get(int)}) method * (index range of 0 to {@code List.size() - 1}). *
@@ -76,6 +78,7 @@ public BaseTestConsumer() { * {@code ConcurrentModificationException}. * @return a list of received onNext values */ + @NonNull public final List
values() { return values; } @@ -89,7 +92,8 @@ public final List values() { * @param message the message to use * @return AssertionError the prepared AssertionError instance */ - protected final AssertionError fail(String message) { + @NonNull + protected final AssertionError fail(@NonNull String message) { StringBuilder b = new StringBuilder(64 + message.length()); b.append(message); @@ -131,11 +135,12 @@ protected final AssertionError fail(String message) { } /** - * Awaits until this TestObserver/TestSubscriber receives an onError or onComplete events. + * Awaits until this {@code TestObserver}/{@code TestSubscriber} receives an {@code onError} or {@code onComplete} events. * @return this * @throws InterruptedException if the current thread is interrupted while waiting */ @SuppressWarnings("unchecked") + @NonNull public final U await() throws InterruptedException { if (done.getCount() == 0) { return (U)this; @@ -146,14 +151,14 @@ public final U await() throws InterruptedException { } /** - * Awaits the specified amount of time or until this TestObserver/TestSubscriber - * receives an onError or onComplete events, whichever happens first. + * Awaits the specified amount of time or until this {@code TestObserver}/{@code TestSubscriber} + * receives an {@code onError} or {@code onComplete} events, whichever happens first. * @param time the waiting time * @param unit the time unit of the waiting time - * @return true if the TestObserver/TestSubscriber terminated, false if timeout happened + * @return true if the {@code TestObserver}/{@code TestSubscriber} terminated, false if timeout happened * @throws InterruptedException if the current thread is interrupted while waiting */ - public final boolean await(long time, TimeUnit unit) throws InterruptedException { + public final boolean await(long time, @NonNull TimeUnit unit) throws InterruptedException { boolean d = done.getCount() == 0 || (done.await(time, unit)); timeout = !d; return d; @@ -162,10 +167,11 @@ public final boolean await(long time, TimeUnit unit) throws InterruptedException // assertion methods /** - * Assert that this TestObserver/TestSubscriber received exactly one onComplete event. + * Assert that this {@code TestObserver}/{@code TestSubscriber} received exactly one {@code onComplete} event. * @return this */ @SuppressWarnings("unchecked") + @NonNull public final U assertComplete() { long c = completions; if (c == 0) { @@ -178,10 +184,11 @@ public final U assertComplete() { } /** - * Assert that this TestObserver/TestSubscriber has not received any onComplete event. + * Assert that this {@code TestObserver}/{@code TestSubscriber} has not received an {@code onComplete} event. * @return this */ @SuppressWarnings("unchecked") + @NonNull public final U assertNotComplete() { long c = completions; if (c == 1) { @@ -194,10 +201,11 @@ public final U assertNotComplete() { } /** - * Assert that this TestObserver/TestSubscriber has not received any onError event. + * Assert that this {@code TestObserver}/{@code TestSubscriber} has not received an {@code onError} event. * @return this */ @SuppressWarnings("unchecked") + @NonNull public final U assertNoErrors() { int s = errors.size(); if (s != 0) { @@ -207,9 +215,9 @@ public final U assertNoErrors() { } /** - * Assert that this TestObserver/TestSubscriber received exactly the specified onError event value. + * Assert that this {@code TestObserver}/{@code TestSubscriber} received exactly the specified {@code onError} event value. * - * The comparison is performed via Objects.equals(); since most exceptions don't + *
The comparison is performed via {@link Objects#equals(Object, Object)}; since most exceptions don't * implement equals(), this assertion may fail. Use the {@link #assertError(Class)} * overload to test against the class of an error instead of an instance of an error * or {@link #assertError(Predicate)} to test with different condition. @@ -218,31 +226,34 @@ public final U assertNoErrors() { * @see #assertError(Class) * @see #assertError(Predicate) */ - public final U assertError(Throwable error) { + @NonNull + public final U assertError(@NonNull Throwable error) { return assertError(Functions.equalsWith(error)); } /** - * Asserts that this TestObserver/TestSubscriber received exactly one onError event which is an - * instance of the specified errorClass class. - * @param errorClass the error class to expect + * Asserts that this {@code TestObserver}/{@code TestSubscriber} received exactly one {@code onError} event which is an + * instance of the specified {@code errorClass} {@link Class}. + * @param errorClass the error {@code Class} to expect * @return this */ @SuppressWarnings({ "unchecked", "rawtypes" }) - public final U assertError(Class errorClass) { + @NonNull + public final U assertError(@NonNull Class errorClass) { return (U)assertError((Predicate)Functions.isInstanceOf(errorClass)); } /** - * Asserts that this TestObserver/TestSubscriber received exactly one onError event for which - * the provided predicate returns true. + * Asserts that this {@code TestObserver}/{@code TestSubscriber} received exactly one {@code onError} event for which + * the provided predicate returns {@code true}. * @param errorPredicate - * the predicate that receives the error Throwable - * and should return true for expected errors. + * the predicate that receives the error {@link Throwable} + * and should return {@code true} for expected errors. * @return this */ @SuppressWarnings("unchecked") - public final U assertError(Predicate
errorPredicate) { + @NonNull + public final U assertError(@NonNull Predicate errorPredicate) { int s = errors.size(); if (s == 0) { throw fail("No errors"); @@ -272,13 +283,14 @@ public final U assertError(Predicate errorPredicate) { } /** - * Assert that this TestObserver/TestSubscriber received exactly one onNext value which is equal to - * the given value with respect to Objects.equals. + * Assert that this {@code TestObserver}/{@code TestSubscriber} received exactly one {@code onNext} value which is equal to + * the given value with respect to {@link Objects#equals(Object, Object)}. * @param value the value to expect * @return this */ @SuppressWarnings("unchecked") - public final U assertValue(T value) { + @NonNull + public final U assertValue(@NonNull T value) { int s = values.size(); if (s != 1) { throw fail("expected: " + valueAndClass(value) + " but was: " + values); @@ -291,15 +303,16 @@ public final U assertValue(T value) { } /** - * Asserts that this TestObserver/TestSubscriber received exactly one onNext value for which - * the provided predicate returns true. + * Asserts that this {@code TestObserver}/{@code TestSubscriber} received exactly one {@code onNext} value for which + * the provided predicate returns {@code true}. * @param valuePredicate - * the predicate that receives the onNext value - * and should return true for the expected value. + * the predicate that receives the {@code onNext} value + * and should return {@code true} for the expected value. * @return this */ @SuppressWarnings("unchecked") - public final U assertValue(Predicate valuePredicate) { + @NonNull + public final U assertValue(@NonNull Predicate valuePredicate) { assertValueAt(0, valuePredicate); if (values.size() > 1) { @@ -310,8 +323,8 @@ public final U assertValue(Predicate valuePredicate) { } /** - * Asserts that this TestObserver/TestSubscriber received an onNext value at the given index - * which is equal to the given value with respect to null-safe Object.equals. + * Asserts that this {@code TestObserver}/{@code TestSubscriber} received an {@code onNext} value at the given index + * which is equal to the given value with respect to {@code null}-safe {@link Objects#equals(Object, Object)}. * History: 2.1.3 - experimental * @param index the position to assert on * @param value the value to expect @@ -319,7 +332,8 @@ public final U assertValue(Predicate
valuePredicate) { * @since 2.2 */ @SuppressWarnings("unchecked") - public final U assertValueAt(int index, T value) { + @NonNull + public final U assertValueAt(int index, @NonNull T value) { int s = values.size(); if (s == 0) { throw fail("No values"); @@ -337,16 +351,17 @@ public final U assertValueAt(int index, T value) { } /** - * Asserts that this TestObserver/TestSubscriber received an onNext value at the given index - * for the provided predicate returns true. + * Asserts that this {@code TestObserver}/{@code TestSubscriber} received an {@code onNext} value at the given index + * for the provided predicate returns {@code true}. * @param index the position to assert on * @param valuePredicate - * the predicate that receives the onNext value - * and should return true for the expected value. + * the predicate that receives the {@code onNext} value + * and should return {@code true} for the expected value. * @return this */ @SuppressWarnings("unchecked") - public final U assertValueAt(int index, Predicate valuePredicate) { + @NonNull + public final U assertValueAt(int index, @NonNull Predicate valuePredicate) { int s = values.size(); if (s == 0) { throw fail("No values"); @@ -373,11 +388,12 @@ public final U assertValueAt(int index, Predicate valuePredicate) { } /** - * Appends the class name to a non-null value. + * Appends the class name to a non-{@code null} value or returns {@code "null"}. * @param o the object * @return the string representation */ - public static String valueAndClass(Object o) { + @NonNull + public static String valueAndClass(@Nullable Object o) { if (o != null) { return o + " (class: " + o.getClass().getSimpleName() + ")"; } @@ -385,11 +401,12 @@ public static String valueAndClass(Object o) { } /** - * Assert that this TestObserver/TestSubscriber received the specified number onNext events. - * @param count the expected number of onNext events + * Assert that this {@code TestObserver}/{@code TestSubscriber} received the specified number {@code onNext} events. + * @param count the expected number of {@code onNext} events * @return this */ @SuppressWarnings("unchecked") + @NonNull public final U assertValueCount(int count) { int s = values.size(); if (s != count) { @@ -399,20 +416,23 @@ public final U assertValueCount(int count) { } /** - * Assert that this TestObserver/TestSubscriber has not received any onNext events. + * Assert that this {@code TestObserver}/{@code TestSubscriber} has not received any {@code onNext} events. * @return this */ + @NonNull public final U assertNoValues() { return assertValueCount(0); } /** - * Assert that the TestObserver/TestSubscriber received only the specified values in the specified order. + * Assert that the {@code TestObserver}/{@code TestSubscriber} received only the specified values in the specified order. * @param values the values expected * @return this */ @SuppressWarnings("unchecked") - public final U assertValues(T... values) { + @SafeVarargs + @NonNull + public final U assertValues(@NonNull T... values) { int s = this.values.size(); if (s != values.length) { throw fail("Value count differs; expected: " + values.length + " " + Arrays.toString(values) @@ -429,14 +449,15 @@ public final U assertValues(T... values) { } /** - * Assert that the TestObserver/TestSubscriber received only the specified values in the specified order without terminating. + * Assert that the {@code TestObserver}/{@code TestSubscriber} received only the specified values in the specified order without terminating. * History: 2.1.4 - experimental * @param values the values expected * @return this * @since 2.2 */ @SafeVarargs - public final U assertValuesOnly(T... values) { + @NonNull + public final U assertValuesOnly(@NonNull T... values) { return assertSubscribed() .assertValues(values) .assertNoErrors() @@ -444,12 +465,13 @@ public final U assertValuesOnly(T... values) { } /** - * Assert that the TestObserver/TestSubscriber received only the specified sequence of values in the same order. + * Assert that the {@code TestObserver}/{@code TestSubscriber} received only the specified sequence of values in the same order. * @param sequence the sequence of expected values in order * @return this */ @SuppressWarnings("unchecked") - public final U assertValueSequence(Iterable sequence) { + @NonNull + public final U assertValueSequence(@NonNull Iterable sequence) { int i = 0; Iterator
actualIterator = values.iterator(); Iterator expectedIterator = sequence.iterator(); @@ -482,20 +504,22 @@ public final U assertValueSequence(Iterable sequence) { } /** - * Assert that the onSubscribe method was called exactly once. + * Assert that the {@code onSubscribe} method was called exactly once. * @return this */ + @NonNull protected abstract U assertSubscribed(); /** - * Assert that the upstream signalled the specified values in order and + * Assert that the upstream signaled the specified values in order and * completed normally. * @param values the expected values, asserted in order * @return this * @see #assertFailure(Class, Object...) */ @SafeVarargs - public final U assertResult(T... values) { + @NonNull + public final U assertResult(@NonNull T... values) { return assertSubscribed() .assertValues(values) .assertNoErrors() @@ -503,14 +527,15 @@ public final U assertResult(T... values) { } /** - * Assert that the upstream signalled the specified values in order - * and then failed with a specific class or subclass of Throwable. - * @param error the expected exception (parent) class + * Assert that the upstream signaled the specified values in order + * and then failed with a specific class or subclass of {@link Throwable}. + * @param error the expected exception (parent) {@link Class} * @param values the expected values, asserted in order * @return this */ @SafeVarargs - public final U assertFailure(Class error, T... values) { + @NonNull + public final U assertFailure(@NonNull Class error, @NonNull T... values) { return assertSubscribed() .assertValues(values) .assertError(error) @@ -519,14 +544,15 @@ public final U assertFailure(Class error, T... values) { /** * Awaits until the internal latch is counted down. - * If the wait times out or gets interrupted, the TestObserver/TestSubscriber is cancelled. + *
If the wait times out or gets interrupted, the {@code TestObserver}/{@code TestSubscriber} is cancelled. * @param time the waiting time * @param unit the time unit of the waiting time * @return this - * @throws RuntimeException wrapping an InterruptedException if the wait is interrupted + * @throws RuntimeException wrapping an {@link InterruptedException} if the wait is interrupted */ @SuppressWarnings("unchecked") - public final U awaitDone(long time, TimeUnit unit) { + @NonNull + public final U awaitDone(long time, @NonNull TimeUnit unit) { try { if (!done.await(time, unit)) { timeout = true; @@ -540,9 +566,12 @@ public final U awaitDone(long time, TimeUnit unit) { } /** - * Assert that the TestObserver/TestSubscriber has received a Disposable but no other events. + * Assert that the {@code TestObserver}/{@code TestSubscriber} has received a + * {@link io.reactivex.rxjava3.disposables.Disposable Disposable}/{@link org.reactivestreams.Subscription Subscription} + * via {@code onSubscribe} but no other events. * @return this */ + @NonNull public final U assertEmpty() { return assertSubscribed() .assertNoValues() @@ -554,18 +583,19 @@ public final U assertEmpty() { * Set the tag displayed along with an assertion failure's * other state information. *
History: 2.0.7 - experimental - * @param tag the string to display (null won't print any tag) + * @param tag the string to display ({@code null} won't print any tag) * @return this * @since 2.1 */ @SuppressWarnings("unchecked") - public final U withTag(CharSequence tag) { + @NonNull + public final U withTag(@Nullable CharSequence tag) { this.tag = tag; return (U)this; } /** - * Await until the TestObserver/TestSubscriber receives the given + * Await until the {@code TestObserver}/{@code TestSubscriber} receives the given * number of items or terminates by sleeping 10 milliseconds at a time * up to 5000 milliseconds of timeout. *
History: 2.0.7 - experimental @@ -574,6 +604,7 @@ public final U withTag(CharSequence tag) { * @since 2.1 */ @SuppressWarnings("unchecked") + @NonNull public final U awaitCount(int atLeast) { long start = System.currentTimeMillis(); long timeoutMillis = 5000; diff --git a/src/main/java/io/reactivex/rxjava3/observers/DisposableCompletableObserver.java b/src/main/java/io/reactivex/rxjava3/observers/DisposableCompletableObserver.java index 25926663d5..940654b622 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/DisposableCompletableObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/DisposableCompletableObserver.java @@ -53,7 +53,7 @@ */ public abstract class DisposableCompletableObserver implements CompletableObserver, Disposable { - final AtomicReference
upstream = new AtomicReference (); + final AtomicReference upstream = new AtomicReference<>(); @Override public final void onSubscribe(@NonNull Disposable d) { @@ -63,7 +63,7 @@ public final void onSubscribe(@NonNull Disposable d) { } /** - * Called once the single upstream Disposable is set via onSubscribe. + * Called once the single upstream {@link Disposable} is set via {@link #onSubscribe(Disposable)}. */ protected void onStart() { } diff --git a/src/main/java/io/reactivex/rxjava3/observers/DisposableMaybeObserver.java b/src/main/java/io/reactivex/rxjava3/observers/DisposableMaybeObserver.java index 66ac6a8620..092994e818 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/DisposableMaybeObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/DisposableMaybeObserver.java @@ -22,7 +22,7 @@ import io.reactivex.rxjava3.internal.util.EndConsumerHelper; /** - * An abstract {@link MaybeObserver} that allows asynchronous cancellation by implementing Disposable. + * An abstract {@link MaybeObserver} that allows asynchronous cancellation by implementing {@link Disposable}. * * All pre-implemented final methods are thread-safe. * @@ -62,7 +62,7 @@ */ public abstract class DisposableMaybeObserver
implements MaybeObserver , Disposable { - final AtomicReference upstream = new AtomicReference (); + final AtomicReference upstream = new AtomicReference<>(); @Override public final void onSubscribe(@NonNull Disposable d) { @@ -72,7 +72,7 @@ public final void onSubscribe(@NonNull Disposable d) { } /** - * Called once the single upstream Disposable is set via onSubscribe. + * Called once the single upstream {@link Disposable} is set via {@link #onSubscribe(Disposable)}. */ protected void onStart() { } diff --git a/src/main/java/io/reactivex/rxjava3/observers/DisposableObserver.java b/src/main/java/io/reactivex/rxjava3/observers/DisposableObserver.java index a15587317c..ff2eba7758 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/DisposableObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/DisposableObserver.java @@ -22,7 +22,7 @@ import io.reactivex.rxjava3.internal.util.EndConsumerHelper; /** - * An abstract {@link Observer} that allows asynchronous cancellation by implementing Disposable. + * An abstract {@link Observer} that allows asynchronous cancellation by implementing {@link Disposable}. * * All pre-implemented final methods are thread-safe. * @@ -66,7 +66,7 @@ */ public abstract class DisposableObserver
implements Observer , Disposable { - final AtomicReference upstream = new AtomicReference (); + final AtomicReference upstream = new AtomicReference<>(); @Override public final void onSubscribe(@NonNull Disposable d) { diff --git a/src/main/java/io/reactivex/rxjava3/observers/DisposableSingleObserver.java b/src/main/java/io/reactivex/rxjava3/observers/DisposableSingleObserver.java index 6d8ce73fb0..56e5c543b7 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/DisposableSingleObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/DisposableSingleObserver.java @@ -22,7 +22,7 @@ import io.reactivex.rxjava3.internal.util.EndConsumerHelper; /** - * An abstract {@link SingleObserver} that allows asynchronous cancellation by implementing Disposable. + * An abstract {@link SingleObserver} that allows asynchronous cancellation by implementing {@link Disposable}. * * All pre-implemented final methods are thread-safe. * @@ -55,7 +55,7 @@ */ public abstract class DisposableSingleObserver
implements SingleObserver , Disposable { - final AtomicReference upstream = new AtomicReference (); + final AtomicReference upstream = new AtomicReference<>(); @Override public final void onSubscribe(@NonNull Disposable d) { @@ -65,7 +65,7 @@ public final void onSubscribe(@NonNull Disposable d) { } /** - * Called once the single upstream Disposable is set via onSubscribe. + * Called once the single upstream {@link Disposable} is set via {@link #onSubscribe(Disposable)}. */ protected void onStart() { } diff --git a/src/main/java/io/reactivex/rxjava3/observers/LambdaConsumerIntrospection.java b/src/main/java/io/reactivex/rxjava3/observers/LambdaConsumerIntrospection.java index 5d74ac5e8c..e3f5c46cfe 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/LambdaConsumerIntrospection.java +++ b/src/main/java/io/reactivex/rxjava3/observers/LambdaConsumerIntrospection.java @@ -24,8 +24,8 @@ public interface LambdaConsumerIntrospection { /** - * Returns true or false if a custom onError consumer has been provided. - * @return {@code true} if a custom onError consumer implementation was supplied. Returns {@code false} if the + * Returns {@code true} or {@code false} if a custom {@code onError} consumer has been provided. + * @return {@code true} if a custom {@code onError} consumer implementation was supplied. Returns {@code false} if the * implementation is missing an error consumer and thus using a throwing default implementation. */ boolean hasCustomOnError(); diff --git a/src/main/java/io/reactivex/rxjava3/observers/ResourceCompletableObserver.java b/src/main/java/io/reactivex/rxjava3/observers/ResourceCompletableObserver.java index b57d77f6f5..df05fe0169 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/ResourceCompletableObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/ResourceCompletableObserver.java @@ -74,17 +74,17 @@ */ public abstract class ResourceCompletableObserver implements CompletableObserver, Disposable { /** The active subscription. */ - private final AtomicReference upstream = new AtomicReference (); + private final AtomicReference upstream = new AtomicReference<>(); /** The resource composite, can never be null. */ private final ListCompositeDisposable resources = new ListCompositeDisposable(); /** - * Adds a resource to this ResourceObserver. + * Adds a resource to this {@code ResourceCompletableObserver}. * * @param resource the resource to add * - * @throws NullPointerException if resource is null + * @throws NullPointerException if resource is {@code null} */ public final void add(@NonNull Disposable resource) { Objects.requireNonNull(resource, "resource is null"); @@ -99,7 +99,7 @@ public final void onSubscribe(@NonNull Disposable d) { } /** - * Called once the upstream sets a Subscription on this ResourceObserver. + * Called once the upstream sets a {@link Disposable} on this {@code ResourceCompletableObserver}. * * You can perform initialization at this moment. The default * implementation does nothing. @@ -109,10 +109,10 @@ protected void onStart() { /** * Cancels the main disposable (if any) and disposes the resources associated with - * this ResourceObserver (if any). + * this {@code ResourceCompletableObserver} (if any). * - *
This method can be called before the upstream calls onSubscribe at which - * case the main Disposable will be immediately disposed. + *
This method can be called before the upstream calls {@link #onSubscribe(Disposable)} at which + * case the main {@link Disposable} will be immediately disposed. */ @Override public final void dispose() { @@ -122,8 +122,8 @@ public final void dispose() { } /** - * Returns true if this ResourceObserver has been disposed/cancelled. - * @return true if this ResourceObserver has been disposed/cancelled + * Returns true if this {@code ResourceCompletableObserver} has been disposed/cancelled. + * @return true if this {@code ResourceCompletableObserver} has been disposed/cancelled */ @Override public final boolean isDisposed() { diff --git a/src/main/java/io/reactivex/rxjava3/observers/ResourceMaybeObserver.java b/src/main/java/io/reactivex/rxjava3/observers/ResourceMaybeObserver.java index cb42a70fbd..22345e60ce 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/ResourceMaybeObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/ResourceMaybeObserver.java @@ -84,17 +84,17 @@ */ public abstract class ResourceMaybeObserver
implements MaybeObserver , Disposable { /** The active subscription. */ - private final AtomicReference upstream = new AtomicReference (); + private final AtomicReference upstream = new AtomicReference<>(); /** The resource composite, can never be null. */ private final ListCompositeDisposable resources = new ListCompositeDisposable(); /** - * Adds a resource to this ResourceObserver. + * Adds a resource to this {@code ResourceMaybeObserver}. * * @param resource the resource to add * - * @throws NullPointerException if resource is null + * @throws NullPointerException if resource is {@code null} */ public final void add(@NonNull Disposable resource) { Objects.requireNonNull(resource, "resource is null"); @@ -109,7 +109,7 @@ public final void onSubscribe(@NonNull Disposable d) { } /** - * Called once the upstream sets a Subscription on this ResourceObserver. + * Called once the upstream sets a {@link Disposable} on this {@code ResourceMaybeObserver}. * * You can perform initialization at this moment. The default * implementation does nothing. @@ -119,10 +119,10 @@ protected void onStart() { /** * Cancels the main disposable (if any) and disposes the resources associated with - * this ResourceObserver (if any). + * this {@code ResourceMaybeObserver} (if any). * - *
This method can be called before the upstream calls onSubscribe at which - * case the main Disposable will be immediately disposed. + *
This method can be called before the upstream calls {@link #onSubscribe(Disposable)} at which + * case the main {@link Disposable} will be immediately disposed. */ @Override public final void dispose() { @@ -132,8 +132,8 @@ public final void dispose() { } /** - * Returns true if this ResourceObserver has been disposed/cancelled. - * @return true if this ResourceObserver has been disposed/cancelled + * Returns true if this {@code ResourceMaybeObserver} has been disposed/cancelled. + * @return true if this {@code ResourceMaybeObserver} has been disposed/cancelled */ @Override public final boolean isDisposed() { diff --git a/src/main/java/io/reactivex/rxjava3/observers/ResourceObserver.java b/src/main/java/io/reactivex/rxjava3/observers/ResourceObserver.java index b6d77a0d5f..f0810f4c5e 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/ResourceObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/ResourceObserver.java @@ -82,17 +82,17 @@ */ public abstract class ResourceObserver
implements Observer , Disposable { /** The active subscription. */ - private final AtomicReference upstream = new AtomicReference (); + private final AtomicReference upstream = new AtomicReference<>(); /** The resource composite, can never be null. */ private final ListCompositeDisposable resources = new ListCompositeDisposable(); /** - * Adds a resource to this ResourceObserver. + * Adds a resource to this {@code ResourceObserver}. * * @param resource the resource to add * - * @throws NullPointerException if resource is null + * @throws NullPointerException if resource is {@code null} */ public final void add(@NonNull Disposable resource) { Objects.requireNonNull(resource, "resource is null"); @@ -107,7 +107,7 @@ public final void onSubscribe(Disposable d) { } /** - * Called once the upstream sets a Subscription on this ResourceObserver. + * Called once the upstream sets a {@link Disposable} on this {@code ResourceObserver}. * * You can perform initialization at this moment. The default * implementation does nothing. @@ -117,10 +117,10 @@ protected void onStart() { /** * Cancels the main disposable (if any) and disposes the resources associated with - * this ResourceObserver (if any). + * this {@code ResourceObserver} (if any). * - *
This method can be called before the upstream calls onSubscribe at which - * case the main Disposable will be immediately disposed. + *
This method can be called before the upstream calls {@link #onSubscribe(Disposable)} at which + * case the main {@link Disposable} will be immediately disposed. */ @Override public final void dispose() { @@ -130,8 +130,8 @@ public final void dispose() { } /** - * Returns true if this ResourceObserver has been disposed/cancelled. - * @return true if this ResourceObserver has been disposed/cancelled + * Returns true if this {@code ResourceObserver} has been disposed/cancelled. + * @return true if this {@code ResourceObserver} has been disposed/cancelled */ @Override public final boolean isDisposed() { diff --git a/src/main/java/io/reactivex/rxjava3/observers/ResourceSingleObserver.java b/src/main/java/io/reactivex/rxjava3/observers/ResourceSingleObserver.java index af723bf789..efebfaa85c 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/ResourceSingleObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/ResourceSingleObserver.java @@ -77,17 +77,17 @@ */ public abstract class ResourceSingleObserver
implements SingleObserver , Disposable { /** The active subscription. */ - private final AtomicReference upstream = new AtomicReference (); + private final AtomicReference upstream = new AtomicReference<>(); /** The resource composite, can never be null. */ private final ListCompositeDisposable resources = new ListCompositeDisposable(); /** - * Adds a resource to this ResourceObserver. + * Adds a resource to this {@code ResourceSingleObserver}. * * @param resource the resource to add * - * @throws NullPointerException if resource is null + * @throws NullPointerException if resource is {@code null} */ public final void add(@NonNull Disposable resource) { Objects.requireNonNull(resource, "resource is null"); @@ -102,7 +102,7 @@ public final void onSubscribe(@NonNull Disposable d) { } /** - * Called once the upstream sets a Subscription on this ResourceObserver. + * Called once the upstream sets a {@link Disposable} on this {@code ResourceSingleObserver}. * * You can perform initialization at this moment. The default * implementation does nothing. @@ -112,10 +112,10 @@ protected void onStart() { /** * Cancels the main disposable (if any) and disposes the resources associated with - * this ResourceObserver (if any). + * this {@code ResourceSingleObserver} (if any). * - *
This method can be called before the upstream calls onSubscribe at which - * case the main Disposable will be immediately disposed. + *
This method can be called before the upstream calls {@link #onSubscribe(Disposable)} at which + * case the main {@link Disposable} will be immediately disposed. */ @Override public final void dispose() { @@ -125,8 +125,8 @@ public final void dispose() { } /** - * Returns true if this ResourceObserver has been disposed/cancelled. - * @return true if this ResourceObserver has been disposed/cancelled + * Returns true if this {@code ResourceSingleObserver} has been disposed/cancelled. + * @return true if this {@code ResourceSingleObserver} has been disposed/cancelled */ @Override public final boolean isDisposed() { diff --git a/src/main/java/io/reactivex/rxjava3/observers/SafeObserver.java b/src/main/java/io/reactivex/rxjava3/observers/SafeObserver.java index f7612378e2..c3935e79dc 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/SafeObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/SafeObserver.java @@ -21,7 +21,7 @@ import io.reactivex.rxjava3.plugins.RxJavaPlugins; /** - * Wraps another Subscriber and ensures all onXXX methods conform the protocol + * Wraps another {@link Observer} and ensures all {@code onXXX} methods conform the protocol * (except the requirement for serialized access). * * @param
the value type @@ -35,8 +35,8 @@ public final class SafeObserver implements Observer , Disposable { boolean done; /** - * Constructs a SafeObserver by wrapping the given actual Observer. - * @param downstream the actual Observer to wrap, not null (not validated) + * Constructs a {@code SafeObserver} by wrapping the given actual {@link Observer}. + * @param downstream the actual {@code Observer} to wrap, not {@code null} (not validated) */ public SafeObserver(@NonNull Observer downstream) { this.downstream = downstream; diff --git a/src/main/java/io/reactivex/rxjava3/observers/SerializedObserver.java b/src/main/java/io/reactivex/rxjava3/observers/SerializedObserver.java index a0ffe75cce..e0d32fbd7a 100644 --- a/src/main/java/io/reactivex/rxjava3/observers/SerializedObserver.java +++ b/src/main/java/io/reactivex/rxjava3/observers/SerializedObserver.java @@ -20,13 +20,14 @@ import io.reactivex.rxjava3.plugins.RxJavaPlugins; /** - * Serializes access to the onNext, onError and onComplete methods of another Observer. + * Serializes access to the {@link Observer#onNext(Object)}, {@link Observer#onError(Throwable)} and + * {@link Observer#onComplete()} methods of another {@link Observer}. * * Note that {@link #onSubscribe(Disposable)} is not serialized in respect of the other methods so - * make sure the {@code onSubscribe()} is called with a non-null {@code Disposable} + * make sure the {@code onSubscribe()} is called with a non-null {@link Disposable} * before any of the other methods are called. * - *
The implementation assumes that the actual Observer's methods don't throw. + *
The implementation assumes that the actual {@code Observer}'s methods don't throw. * * @param
the value type */ @@ -44,19 +45,19 @@ public final class SerializedObserver implements Observer , Disposable { volatile boolean done; /** - * Construct a SerializedObserver by wrapping the given actual Observer. - * @param downstream the actual Observer, not null (not verified) + * Construct a {@code SerializedObserver} by wrapping the given actual {@link Observer}. + * @param downstream the actual {@code Observer}, not {@code null} (not verified) */ public SerializedObserver(@NonNull Observer downstream) { this(downstream, false); } /** - * Construct a SerializedObserver by wrapping the given actual Observer and + * Construct a SerializedObserver by wrapping the given actual {@link Observer} and * optionally delaying the errors till all regular values have been emitted * from the internal buffer. - * @param actual the actual Observer, not null (not verified) - * @param delayError if true, errors are emitted after regular values have been emitted + * @param actual the actual {@code Observer}, not {@code null} (not verified) + * @param delayError if {@code true}, errors are emitted after regular values have been emitted */ public SerializedObserver(@NonNull Observer actual, boolean delayError) { this.downstream = actual; @@ -100,7 +101,7 @@ public void onNext(@NonNull T t) { if (emitting) { AppendOnlyLinkedArrayList
+ * {@code Publisher} is consumed in an unbounded fashion (requesting {@link Long#MAX_VALUE}).
* - Scheduler: *
- {@code BehaviorProcessor} does not operate by default on a particular {@link io.reactivex.rxjava3.core.Scheduler} and * the {@code Subscriber}s get notified on the thread the respective {@code onXXX} methods were invoked. @@ -229,7 +229,7 @@ public static
- Backpressure: *
- The processor does not coordinate backpressure for its subscribers and implements a weaker {@code onSubscribe} which - * calls requests Long.MAX_VALUE from the incoming Subscriptions. This makes it possible to subscribe the {@code PublishProcessor} + * calls requests {@link Long#MAX_VALUE} from the incoming Subscriptions. This makes it possible to subscribe the {@code PublishProcessor} * to multiple sources (note on serialization though) unlike the standard {@code Subscriber} contract. Child subscribers, however, are not overflown but receive an * {@link IllegalStateException} in case their requested amount is zero. *
- Scheduler: diff --git a/src/main/java/io/reactivex/rxjava3/processors/ReplayProcessor.java b/src/main/java/io/reactivex/rxjava3/processors/ReplayProcessor.java index fcd2cb913f..04de41a77f 100644 --- a/src/main/java/io/reactivex/rxjava3/processors/ReplayProcessor.java +++ b/src/main/java/io/reactivex/rxjava3/processors/ReplayProcessor.java @@ -66,7 +66,7 @@ *
- Backpressure: *
- This {@code ReplayProcessor} respects the individual backpressure behavior of its {@code Subscriber}s but * does not coordinate their request amounts towards the upstream (because there might not be any) and - * consumes the upstream in an unbounded manner (requesting {@code Long.MAX_VALUE}). + * consumes the upstream in an unbounded manner (requesting {@link Long#MAX_VALUE}). * Note that {@code Subscriber}s receive a continuous sequence of values after they subscribed even * if an individual item gets delayed due to backpressure. *
- Scheduler: diff --git a/src/main/java/io/reactivex/rxjava3/processors/UnicastProcessor.java b/src/main/java/io/reactivex/rxjava3/processors/UnicastProcessor.java index 4a1862a95f..ee93a5ac9c 100644 --- a/src/main/java/io/reactivex/rxjava3/processors/UnicastProcessor.java +++ b/src/main/java/io/reactivex/rxjava3/processors/UnicastProcessor.java @@ -67,7 +67,7 @@ * {@link NullPointerException} being thrown and the processor's state is not changed. *
- Backpressure: *
- {@code UnicastProcessor} honors the downstream backpressure but consumes an upstream source - * (if any) in an unbounded manner (requesting {@code Long.MAX_VALUE}). + * (if any) in an unbounded manner (requesting {@link Long#MAX_VALUE}). *
- Scheduler: *
- {@code UnicastProcessor} does not operate by default on a particular {@link io.reactivex.rxjava3.core.Scheduler} and * the single {@code Subscriber} gets notified on the thread the respective {@code onXXX} methods were invoked. diff --git a/src/main/java/io/reactivex/rxjava3/schedulers/Timed.java b/src/main/java/io/reactivex/rxjava3/schedulers/Timed.java index 845273ff22..814b223dcf 100644 --- a/src/main/java/io/reactivex/rxjava3/schedulers/Timed.java +++ b/src/main/java/io/reactivex/rxjava3/schedulers/Timed.java @@ -29,11 +29,11 @@ public final class Timed
* -
*
* -
*
* -
*
* -
*
* -
*
* -
*
* -
*
-
*
-
*
-
*
-
*
-
*
-
*
* Note also that ignoring groups or subscribing later (i.e., on another thread) will result in
* so-called group abandonment where a group will only contain one element and the group will be
@@ -10473,7 +10473,7 @@ public final This operator allows preventing the downstream to trigger unbounded mode via {@code request(Long.MAX_VALUE)}
+ * This operator allows preventing the downstream to trigger unbounded mode via {@code request(}{@link Long#MAX_VALUE}{@code )}
* or compensate for the per-item overhead of small and frequent requests.
*
* History: 2.1.7 - experimental
* @param This method is thread-safe.
+ * @param d the disposable, {@code null} is allowed
*/
void setDisposable(@Nullable Disposable d);
/**
- * Sets a Cancellable on this emitter; any previous {@link Disposable}
- * or {@link Cancellable} will be disposed/cancelled.
- * @param c the cancellable resource, null is allowed
+ * Sets a {@link Cancellable} on this emitter; any previous {@link Disposable}
+ * or {@code Cancellable} will be disposed/cancelled.
+ * This method is thread-safe.
+ * @param c the {@code Cancellable} resource, {@code null} is allowed
*/
void setCancellable(@Nullable Cancellable c);
@@ -81,19 +83,19 @@ public interface FlowableEmitter
- * Unlike {@link #onError(Throwable)}, the {@code RxJavaPlugins.onError} is not called
- * if the error could not be delivered.
+ * Unlike {@link #onError(Throwable)}, the {@link io.reactivex.rxjava3.plugins.RxJavaPlugins#onError(Throwable) RxjavaPlugins.onError}
+ * is not called if the error could not be delivered.
* History: 2.1.1 - experimental
* @param t the throwable error to signal if possible
* @return true if successful, false if the downstream is not able to accept further
diff --git a/src/main/java/io/reactivex/rxjava3/core/FlowableOnSubscribe.java b/src/main/java/io/reactivex/rxjava3/core/FlowableOnSubscribe.java
index 2bbc8b57a3..156b9f2dc7 100644
--- a/src/main/java/io/reactivex/rxjava3/core/FlowableOnSubscribe.java
+++ b/src/main/java/io/reactivex/rxjava3/core/FlowableOnSubscribe.java
@@ -25,8 +25,8 @@
public interface FlowableOnSubscribe History: 2.0.7 - experimental; 2.1 - beta
diff --git a/src/main/java/io/reactivex/rxjava3/core/FlowableTransformer.java b/src/main/java/io/reactivex/rxjava3/core/FlowableTransformer.java
index 877ad1916a..3b98a97dad 100644
--- a/src/main/java/io/reactivex/rxjava3/core/FlowableTransformer.java
+++ b/src/main/java/io/reactivex/rxjava3/core/FlowableTransformer.java
@@ -18,7 +18,7 @@
import io.reactivex.rxjava3.annotations.NonNull;
/**
- * Interface to compose Flowables.
+ * Interface to compose {@link Flowable}s.
*
* @param History: 2.1.7 - experimental
* @param This method is thread-safe.
+ * @param d the disposable, {@code null} is allowed
*/
void setDisposable(@Nullable Disposable d);
/**
- * Sets a Cancellable on this emitter; any previous {@link Disposable}
- * or {@link Cancellable} will be disposed/cancelled.
- * @param c the cancellable resource, null is allowed
+ * Sets a {@link Cancellable} on this emitter; any previous {@link Disposable}
+ * or {@code Cancellable} will be disposed/cancelled.
+ * This method is thread-safe.
+ * @param c the {@code Cancellable} resource, {@code null} is allowed
*/
void setCancellable(@Nullable Cancellable c);
@@ -91,14 +93,14 @@ public interface MaybeEmitter
- * Unlike {@link #onError(Throwable)}, the {@code RxJavaPlugins.onError} is not called
- * if the error could not be delivered.
+ * Unlike {@link #onError(Throwable)}, the {@link io.reactivex.rxjava3.plugins.RxJavaPlugins#onError(Throwable) RxjavaPlugins.onError}
+ * is not called if the error could not be delivered.
* History: 2.1.1 - experimental
- * @param t the throwable error to signal if possible
+ * @param t the {@code Throwable} error to signal if possible
* @return true if successful, false if the downstream is not able to accept further
* events
* @since 2.2
diff --git a/src/main/java/io/reactivex/rxjava3/core/MaybeObserver.java b/src/main/java/io/reactivex/rxjava3/core/MaybeObserver.java
index 7b7eb1a0d4..52bee5923a 100644
--- a/src/main/java/io/reactivex/rxjava3/core/MaybeObserver.java
+++ b/src/main/java/io/reactivex/rxjava3/core/MaybeObserver.java
@@ -56,32 +56,32 @@
public interface MaybeObserver
* The {@link Maybe} will not call this method if it calls {@link #onError}.
*
* @param t
- * the item emitted by the Maybe
+ * the item emitted by the {@code Maybe}
*/
void onSuccess(@NonNull T t);
/**
- * Notifies the MaybeObserver that the {@link Maybe} has experienced an error condition.
+ * Notifies the {@link MaybeObserver} that the {@link Maybe} has experienced an error condition.
*
* If the {@link Maybe} calls this method, it will not thereafter call {@link #onSuccess}.
*
* @param e
- * the exception encountered by the Maybe
+ * the exception encountered by the {@code Maybe}
*/
void onError(@NonNull Throwable e);
diff --git a/src/main/java/io/reactivex/rxjava3/core/MaybeOnSubscribe.java b/src/main/java/io/reactivex/rxjava3/core/MaybeOnSubscribe.java
index 2dde118757..350d17af5d 100644
--- a/src/main/java/io/reactivex/rxjava3/core/MaybeOnSubscribe.java
+++ b/src/main/java/io/reactivex/rxjava3/core/MaybeOnSubscribe.java
@@ -25,8 +25,8 @@
public interface MaybeOnSubscribe History: 2.1.7 - experimental
* @param This method is thread-safe.
+ * @param d the {@code Disposable}, {@code null} is allowed
*/
void setDisposable(@Nullable Disposable d);
/**
- * Sets a Cancellable on this emitter; any previous {@link Disposable}
- * or {@link Cancellable} will be disposed/cancelled.
- * @param c the cancellable resource, null is allowed
+ * Sets a {@link Cancellable} on this emitter; any previous {@link Disposable}
+ * or {@code Cancellable} will be disposed/cancelled.
+ * This method is thread-safe.
+ * @param c the {@code Cancellable} resource, {@code null} is allowed
*/
void setCancellable(@Nullable Cancellable c);
@@ -73,21 +75,21 @@ public interface ObservableEmitter
- * Unlike {@link #onError(Throwable)}, the {@code RxJavaPlugins.onError} is not called
- * if the error could not be delivered.
+ * Unlike {@link #onError(Throwable)}, the {@link io.reactivex.rxjava3.plugins.RxJavaPlugins#onError(Throwable) RxjavaPlugins.onError}
+ * is not called if the error could not be delivered.
* History: 2.1.1 - experimental
- * @param t the throwable error to signal if possible
+ * @param t the {@code Throwable} error to signal if possible
* @return true if successful, false if the downstream is not able to accept further
* events
* @since 2.2
diff --git a/src/main/java/io/reactivex/rxjava3/core/ObservableOnSubscribe.java b/src/main/java/io/reactivex/rxjava3/core/ObservableOnSubscribe.java
index 43d6252849..bcde496462 100644
--- a/src/main/java/io/reactivex/rxjava3/core/ObservableOnSubscribe.java
+++ b/src/main/java/io/reactivex/rxjava3/core/ObservableOnSubscribe.java
@@ -25,8 +25,8 @@
public interface ObservableOnSubscribe
* The {@link Observable} may call this method 0 or more times.
*
@@ -99,9 +99,9 @@ public interface Observer
- * If the {@link Observable} calls this method, it will not thereafter call {@link #onNext} or
+ * If the {@code Observable} calls this method, it will not thereafter call {@link #onNext} or
* {@link #onComplete}.
*
* @param e
@@ -110,9 +110,9 @@ public interface Observer
- * The {@link Observable} will not call this method if it calls {@link #onError}.
+ * The {@code Observable} will not call this method if it calls {@link #onError}.
*/
void onComplete();
diff --git a/src/main/java/io/reactivex/rxjava3/core/Single.java b/src/main/java/io/reactivex/rxjava3/core/Single.java
index 27fad6d53e..612d57a25e 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Single.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Single.java
@@ -2346,7 +2346,7 @@ public final Single If the delaying source signals an error, that error is re-emitted and no subscription
* to the current Single happens.
- * The other source is consumed in an unbounded manner (requesting Long.MAX_VALUE from it).
+ * The other source is consumed in an unbounded manner (requesting {@link Long#MAX_VALUE} from it).
*
* This {@code ReplayProcessor} respects the individual backpressure behavior of its {@code Subscriber}s but
* does not coordinate their request amounts towards the upstream (because there might not be any) and
- * consumes the upstream in an unbounded manner (requesting {@code Long.MAX_VALUE}).
+ * consumes the upstream in an unbounded manner (requesting {@link Long#MAX_VALUE}).
* Note that {@code Subscriber}s receive a continuous sequence of values after they subscribed even
* if an individual item gets delayed due to backpressure.
* Due to concurrency requirements, a size-bounded {@code ReplayProcessor} may hold strong references to more source
@@ -104,7 +104,7 @@
*
* Since a {@code UnicastProcessor} is a {@link io.reactivex.rxjava3.core.Flowable} as well as a {@link FlowableProcessor}, it
- * honors the downstream backpressure but consumes an upstream source in an unbounded manner (requesting {@code Long.MAX_VALUE}).
+ * honors the downstream backpressure but consumes an upstream source in an unbounded manner (requesting {@link Long#MAX_VALUE}).
*
* When this {@code UnicastProcessor} is terminated via {@link #onError(Throwable)} the current or late single {@code Subscriber}
* may receive the {@code Throwable} before any available items could be emitted. To make sure an {@code onError} event is delivered
@@ -91,7 +91,7 @@
* All pre-implemented final methods are thread-safe.
*
- * The default {@link #onStart()} requests Long.MAX_VALUE by default. Override
+ * The default {@link #onStart()} requests {@link Long#MAX_VALUE} by default. Override
* the method to request a custom positive amount.
*
* Note that calling {@link #request(long)} from {@link #onStart()} may trigger
@@ -85,7 +85,7 @@ public final void onSubscribe(Subscription s) {
}
/**
- * Requests from the upstream Subscription.
+ * Requests from the upstream {@link Subscription}.
* @param n the request amount, positive
*/
protected final void request(long n) {
@@ -96,7 +96,7 @@ protected final void request(long n) {
}
/**
- * Cancels the upstream's Subscription.
+ * Cancels the upstream's {@link Subscription}.
*/
protected final void cancel() {
Subscription s = this.upstream;
diff --git a/src/main/java/io/reactivex/rxjava3/subscribers/DisposableSubscriber.java b/src/main/java/io/reactivex/rxjava3/subscribers/DisposableSubscriber.java
index 2c6c126f8d..566b0097f7 100644
--- a/src/main/java/io/reactivex/rxjava3/subscribers/DisposableSubscriber.java
+++ b/src/main/java/io/reactivex/rxjava3/subscribers/DisposableSubscriber.java
@@ -23,11 +23,11 @@
import io.reactivex.rxjava3.internal.util.EndConsumerHelper;
/**
- * An abstract Subscriber that allows asynchronous, external cancellation by implementing Disposable.
+ * An abstract Subscriber that allows asynchronous, external cancellation by implementing {@link Disposable}.
*
* All pre-implemented final methods are thread-safe.
*
- * The default {@link #onStart()} requests Long.MAX_VALUE by default. Override
+ * The default {@link #onStart()} requests {@link Long#MAX_VALUE} by default. Override
* the method to request a custom positive amount. Use the protected {@link #request(long)}
* to request more items and {@link #cancel()} to cancel the sequence from within an
* {@code onNext} implementation.
@@ -74,7 +74,7 @@
* @param 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.
+ * Note that calling this method before a {@link Subscription} is set via {@link #onSubscribe(Subscription)}
+ * leads to {@link NullPointerException} and meant to be called from inside {@link #onStart()} or
+ * {@link #onNext(Object)}.
* @param n the request amount, positive
*/
protected final void request(long n) {
@@ -103,8 +103,8 @@ protected final void request(long n) {
}
/**
- * Cancels the Subscription set via onSubscribe or makes sure a
- * Subscription set asynchronously (later) is cancelled immediately.
+ * Cancels the Subscription set via {@link #onSubscribe(Subscription)} or makes sure a
+ * {@link Subscription} set asynchronously (later) is cancelled immediately.
* This method is thread-safe and can be exposed as a public API.
*/
protected final void cancel() {
diff --git a/src/main/java/io/reactivex/rxjava3/subscribers/ResourceSubscriber.java b/src/main/java/io/reactivex/rxjava3/subscribers/ResourceSubscriber.java
index 1990939311..7d9411048a 100644
--- a/src/main/java/io/reactivex/rxjava3/subscribers/ResourceSubscriber.java
+++ b/src/main/java/io/reactivex/rxjava3/subscribers/ResourceSubscriber.java
@@ -40,7 +40,7 @@
* {@code ResourceSubscriber} and then add/remove resources to/from the {@code CompositeDisposable}
* freely.
*
- * The default {@link #onStart()} requests Long.MAX_VALUE by default. Override
+ * The default {@link #onStart()} requests {@link Long#MAX_VALUE} by default. Override
* the method to request a custom positive amount. Use the protected {@link #request(long)}
* to request more items and {@link #dispose()} to cancel the sequence from within an
* {@code onNext} implementation.
@@ -94,7 +94,7 @@
*/
public abstract class ResourceSubscriber You can perform initialization at this moment. The default
- * implementation requests Long.MAX_VALUE from upstream.
+ * implementation requests {@link Long#MAX_VALUE} from upstream.
*/
protected void onStart() {
request(Long.MAX_VALUE);
@@ -138,7 +138,7 @@ protected void onStart() {
/**
* Request the specified amount of elements from upstream.
*
- * This method can be called before the upstream calls onSubscribe().
+ * This method can be called before the upstream calls {@link #onSubscribe(Subscription)}.
* When the subscription happens, all missed requests are requested.
*
* @param n the request amount, must be positive
@@ -149,10 +149,10 @@ protected final void request(long n) {
/**
* Cancels the subscription (if any) and disposes the resources associated with
- * this AsyncObserver (if any).
+ * this {@code ResourceSubscriber} (if any).
*
- * This method can be called before the upstream calls onSubscribe at which
- * case the Subscription will be immediately cancelled.
+ * This method can be called before the upstream calls {@link #onSubscribe(Subscription)} at which
+ * case the {@link Subscription} will be immediately cancelled.
*/
@Override
public final void dispose() {
@@ -162,8 +162,8 @@ public final void dispose() {
}
/**
- * Returns true if this AsyncObserver has been disposed/cancelled.
- * @return true if this AsyncObserver has been disposed/cancelled
+ * Returns true if this {@code ResourceSubscriber} has been disposed/cancelled.
+ * @return true if this {@code ResourceSubscriber} has been disposed/cancelled
*/
@Override
public final boolean isDisposed() {
diff --git a/src/main/java/io/reactivex/rxjava3/subscribers/SafeSubscriber.java b/src/main/java/io/reactivex/rxjava3/subscribers/SafeSubscriber.java
index 096a99e623..0db19796fa 100644
--- a/src/main/java/io/reactivex/rxjava3/subscribers/SafeSubscriber.java
+++ b/src/main/java/io/reactivex/rxjava3/subscribers/SafeSubscriber.java
@@ -14,6 +14,7 @@
import org.reactivestreams.*;
+import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.FlowableSubscriber;
import io.reactivex.rxjava3.exceptions.*;
import io.reactivex.rxjava3.internal.subscriptions.*;
@@ -21,7 +22,7 @@
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
/**
- * Wraps another Subscriber and ensures all onXXX methods conform the protocol
+ * Wraps another {@link Subscriber} and ensures all {@code onXXX} methods conform the protocol
* (except the requirement for serialized access).
*
* @param Note that {@link #onSubscribe(Subscription)} is not serialized in respect of the other methods so
- * make sure the {@code onSubscribe} is called with a non-null {@code Subscription}
+ * make sure the {@code onSubscribe} is called with a non-{@code null} {@link Subscription}
* before any of the other methods are called.
*
- * The implementation assumes that the actual Subscriber's methods don't throw.
+ * The implementation assumes that the actual {@code Subscriber}'s methods don't throw.
*
* @param
@@ -17224,7 +17224,7 @@ public final
*
@@ -18552,8 +18552,8 @@ public final
*
* @param 
*
*
*
*