[](https://codecov.io/gh/ReactiveX/RxJava/branch/3.x)
[](https://maven-badges.herokuapp.com/maven-central/io.reactivex.rxjava3/rxjava)
+[](https://gitpod.io/#https://github.com/ReactiveX/RxJava)
RxJava is a Java VM implementation of [Reactive Extensions](http://reactivex.io): a library for composing asynchronous and event-based programs by using observable sequences.
diff --git a/build.gradle b/build.gradle
index 2339c0df09..69d41f1dfd 100644
--- a/build.gradle
+++ b/build.gradle
@@ -4,10 +4,10 @@ plugins {
id("eclipse")
id("jacoco")
id("maven-publish")
- id("ru.vyarus.animalsniffer") version "1.5.4"
+ id("ru.vyarus.animalsniffer") version "1.6.0"
id("me.champeau.gradle.jmh") version "0.5.3"
id("com.github.hierynomus.license") version "0.16.1"
- id("biz.aQute.bnd.builder") version "6.2.0"
+ id("biz.aQute.bnd.builder") version "6.4.0"
id("com.vanniktech.maven.publish") version "0.19.0"
id("org.beryx.jar") version "1.2.0"
}
@@ -16,7 +16,7 @@ ext {
reactiveStreamsVersion = "1.0.4"
junitVersion = "4.13.2"
testNgVersion = "7.5"
- mockitoVersion = "4.6.0"
+ mockitoVersion = "4.11.0"
jmhLibVersion = "1.21"
guavaVersion = "31.1-jre"
}
diff --git a/gradle/javadoc_cleanup.gradle b/gradle/javadoc_cleanup.gradle
index 4c3386a6ef..63b4f7f045 100644
--- a/gradle/javadoc_cleanup.gradle
+++ b/gradle/javadoc_cleanup.gradle
@@ -26,6 +26,10 @@ task javadocCleanup(dependsOn: "javadoc") doLast {
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava3/parallel/ParallelFlowable.html'))
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava3/disposables/Disposable.html'))
+
+ fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava3/observers/TestObserver.html'))
+ fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava3/observers/BaseTestConsumer.html'))
+ fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava3/subscribers/TestSubscriber.html'))
}
def fixJavadocFile(file) {
diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
index 41d9927a4d..943f0cbfa7 100644
Binary files a/gradle/wrapper/gradle-wrapper.jar and b/gradle/wrapper/gradle-wrapper.jar differ
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index 00e33edef6..f398c33c4b 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,6 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.6-bin.zip
+networkTimeout=10000
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
index 1b6c787337..65dcd68d65 100755
--- a/gradlew
+++ b/gradlew
@@ -55,7 +55,7 @@
# Darwin, MinGW, and NonStop.
#
# (3) This script is generated from the Groovy template
-# https://github.com/gradle/gradle/blob/master/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+# https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
# within the Gradle project.
#
# You can find Gradle at https://github.com/gradle/gradle/.
@@ -80,10 +80,10 @@ do
esac
done
-APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
-
-APP_NAME="Gradle"
+# This is normally unused
+# shellcheck disable=SC2034
APP_BASE_NAME=${0##*/}
+APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
@@ -143,12 +143,16 @@ fi
if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
case $MAX_FD in #(
max*)
+ # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked.
+ # shellcheck disable=SC3045
MAX_FD=$( ulimit -H -n ) ||
warn "Could not query maximum file descriptor limit"
esac
case $MAX_FD in #(
'' | soft) :;; #(
*)
+ # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked.
+ # shellcheck disable=SC3045
ulimit -n "$MAX_FD" ||
warn "Could not set maximum file descriptor limit to $MAX_FD"
esac
@@ -205,6 +209,12 @@ set -- \
org.gradle.wrapper.GradleWrapperMain \
"$@"
+# Stop when "xargs" is not available.
+if ! command -v xargs >/dev/null 2>&1
+then
+ die "xargs is not available"
+fi
+
# Use "xargs" to parse quoted args.
#
# With -n1 it outputs one arg per line, with the quotes and backslashes removed.
diff --git a/gradlew.bat b/gradlew.bat
index 107acd32c4..93e3f59f13 100644
--- a/gradlew.bat
+++ b/gradlew.bat
@@ -14,7 +14,7 @@
@rem limitations under the License.
@rem
-@if "%DEBUG%" == "" @echo off
+@if "%DEBUG%"=="" @echo off
@rem ##########################################################################
@rem
@rem Gradle startup script for Windows
@@ -25,7 +25,8 @@
if "%OS%"=="Windows_NT" setlocal
set DIRNAME=%~dp0
-if "%DIRNAME%" == "" set DIRNAME=.
+if "%DIRNAME%"=="" set DIRNAME=.
+@rem This is normally unused
set APP_BASE_NAME=%~n0
set APP_HOME=%DIRNAME%
@@ -40,7 +41,7 @@ if defined JAVA_HOME goto findJavaFromJavaHome
set JAVA_EXE=java.exe
%JAVA_EXE% -version >NUL 2>&1
-if "%ERRORLEVEL%" == "0" goto execute
+if %ERRORLEVEL% equ 0 goto execute
echo.
echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
@@ -75,13 +76,15 @@ set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
:end
@rem End local scope for the variables with windows NT shell
-if "%ERRORLEVEL%"=="0" goto mainEnd
+if %ERRORLEVEL% equ 0 goto mainEnd
:fail
rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
rem the _cmd.exe /c_ return code!
-if not "" == "%GRADLE_EXIT_CONSOLE%" exit 1
-exit /b 1
+set EXIT_CODE=%ERRORLEVEL%
+if %EXIT_CODE% equ 0 set EXIT_CODE=1
+if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE%
+exit /b %EXIT_CODE%
:mainEnd
if "%OS%"=="Windows_NT" endlocal
diff --git a/src/main/java/io/reactivex/rxjava3/core/Completable.java b/src/main/java/io/reactivex/rxjava3/core/Completable.java
index d16dd4dc85..e31a44c32c 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Completable.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Completable.java
@@ -601,6 +601,7 @@ public static Completable fromCallable(@NonNull Callable callable) {
* @param future the {@code Future} to react to
* @return the new {@code Completable} instance
* @throws NullPointerException if {@code future} is {@code null}
+ * @see #fromCompletionStage(CompletionStage)
*/
@CheckReturnValue
@NonNull
@@ -3410,7 +3411,7 @@ public final TestObserver
* * Note that the operator takes an already instantiated, running or terminated {@code CompletionStage}. - * If the optional is to be created per consumer upon subscription, use {@link #defer(Supplier)} + * If the {@code CompletionStage} is to be created per consumer upon subscription, use {@link #defer(Supplier)} * around {@code fromCompletionStage}: *
* Maybe.defer(() -> Completable.fromCompletionStage(createCompletionStage()));
diff --git a/src/main/java/io/reactivex/rxjava3/core/Flowable.java b/src/main/java/io/reactivex/rxjava3/core/Flowable.java
index b06164d1ae..d9cea6954f 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Flowable.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Flowable.java
@@ -8930,7 +8930,59 @@ public final Flowable debounce(long timeout, @NonNull TimeUnit unit) {
public final Flowable debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler));
+ return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler, null));
+ }
+
+ /**
+ * Returns a {@code Flowable} that mirrors the current {@code Flowable}, except that it drops items emitted by the
+ * current {@code Flowable} that are followed by newer items before a timeout value expires on a specified
+ * {@link Scheduler}. The timer resets on each emission.
+ *
+ * Note: If items keep being emitted by the current {@code Flowable} faster than the timeout then no items
+ * will be emitted by the resulting {@code Flowable}.
+ *
+ * 
+ *
+ * Delivery of the item after the grace period happens on the given {@code Scheduler}'s
+ * {@code Worker} which if takes too long, a newer item may arrive from the upstream, causing the
+ * {@code Worker}'s task to get disposed, which may also interrupt any downstream blocking operation
+ * (yielding an {@code InterruptedException}). It is recommended processing items
+ * that may take long time to be moved to another thread via {@link #observeOn} applied after
+ * {@code debounce} itself.
+ *
+ * - Backpressure:
+ * - This operator does not support backpressure as it uses time to control data flow.
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param timeout
+ * the time each item has to be "the most recent" of those emitted by the current {@code Flowable} to
+ * ensure that it's not dropped
+ * @param unit
+ * the unit of time for the specified {@code timeout}
+ * @param scheduler
+ * the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
+ * item
+ * @param onDropped
+ * called with the current entry when it has been replaced by a new one
+ * @return the new {@code Flowable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Debounce
+ * @see RxJava wiki: Backpressure
+ * @see #throttleWithTimeout(long, TimeUnit, Scheduler, Consumer)
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @NonNull
+ @BackpressureSupport(BackpressureKind.ERROR)
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @Experimental
+ public final Flowable debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ Objects.requireNonNull(onDropped, "onDropped is null");
+ return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler, onDropped));
}
/**
@@ -14672,7 +14724,7 @@ public final Flowable sample(long period, @NonNull TimeUnit unit, boolean emi
public final Flowable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new FlowableSampleTimed<>(this, period, unit, scheduler, false));
+ return RxJavaPlugins.onAssembly(new FlowableSampleTimed<>(this, period, unit, scheduler, false, null));
}
/**
@@ -14713,7 +14765,51 @@ public final Flowable sample(long period, @NonNull TimeUnit unit, @NonNull Sc
public final Flowable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new FlowableSampleTimed<>(this, period, unit, scheduler, emitLast));
+ return RxJavaPlugins.onAssembly(new FlowableSampleTimed<>(this, period, unit, scheduler, emitLast, null));
+ }
+
+ /**
+ * Returns a {@code Flowable} that emits the most recently emitted item (if any) emitted by the current {@code Flowable}
+ * within periodic time intervals, where the intervals are defined on a particular {@link Scheduler}
+ * and optionally emit the very last upstream item when the upstream completes.
+ *
+ * 
+ *
+ * - Backpressure:
+ * - This operator does not support backpressure as it uses time to control data flow.
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param period
+ * the sampling rate
+ * @param unit
+ * the {@link TimeUnit} in which {@code period} is defined
+ * @param scheduler
+ * the {@code Scheduler} to use when sampling
+ * @param emitLast
+ * if {@code true} and the upstream completes while there is still an unsampled item available,
+ * that item is emitted to downstream before completion
+ * if {@code false}, an unsampled last item is ignored.
+ * @param onDropped
+ * called with the current entry when it has been replaced by a new one
+ * @return the new {@code Flowable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Sample
+ * @see RxJava wiki: Backpressure
+ * @see #throttleLast(long, TimeUnit, Scheduler)
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @NonNull
+ @BackpressureSupport(BackpressureKind.ERROR)
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @Experimental
+ public final Flowable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast, @NonNull Consumer onDropped) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ Objects.requireNonNull(onDropped, "onDropped is null");
+ return RxJavaPlugins.onAssembly(new FlowableSampleTimed<>(this, period, unit, scheduler, emitLast, onDropped));
}
/**
@@ -15816,7 +15912,7 @@ public final Disposable subscribe(@NonNull Consumer onNext, @NonNull
* terminates or this particular {@code Disposable} is disposed, the {@code Subscriber} is removed
* from the given container.
*
- * The {@coded Subscriber} will be removed after the callback for the terminal event has been invoked.
+ * The {@code Subscriber} will be removed after the callback for the terminal event has been invoked.
*
* - Backpressure:
* - The operator consumes the current {@code Flowable} in an unbounded manner (i.e., no
@@ -17096,7 +17192,50 @@ public final Flowable
throttleFirst(long windowDuration, @NonNull TimeUnit un
public final Flowable throttleFirst(long skipDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new FlowableThrottleFirstTimed<>(this, skipDuration, unit, scheduler));
+ return RxJavaPlugins.onAssembly(new FlowableThrottleFirstTimed<>(this, skipDuration, unit, scheduler, null));
+ }
+
+ /**
+ * Returns a {@code Flowable} that emits only the first item emitted by the current {@code Flowable} during sequential
+ * time windows of a specified duration, where the windows are managed by a specified {@link Scheduler}.
+ *
+ * This differs from {@link #throttleLast} in that this only tracks the passage of time whereas
+ * {@link #throttleLast} ticks at scheduled intervals.
+ *
+ * 
+ *
+ * - Backpressure:
+ * - This operator does not support backpressure as it uses time to control data flow.
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param skipDuration
+ * time to wait before emitting another item after emitting the last item
+ * @param unit
+ * the unit of time of {@code skipDuration}
+ * @param scheduler
+ * the {@code Scheduler} to use internally to manage the timers that handle timeout for each
+ * event
+ * @param onDropped
+ * called when an item doesn't get delivered to the downstream
+ *
+ * @return the new {@code Flowable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Sample
+ * @see RxJava wiki: Backpressure
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @NonNull
+ @BackpressureSupport(BackpressureKind.ERROR)
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @Experimental
+ public final Flowable throttleFirst(long skipDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ Objects.requireNonNull(onDropped, "onDropped is null");
+ return RxJavaPlugins.onAssembly(new FlowableThrottleFirstTimed<>(this, skipDuration, unit, scheduler, onDropped));
}
/**
@@ -17170,6 +17309,47 @@ public final Flowable throttleLast(long intervalDuration, @NonNull TimeUnit u
return sample(intervalDuration, unit, scheduler);
}
+ /**
+ * Returns a {@code Flowable} that emits only the last item emitted by the current {@code Flowable} during sequential
+ * time windows of a specified duration, where the duration is governed by a specified {@link Scheduler}.
+ *
+ * This differs from {@link #throttleFirst(long, TimeUnit, Scheduler)} in that this ticks along at a scheduled interval whereas
+ * {@code throttleFirst} does not tick, it just tracks the passage of time.
+ *
+ * 
+ *
+ * - Backpressure:
+ * - This operator does not support backpressure as it uses time to control data flow.
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param intervalDuration
+ * duration of windows within which the last item emitted by the current {@code Flowable} will be
+ * emitted
+ * @param unit
+ * the unit of time of {@code intervalDuration}
+ * @param scheduler
+ * the {@code Scheduler} to use internally to manage the timers that handle timeout for each
+ * event
+ * @param onDropped
+ * called with the current entry when it has been replaced by a new one
+ * @return the new {@code Flowable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Sample
+ * @see RxJava wiki: Backpressure
+ * @see #sample(long, TimeUnit, Scheduler)
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @BackpressureSupport(BackpressureKind.ERROR)
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @NonNull
+ @Experimental
+ public final Flowable throttleLast(long intervalDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) {
+ return sample(intervalDuration, unit, scheduler, false, onDropped);
+ }
+
/**
* Throttles items from the upstream {@code Flowable} by first emitting the next
* item from upstream, then periodically emitting the latest item (if any) when
@@ -17327,7 +17507,61 @@ public final Flowable throttleLatest(long timeout, @NonNull TimeUnit unit, @N
public final Flowable throttleLatest(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new FlowableThrottleLatest<>(this, timeout, unit, scheduler, emitLast));
+ return RxJavaPlugins.onAssembly(new FlowableThrottleLatest<>(this, timeout, unit, scheduler, emitLast, null));
+ }
+
+ /**
+ * Throttles items from the upstream {@code Flowable} by first emitting the next
+ * item from upstream, then periodically emitting the latest item (if any) when
+ * the specified timeout elapses between them, invoking the consumer for any dropped item.
+ *
+ * 
+ *
+ * If no items were emitted from the upstream during this timeout phase, the next
+ * upstream item is emitted immediately and the timeout window starts from then.
+ *
+ * - Backpressure:
+ * - This operator does not support backpressure as it uses time to control data flow.
+ * If the downstream is not ready to receive items, a
+ * {@link io.reactivex.rxjava3.exceptions.MissingBackpressureException MissingBackpressureException}
+ * will be signaled.
+ * - Scheduler:
+ * - You specify which {@link Scheduler} this operator will use.
+ * - Error handling:
+ * -
+ * If the upstream signals an {@code onError} or {@code onDropped} callback crashes,
+ * the error is delivered immediately to the downstream. If both happen, a {@link CompositeException}
+ * is created, containing both the upstream and the callback error.
+ * If the {@code onDropped} callback crashes during cancellation, the exception is forwarded
+ * to the global error handler via {@link RxJavaPlugins#onError(Throwable)}.
+ *
+ *
+ * @param timeout the time to wait after an item emission towards the downstream
+ * before trying to emit the latest item from upstream again
+ * @param unit the time unit
+ * @param scheduler the {@code Scheduler} where the timed wait and latest item
+ * emission will be performed
+ * @param emitLast If {@code true}, the very last item from the upstream will be emitted
+ * immediately when the upstream completes, regardless if there is
+ * a timeout window active or not. If {@code false}, the very last
+ * upstream item is ignored and the flow terminates.
+ * @param onDropped called when an item is replaced by a newer item that doesn't get delivered
+ * to the downstream, including the very last item if {@code emitLast} is {@code false}
+ * and the current undelivered item when the sequence gets canceled.
+ * @return the new {@code Flowable} instance
+ * @throws NullPointerException if {@code unit}, {@code scheduler} or {@code onDropped} is {@code null}
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @NonNull
+ @BackpressureSupport(BackpressureKind.ERROR)
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @Experimental
+ public final Flowable throttleLatest(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast, @NonNull Consumer onDropped) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ Objects.requireNonNull(onDropped, "onDropped is null");
+ return RxJavaPlugins.onAssembly(new FlowableThrottleLatest<>(this, timeout, unit, scheduler, emitLast, onDropped));
}
/**
@@ -17405,6 +17639,49 @@ public final Flowable throttleWithTimeout(long timeout, @NonNull TimeUnit uni
return debounce(timeout, unit, scheduler);
}
+ /**
+ * Returns a {@code Flowable} that mirrors the current {@code Flowable}, except that it drops items emitted by the
+ * current {@code Flowable} that are followed by newer items before a timeout value expires on a specified
+ * {@link Scheduler}. The timer resets on each emission (alias to {@link #debounce(long, TimeUnit, Scheduler, Consumer)}).
+ *
+ * Note: If items keep being emitted by the current {@code Flowable} faster than the timeout then no items
+ * will be emitted by the resulting {@code Flowable}.
+ *
+ * 
+ *
+ * - Backpressure:
+ * - This operator does not support backpressure as it uses time to control data flow.
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param timeout
+ * the length of the window of time that must pass after the emission of an item from the current
+ * {@code Flowable} in which it emits no items in order for the item to be emitted by the
+ * resulting {@code Flowable}
+ * @param unit
+ * the unit of time for the specified {@code timeout}
+ * @param scheduler
+ * the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
+ * item
+ * @param onDropped
+ * called with the current entry when it has been replaced by a new one
+ * @return the new {@code Flowable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Debounce
+ * @see RxJava wiki: Backpressure
+ * @see #debounce(long, TimeUnit, Scheduler, Consumer)
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @BackpressureSupport(BackpressureKind.ERROR)
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @NonNull
+ @Experimental
+ public final Flowable throttleWithTimeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) {
+ return debounce(timeout, unit, scheduler, onDropped);
+ }
+
/**
* Returns a {@code Flowable} that emits records of the time interval between consecutive items emitted by the
* current {@code Flowable}.
@@ -19851,7 +20128,7 @@ public final TestSubscriber test(long initialRequest, boolean cancel) { // No
* 
*
* Note that the operator takes an already instantiated, running or terminated {@code CompletionStage}.
- * If the optional is to be created per consumer upon subscription, use {@link #defer(Supplier)}
+ * If the {@code CompletionStage} is to be created per consumer upon subscription, use {@link #defer(Supplier)}
* around {@code fromCompletionStage}:
*
* Flowable.defer(() -> Flowable.fromCompletionStage(createCompletionStage()));
diff --git a/src/main/java/io/reactivex/rxjava3/core/Maybe.java b/src/main/java/io/reactivex/rxjava3/core/Maybe.java
index 83bcd68144..8ae4137c83 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Maybe.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Maybe.java
@@ -110,7 +110,7 @@
* @since 2.0
* @see io.reactivex.rxjava3.observers.DisposableMaybeObserver
*/
-public abstract class Maybe implements MaybeSource {
+public abstract class Maybe<@NonNull T> implements MaybeSource {
/**
* Runs multiple {@link MaybeSource}s provided by an {@link Iterable} sequence and
@@ -1109,6 +1109,7 @@ public abstract class Maybe implements MaybeSource {
* @return the new {@code Maybe} instance
* @throws NullPointerException if {@code future} is {@code null}
* @see ReactiveX operators documentation: From
+ * @see #fromCompletionStage(CompletionStage)
*/
@CheckReturnValue
@NonNull
@@ -1147,6 +1148,7 @@ public abstract class Maybe implements MaybeSource {
* @return the new {@code Maybe} instance
* @throws NullPointerException if {@code future} or {@code unit} is {@code null}
* @see ReactiveX operators documentation: From
+ * @see #fromCompletionStage(CompletionStage)
*/
@CheckReturnValue
@NonNull
@@ -6145,7 +6147,7 @@ public final TestObserver test(boolean dispose) {
* 
*
* Note that the operator takes an already instantiated, running or terminated {@code CompletionStage}.
- * If the optional is to be created per consumer upon subscription, use {@link #defer(Supplier)}
+ * If the {@code CompletionStage} is to be created per consumer upon subscription, use {@link #defer(Supplier)}
* around {@code fromCompletionStage}:
*
* Maybe.defer(() -> Maybe.fromCompletionStage(createCompletionStage()));
diff --git a/src/main/java/io/reactivex/rxjava3/core/Observable.java b/src/main/java/io/reactivex/rxjava3/core/Observable.java
index f355bedf13..792b740104 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Observable.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Observable.java
@@ -2020,6 +2020,7 @@ public static int bufferSize() {
* @return the new {@code Observable} instance
* @throws NullPointerException if {@code future} is {@code null}
* @see ReactiveX operators documentation: From
+ * @see #fromCompletionStage(CompletionStage)
*/
@CheckReturnValue
@NonNull
@@ -2062,6 +2063,7 @@ public static int bufferSize() {
* @return the new {@code Observable} instance
* @throws NullPointerException if {@code future} or {@code unit} is {@code null}
* @see ReactiveX operators documentation: From
+ * @see #fromCompletionStage(CompletionStage)
*/
@CheckReturnValue
@NonNull
@@ -7894,7 +7896,55 @@ public final Observable debounce(long timeout, @NonNull TimeUnit unit) {
public final Observable debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new ObservableDebounceTimed<>(this, timeout, unit, scheduler));
+ return RxJavaPlugins.onAssembly(new ObservableDebounceTimed<>(this, timeout, unit, scheduler, null));
+ }
+
+ /**
+ * Returns an {@code Observable} that mirrors the current {@code Observable}, except that it drops items emitted by the
+ * current {@code Observable} that are followed by newer items before a timeout value expires on a specified
+ * {@link Scheduler}. The timer resets on each emission.
+ *
+ * Note: If items keep being emitted by the current {@code Observable} faster than the timeout then no items
+ * will be emitted by the resulting {@code Observable}.
+ *
+ *
+ *
+ * Delivery of the item after the grace period happens on the given {@code Scheduler}'s
+ * {@code Worker} which if takes too long, a newer item may arrive from the upstream, causing the
+ * {@code Worker}'s task to get disposed, which may also interrupt any downstream blocking operation
+ * (yielding an {@code InterruptedException}). It is recommended processing items
+ * that may take long time to be moved to another thread via {@link #observeOn} applied after
+ * {@code debounce} itself.
+ *
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param timeout
+ * the time each item has to be "the most recent" of those emitted by the current {@code Observable} to
+ * ensure that it's not dropped
+ * @param unit
+ * the unit of time for the specified {@code timeout}
+ * @param scheduler
+ * the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
+ * item
+ * @param onDropped
+ * called with the current entry when it has been replaced by a new one
+ * @return the new {@code Observable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} } or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Debounce
+ * @see #throttleWithTimeout(long, TimeUnit, Scheduler, Consumer)
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @NonNull
+ @Experimental
+ public final Observable debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ Objects.requireNonNull(onDropped, "onDropped is null");
+ return RxJavaPlugins.onAssembly(new ObservableDebounceTimed<>(this, timeout, unit, scheduler, onDropped));
}
/**
@@ -12128,7 +12178,7 @@ public final Observable sample(long period, @NonNull TimeUnit unit, boolean e
public final Observable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new ObservableSampleTimed<>(this, period, unit, scheduler, false));
+ return RxJavaPlugins.onAssembly(new ObservableSampleTimed<>(this, period, unit, scheduler, false, null));
}
/**
@@ -12165,7 +12215,46 @@ public final Observable sample(long period, @NonNull TimeUnit unit, @NonNull
public final Observable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new ObservableSampleTimed<>(this, period, unit, scheduler, emitLast));
+ return RxJavaPlugins.onAssembly(new ObservableSampleTimed<>(this, period, unit, scheduler, emitLast, null));
+ }
+
+ /**
+ * Returns an {@code Observable} that emits the most recently emitted item (if any) emitted by the current {@code Observable}
+ * within periodic time intervals, where the intervals are defined on a particular {@link Scheduler}.
+ *
+ * 
+ *
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param period
+ * the sampling rate
+ * @param unit
+ * the {@link TimeUnit} in which {@code period} is defined
+ * @param scheduler
+ * the {@code Scheduler} to use when sampling
+ * @param emitLast
+ * if {@code true} and the upstream completes while there is still an unsampled item available,
+ * that item is emitted to downstream before completion
+ * if {@code false}, an unsampled last item is ignored.
+ * @param onDropped
+ * called with the current entry when it has been replaced by a new one
+ * @return the new {@code Observable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Sample
+ * @see #throttleLast(long, TimeUnit, Scheduler)
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @NonNull
+ @Experimental
+ public final Observable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast, @NonNull Consumer onDropped) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ Objects.requireNonNull(onDropped, "onDropped is null");
+ return RxJavaPlugins.onAssembly(new ObservableSampleTimed<>(this, period, unit, scheduler, emitLast, onDropped));
}
/**
@@ -14163,7 +14252,46 @@ public final Observable throttleFirst(long windowDuration, @NonNull TimeUnit
public final Observable throttleFirst(long skipDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new ObservableThrottleFirstTimed<>(this, skipDuration, unit, scheduler));
+ return RxJavaPlugins.onAssembly(new ObservableThrottleFirstTimed<>(this, skipDuration, unit, scheduler, null));
+ }
+
+ /**
+ * Returns an {@code Observable} that emits only the first item emitted by the current {@code Observable} during sequential
+ * time windows of a specified duration, where the windows are managed by a specified {@link Scheduler}.
+ *
+ * This differs from {@link #throttleLast} in that this only tracks passage of time whereas
+ * {@code throttleLast} ticks at scheduled intervals.
+ *
+ *
+ *
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param skipDuration
+ * time to wait before emitting another item after emitting the last item
+ * @param unit
+ * the unit of time of {@code skipDuration}
+ * @param scheduler
+ * the {@code Scheduler} to use internally to manage the timers that handle timeout for each
+ * event
+ * @param onDropped
+ * called when an item doesn't get delivered to the downstream
+ *
+ * @return the new {@code Observable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Sample
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @NonNull
+ @Experimental
+ public final Observable throttleFirst(long skipDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ Objects.requireNonNull(onDropped, "onDropped is null");
+ return RxJavaPlugins.onAssembly(new ObservableThrottleFirstTimed<>(this, skipDuration, unit, scheduler, onDropped));
}
/**
@@ -14196,6 +14324,43 @@ public final Observable throttleLast(long intervalDuration, @NonNull TimeUnit
return sample(intervalDuration, unit);
}
+ /**
+ * Returns an {@code Observable} that emits only the last item emitted by the current {@code Observable} during sequential
+ * time windows of a specified duration, where the duration is governed by a specified {@link Scheduler}.
+ *
+ * This differs from {@link #throttleFirst} in that this ticks along at a scheduled interval whereas
+ * {@code throttleFirst} does not tick, it just tracks passage of time.
+ *
+ *
+ *
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param intervalDuration
+ * duration of windows within which the last item emitted by the current {@code Observable} will be
+ * emitted
+ * @param unit
+ * the unit of time of {@code intervalDuration}
+ * @param scheduler
+ * the {@code Scheduler} to use internally to manage the timers that handle timeout for each
+ * event
+ * @param onDropped
+ * called with the current entry when it has been replaced by a new one
+ * @return the new {@code Observable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Sample
+ * @see #sample(long, TimeUnit, Scheduler)
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @NonNull
+ @Experimental
+ public final Observable throttleLast(long intervalDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) {
+ return sample(intervalDuration, unit, scheduler, false, onDropped);
+ }
+
/**
* Returns an {@code Observable} that emits only the last item emitted by the current {@code Observable} during sequential
* time windows of a specified duration, where the duration is governed by a specified {@link Scheduler}.
@@ -14362,7 +14527,55 @@ public final Observable throttleLatest(long timeout, @NonNull TimeUnit unit,
public final Observable throttleLatest(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
- return RxJavaPlugins.onAssembly(new ObservableThrottleLatest<>(this, timeout, unit, scheduler, emitLast));
+ return RxJavaPlugins.onAssembly(new ObservableThrottleLatest<>(this, timeout, unit, scheduler, emitLast, null));
+ }
+
+ /**
+ * Throttles items from the current {@code Observable} by first emitting the next
+ * item from upstream, then periodically emitting the latest item (if any) when
+ * the specified timeout elapses between them, invoking the consumer for any dropped item.
+ *
+ *
+ *
+ * If no items were emitted from the upstream during this timeout phase, the next
+ * upstream item is emitted immediately and the timeout window starts from then.
+ *
+ * - Scheduler:
+ * - You specify which {@link Scheduler} this operator will use.
+ * - Error handling:
+ * -
+ * If the upstream signals an {@code onError} or {@code onDropped} callback crashes,
+ * the error is delivered immediately to the downstream. If both happen, a {@link CompositeException}
+ * is created, containing both the upstream and the callback error.
+ * If the {@code onDropped} callback crashes when the sequence gets disposed, the exception is forwarded
+ * to the global error handler via {@link RxJavaPlugins#onError(Throwable)}.
+ *
+ *
+ * @param timeout the time to wait after an item emission towards the downstream
+ * before trying to emit the latest item from upstream again
+ * @param unit the time unit
+ * @param scheduler the {@code Scheduler} where the timed wait and latest item
+ * emission will be performed
+ * @param emitLast If {@code true}, the very last item from the upstream will be emitted
+ * immediately when the upstream completes, regardless if there is
+ * a timeout window active or not. If {@code false}, the very last
+ * upstream item is ignored and the flow terminates.
+ * @param onDropped called when an item is replaced by a newer item that doesn't get delivered
+ * to the downstream, including the very last item if {@code emitLast} is {@code false}
+ * and the current undelivered item when the sequence gets disposed.
+ * @return the new {@code Observable} instance
+ * @throws NullPointerException if {@code unit}, {@code scheduler} or {@code onDropped} is {@code null}
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @NonNull
+ @Experimental
+ public final Observable throttleLatest(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast, @NonNull Consumer onDropped) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ Objects.requireNonNull(onDropped, "onDropped is null");
+ return RxJavaPlugins.onAssembly(new ObservableThrottleLatest<>(this, timeout, unit, scheduler, emitLast, onDropped));
}
/**
@@ -14432,6 +14645,45 @@ public final Observable throttleWithTimeout(long timeout, @NonNull TimeUnit u
return debounce(timeout, unit, scheduler);
}
+ /**
+ * Returns an {@code Observable} that mirrors the current {@code Observable}, except that it drops items emitted by the
+ * current {@code Observable} that are followed by newer items before a timeout value expires on a specified
+ * {@link Scheduler}. The timer resets on each emission (Alias to {@link #debounce(long, TimeUnit, Scheduler)}).
+ *
+ * Note: If items keep being emitted by the current {@code Observable} faster than the timeout then no items
+ * will be emitted by the resulting {@code Observable}.
+ *
+ *
+ *
+ * - Scheduler:
+ * - You specify which {@code Scheduler} this operator will use.
+ *
+ *
+ * @param timeout
+ * the length of the window of time that must pass after the emission of an item from the current
+ * {@code Observable}, in which the current {@code Observable} emits no items, in order for the item to be emitted by the
+ * resulting {@code Observable}
+ * @param unit
+ * the unit of time for the specified {@code timeout}
+ * @param scheduler
+ * the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
+ * item
+ * @param onDropped
+ * called with the current entry when it has been replaced by a new one
+ * @return the new {@code Observable} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+ * @see ReactiveX operators documentation: Debounce
+ * @see #debounce(long, TimeUnit, Scheduler, Consumer)
+ * @since 3.1.6 - Experimental
+ */
+ @CheckReturnValue
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ @NonNull
+ @Experimental
+ public final Observable throttleWithTimeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) {
+ return debounce(timeout, unit, scheduler, onDropped);
+ }
+
/**
* Returns an {@code Observable} that emits records of the time interval between consecutive items emitted by the
* current {@code Observable}.
@@ -16613,7 +16865,7 @@ public final TestObserver test(boolean dispose) { // NoPMD
* 
*
* Note that the operator takes an already instantiated, running or terminated {@code CompletionStage}.
- * If the optional is to be created per consumer upon subscription, use {@link #defer(Supplier)}
+ * If the {@code CompletionStage} is to be created per consumer upon subscription, use {@link #defer(Supplier)}
* around {@code fromCompletionStage}:
*
* Observable.defer(() -> Observable.fromCompletionStage(createCompletionStage()));
diff --git a/src/main/java/io/reactivex/rxjava3/core/Single.java b/src/main/java/io/reactivex/rxjava3/core/Single.java
index ac7ecd48b6..6cf5a3f789 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Single.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Single.java
@@ -5598,7 +5598,7 @@ private static Single toSingle(@NonNull Flowable source) {
*
*
* Note that the operator takes an already instantiated, running or terminated {@code CompletionStage}.
- * If the optional is to be created per consumer upon subscription, use {@link #defer(Supplier)}
+ * If the {@code CompletionStage} is to be created per consumer upon subscription, use {@link #defer(Supplier)}
* around {@code fromCompletionStage}:
*
* Single.defer(() -> Single.fromCompletionStage(createCompletionStage()));
diff --git a/src/main/java/io/reactivex/rxjava3/exceptions/MissingBackpressureException.java b/src/main/java/io/reactivex/rxjava3/exceptions/MissingBackpressureException.java
index a817e6fb93..f0a173ba59 100644
--- a/src/main/java/io/reactivex/rxjava3/exceptions/MissingBackpressureException.java
+++ b/src/main/java/io/reactivex/rxjava3/exceptions/MissingBackpressureException.java
@@ -20,6 +20,15 @@ public final class MissingBackpressureException extends RuntimeException {
private static final long serialVersionUID = 8517344746016032542L;
+ /**
+ * The default error message.
+ *
+ * This can happen if the downstream doesn't call {@link org.reactivestreams.Subscription#request(long)}
+ * in time or at all.
+ * @since 3.1.6
+ */
+ public static final String DEFAULT_MESSAGE = "Could not emit value due to lack of requests";
+
/**
* Constructs a MissingBackpressureException without message or cause.
*/
@@ -35,4 +44,13 @@ public MissingBackpressureException(String message) {
super(message);
}
+ /**
+ * Constructs a new {@code MissingBackpressureException} with the
+ * default message {@value #DEFAULT_MESSAGE}.
+ * @return the new {@code MissingBackpressureException} instance.
+ * @since 3.1.6
+ */
+ public static MissingBackpressureException createDefault() {
+ return new MissingBackpressureException(DEFAULT_MESSAGE);
+ }
}
diff --git a/src/main/java/io/reactivex/rxjava3/exceptions/QueueOverflowException.java b/src/main/java/io/reactivex/rxjava3/exceptions/QueueOverflowException.java
new file mode 100644
index 0000000000..bdd8a25e6f
--- /dev/null
+++ b/src/main/java/io/reactivex/rxjava3/exceptions/QueueOverflowException.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava3.exceptions;
+
+/**
+ * Indicates an overflow happened because the upstream disregarded backpressure completely or
+ * {@link org.reactivestreams.Subscriber#onNext(Object)} was called concurrently from multiple threads
+ * without synchronization. Rarely, it is an indication of bugs inside an operator.
+ * @since 3.1.6
+ */
+public final class QueueOverflowException extends RuntimeException {
+
+ private static final long serialVersionUID = 8517344746016032542L;
+
+ /**
+ * The message for queue overflows.
+ *
+ * This can happen if the upstream disregards backpressure completely or calls
+ * {@link org.reactivestreams.Subscriber#onNext(Object)} concurrently from multiple threads
+ * without synchronization. Rarely, it is an indication of bugs inside an operator.
+ */
+ private static final String DEFAULT_MESSAGE = "Queue overflow due to illegal concurrent onNext calls or a bug in an operator";
+
+ /**
+ * Constructs a QueueOverflowException with the default message.
+ */
+ public QueueOverflowException() {
+ this(DEFAULT_MESSAGE);
+ }
+
+ /**
+ * Constructs a QueueOverflowException with the given message but no cause.
+ * @param message the error message
+ */
+ public QueueOverflowException(String message) {
+ super(message);
+ }
+}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableFlatMapStream.java b/src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableFlatMapStream.java
index 35292bb17c..48ea8b6e5f 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableFlatMapStream.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableFlatMapStream.java
@@ -174,7 +174,7 @@ public void onNext(T t) {
if (sourceMode != QueueFuseable.ASYNC) {
if (!queue.offer(t)) {
upstream.cancel();
- onError(new MissingBackpressureException("Queue full?!"));
+ onError(new QueueOverflowException());
return;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableConcat.java b/src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableConcat.java
index d83ce2d398..00a53f3ad7 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableConcat.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableConcat.java
@@ -120,7 +120,7 @@ public void onSubscribe(Subscription s) {
public void onNext(CompletableSource t) {
if (sourceFused == QueueSubscription.NONE) {
if (!queue.offer(t)) {
- onError(new MissingBackpressureException());
+ onError(new QueueOverflowException());
return;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/BlockingFlowableIterable.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/BlockingFlowableIterable.java
index a25032e33b..36436d1370 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/BlockingFlowableIterable.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/BlockingFlowableIterable.java
@@ -140,7 +140,7 @@ public void onNext(T t) {
if (!queue.offer(t)) {
SubscriptionHelper.cancel(this);
- onError(new MissingBackpressureException("Queue full?!"));
+ onError(new QueueOverflowException());
} else {
signalConsumer();
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableBufferExactBoundary.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableBufferExactBoundary.java
index e543272a39..3e2abefdd9 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableBufferExactBoundary.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableBufferExactBoundary.java
@@ -45,7 +45,7 @@ protected void subscribeActual(Subscriber s) {
}
static final class BufferExactBoundarySubscriber, B>
- extends QueueDrainSubscriber implements FlowableSubscriber, Subscription, Disposable {
+ extends QueueDrainSubscriber implements Subscription, Disposable {
final Supplier bufferSupplier;
final Publisher boundary;
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMap.java
index bf0f0ff163..5d1b6e4c33 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMap.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMap.java
@@ -19,7 +19,7 @@
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.exceptions.*;
import io.reactivex.rxjava3.functions.*;
import io.reactivex.rxjava3.internal.subscriptions.*;
import io.reactivex.rxjava3.internal.util.*;
@@ -152,7 +152,7 @@ public final void onNext(T t) {
if (sourceMode != QueueSubscription.ASYNC) {
if (!queue.offer(t)) {
upstream.cancel();
- onError(new IllegalStateException("Queue full?!"));
+ onError(new QueueOverflowException());
return;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapEager.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapEager.java
index af33ec9281..dc423d89fc 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapEager.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapEager.java
@@ -200,7 +200,7 @@ public void innerNext(InnerQueuedSubscriber inner, R value) {
drain();
} else {
inner.cancel();
- innerError(inner, new MissingBackpressureException());
+ innerError(inner, MissingBackpressureException.createDefault());
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapScheduler.java
index c53ebe0867..707d10a19e 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapScheduler.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapScheduler.java
@@ -19,7 +19,7 @@
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.exceptions.*;
import io.reactivex.rxjava3.functions.*;
import io.reactivex.rxjava3.internal.operators.flowable.FlowableConcatMap.*;
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
@@ -151,7 +151,7 @@ public final void onNext(T t) {
if (sourceMode != QueueSubscription.ASYNC) {
if (!queue.offer(t)) {
upstream.cancel();
- onError(new IllegalStateException("Queue full?!"));
+ onError(new QueueOverflowException());
return;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCreate.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCreate.java
index 673fc4bc22..1e87c4b7e6 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCreate.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCreate.java
@@ -442,7 +442,7 @@ static final class ErrorAsyncEmitter extends NoOverflowBaseAsyncEmitter {
@Override
void onOverflow() {
- onError(new MissingBackpressureException("create: could not emit value due to lack of requests"));
+ onError(new MissingBackpressureException("create: " + MissingBackpressureException.DEFAULT_MESSAGE));
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounce.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounce.java
index b1066c0133..51838a1864 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounce.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounce.java
@@ -148,7 +148,7 @@ void emit(long idx, T value) {
BackpressureHelper.produced(this, 1);
} else {
cancel();
- downstream.onError(new MissingBackpressureException("Could not deliver value due to lack of requests"));
+ downstream.onError(MissingBackpressureException.createDefault());
}
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounceTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounceTimed.java
index 35bfaec397..b9e5dff7f3 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounceTimed.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounceTimed.java
@@ -16,6 +16,8 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.*;
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.Consumer;
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;
@@ -32,19 +34,20 @@ public final class FlowableDebounceTimed extends AbstractFlowableWithUpstream
final long timeout;
final TimeUnit unit;
final Scheduler scheduler;
+ final Consumer onDropped;
- public FlowableDebounceTimed(Flowable source, long timeout, TimeUnit unit, Scheduler scheduler) {
+ public FlowableDebounceTimed(Flowable source, long timeout, TimeUnit unit, Scheduler scheduler, Consumer onDropped) {
super(source);
this.timeout = timeout;
this.unit = unit;
this.scheduler = scheduler;
+ this.onDropped = onDropped;
}
@Override
protected void subscribeActual(Subscriber s) {
source.subscribe(new DebounceTimedSubscriber<>(
- new SerializedSubscriber<>(s),
- timeout, unit, scheduler.createWorker()));
+ new SerializedSubscriber<>(s), timeout, unit, scheduler.createWorker(), onDropped));
}
static final class DebounceTimedSubscriber extends AtomicLong
@@ -55,20 +58,22 @@ static final class DebounceTimedSubscriber extends AtomicLong
final long timeout;
final TimeUnit unit;
final Scheduler.Worker worker;
+ final Consumer onDropped;
Subscription upstream;
- Disposable timer;
+ DebounceEmitter timer;
volatile long index;
boolean done;
- DebounceTimedSubscriber(Subscriber actual, long timeout, TimeUnit unit, Worker worker) {
+ DebounceTimedSubscriber(Subscriber actual, long timeout, TimeUnit unit, Worker worker, Consumer onDropped) {
this.downstream = actual;
this.timeout = timeout;
this.unit = unit;
this.worker = worker;
+ this.onDropped = onDropped;
}
@Override
@@ -88,15 +93,26 @@ public void onNext(T t) {
long idx = index + 1;
index = idx;
- Disposable d = timer;
- if (d != null) {
- d.dispose();
+ DebounceEmitter currentEmitter = timer;
+ if (currentEmitter != null) {
+ currentEmitter.dispose();
+ }
+
+ if (onDropped != null && currentEmitter != null) {
+ try {
+ onDropped.accept(currentEmitter.value);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ upstream.cancel();
+ done = true;
+ downstream.onError(ex);
+ worker.dispose();
+ }
}
- DebounceEmitter de = new DebounceEmitter<>(t, idx, this);
- timer = de;
- d = worker.schedule(de, timeout, unit);
- de.setResource(d);
+ DebounceEmitter newEmitter = new DebounceEmitter<>(t, idx, this);
+ timer = newEmitter;
+ newEmitter.setResource(worker.schedule(newEmitter, timeout, unit));
}
@Override
@@ -121,15 +137,13 @@ public void onComplete() {
}
done = true;
- Disposable d = timer;
+ DebounceEmitter d = timer;
if (d != null) {
d.dispose();
}
- @SuppressWarnings("unchecked")
- DebounceEmitter de = (DebounceEmitter)d;
- if (de != null) {
- de.emit();
+ if (d != null) {
+ d.emit();
}
downstream.onComplete();
@@ -159,7 +173,7 @@ void emit(long idx, T t, DebounceEmitter emitter) {
emitter.dispose();
} else {
cancel();
- downstream.onError(new MissingBackpressureException("Could not deliver value due to lack of requests"));
+ downstream.onError(MissingBackpressureException.createDefault());
}
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMap.java
index 78c9a9e542..c250d3b165 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMap.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMap.java
@@ -243,7 +243,7 @@ void tryEmitScalar(U value) {
q = getMainQueue();
}
if (!q.offer(value)) {
- onError(new MissingBackpressureException("Scalar queue full?!"));
+ onError(new QueueOverflowException());
}
}
if (decrementAndGet() == 0) {
@@ -252,7 +252,7 @@ void tryEmitScalar(U value) {
} else {
SimpleQueue q = getMainQueue();
if (!q.offer(value)) {
- onError(new MissingBackpressureException("Scalar queue full?!"));
+ onError(new QueueOverflowException());
return;
}
if (getAndIncrement() != 0) {
@@ -278,7 +278,7 @@ void tryEmit(U value, InnerSubscriber inner) {
inner.queue = q;
}
if (!q.offer(value)) {
- onError(new MissingBackpressureException("Inner queue full?!"));
+ onError(new QueueOverflowException());
}
}
if (decrementAndGet() == 0) {
@@ -291,7 +291,7 @@ void tryEmit(U value, InnerSubscriber inner) {
inner.queue = q;
}
if (!q.offer(value)) {
- onError(new MissingBackpressureException("Inner queue full?!"));
+ onError(new QueueOverflowException());
return;
}
if (getAndIncrement() != 0) {
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlattenIterable.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlattenIterable.java
index 019a132329..0b9164e0cd 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlattenIterable.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlattenIterable.java
@@ -180,7 +180,7 @@ public void onNext(T t) {
return;
}
if (fusionMode == NONE && !queue.offer(t)) {
- onError(new MissingBackpressureException("Queue is full?!"));
+ onError(new QueueOverflowException());
return;
}
drain();
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java
index 344b029018..50e641ffc1 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java
@@ -168,7 +168,7 @@ public void onNext(T t) {
if (emittedGroups != get()) {
downstream.onNext(group);
} else {
- MissingBackpressureException mbe = new MissingBackpressureException(groupHangWarning(emittedGroups));
+ MissingBackpressureException mbe = groupHangWarning(emittedGroups);
mbe.initCause(ex);
onError(mbe);
return;
@@ -194,13 +194,13 @@ public void onNext(T t) {
}
} else {
upstream.cancel();
- onError(new MissingBackpressureException(groupHangWarning(emittedGroups)));
+ onError(groupHangWarning(emittedGroups));
}
}
}
- static String groupHangWarning(long n) {
- return "Unable to emit a new group (#" + n + ") due to lack of requests. Please make sure the downstream can always accept a new group as well as each group is consumed in order for the whole operator to be able to proceed.";
+ static MissingBackpressureException groupHangWarning(long n) {
+ return new MissingBackpressureException("Unable to emit a new group (#" + n + ") due to lack of requests. Please make sure the downstream can always accept a new group as well as each group is consumed in order for the whole operator to be able to proceed.");
}
@Override
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupJoin.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupJoin.java
index 3021b9ddc5..c9619c48b7 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupJoin.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupJoin.java
@@ -276,7 +276,7 @@ void drain() {
a.onNext(w);
BackpressureHelper.produced(requested, 1);
} else {
- fail(new MissingBackpressureException("Could not emit value due to lack of requests"), a, q);
+ fail(MissingBackpressureException.createDefault(), a, q);
return;
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableInterval.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableInterval.java
index 536e36c2ea..98fbe6dee1 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableInterval.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableInterval.java
@@ -93,7 +93,7 @@ public void run() {
downstream.onNext(count++);
BackpressureHelper.produced(this, 1);
} else {
- downstream.onError(new MissingBackpressureException("Can't deliver value " + count + " due to lack of requests"));
+ downstream.onError(new MissingBackpressureException("Could not emit value " + count + " due to lack of requests"));
DisposableHelper.dispose(resource);
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableIntervalRange.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableIntervalRange.java
index 7c556acdac..cfa605882d 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableIntervalRange.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableIntervalRange.java
@@ -114,7 +114,7 @@ public void run() {
decrementAndGet();
}
} else {
- downstream.onError(new MissingBackpressureException("Can't deliver value " + count + " due to lack of requests"));
+ downstream.onError(new MissingBackpressureException("Could not emit value " + count + " due to lack of requests"));
DisposableHelper.dispose(resource);
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableJoin.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableJoin.java
index 2c269cee4d..9cda3d457d 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableJoin.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableJoin.java
@@ -260,7 +260,7 @@ void drain() {
e++;
} else {
- ExceptionHelper.addThrowable(error, new MissingBackpressureException("Could not emit value due to lack of requests"));
+ ExceptionHelper.addThrowable(error, MissingBackpressureException.createDefault());
q.clear();
cancelAll();
errorAll(a);
@@ -321,7 +321,7 @@ else if (mode == RIGHT_VALUE) {
e++;
} else {
- ExceptionHelper.addThrowable(error, new MissingBackpressureException("Could not emit value due to lack of requests"));
+ ExceptionHelper.addThrowable(error, MissingBackpressureException.createDefault());
q.clear();
cancelAll();
errorAll(a);
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableObserveOn.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableObserveOn.java
index 4e9f83e41e..576debbf2b 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableObserveOn.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableObserveOn.java
@@ -113,7 +113,7 @@ public final void onNext(T t) {
if (!queue.offer(t)) {
upstream.cancel();
- error = new MissingBackpressureException("Queue is full?!");
+ error = new QueueOverflowException();
done = true;
}
trySchedule();
@@ -244,8 +244,7 @@ public final boolean isEmpty() {
}
}
- static final class ObserveOnSubscriber extends BaseObserveOnSubscriber
- implements FlowableSubscriber {
+ static final class ObserveOnSubscriber extends BaseObserveOnSubscriber {
private static final long serialVersionUID = -4547113800637756442L;
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java
index 9b0b8a7b20..29a207fba5 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java
@@ -140,7 +140,7 @@ public void onNext(T t) {
}
} else if (callError) {
upstream.cancel();
- onError(new MissingBackpressureException());
+ onError(MissingBackpressureException.createDefault());
} else {
drain();
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureError.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureError.java
index 03e668b2ba..acaf165069 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureError.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureError.java
@@ -66,7 +66,7 @@ public void onNext(T t) {
BackpressureHelper.produced(this, 1);
} else {
upstream.cancel();
- onError(new MissingBackpressureException("could not emit value due to lack of requests"));
+ onError(MissingBackpressureException.createDefault());
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublish.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublish.java
index 87f43b762b..b05bc8b748 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublish.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublish.java
@@ -226,7 +226,7 @@ public void onSubscribe(Subscription s) {
public void onNext(T t) {
// we expect upstream to honor backpressure requests
if (sourceMode == QueueSubscription.NONE && !queue.offer(t)) {
- onError(new MissingBackpressureException("Prefetch queue is full?!"));
+ onError(new QueueOverflowException());
return;
}
// since many things can happen concurrently, we have a common dispatch
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublishMulticast.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublishMulticast.java
index c97b8b6bbf..08d0e40058 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublishMulticast.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublishMulticast.java
@@ -212,7 +212,7 @@ public void onNext(T t) {
}
if (sourceMode == QueueSubscription.NONE && !queue.offer(t)) {
upstream.get().cancel();
- onError(new MissingBackpressureException());
+ onError(MissingBackpressureException.createDefault());
return;
}
drain();
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSamplePublisher.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSamplePublisher.java
index 778f3faf21..19474d2cc0 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSamplePublisher.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSamplePublisher.java
@@ -129,7 +129,7 @@ void emit() {
BackpressureHelper.produced(requested, 1);
} else {
cancel();
- downstream.onError(new MissingBackpressureException("Couldn't emit value due to lack of requests!"));
+ downstream.onError(MissingBackpressureException.createDefault());
}
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSampleTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSampleTimed.java
index e17ad04a66..40551f4a8e 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSampleTimed.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSampleTimed.java
@@ -16,6 +16,8 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.*;
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.Consumer;
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;
@@ -29,24 +31,25 @@ public final class FlowableSampleTimed extends AbstractFlowableWithUpstream onDropped;
- public FlowableSampleTimed(Flowable source, long period, TimeUnit unit, Scheduler scheduler, boolean emitLast) {
+ public FlowableSampleTimed(Flowable source, long period, TimeUnit unit, Scheduler scheduler, boolean emitLast, Consumer onDropped) {
super(source);
this.period = period;
this.unit = unit;
this.scheduler = scheduler;
this.emitLast = emitLast;
+ this.onDropped = onDropped;
}
@Override
protected void subscribeActual(Subscriber s) {
SerializedSubscriber serial = new SerializedSubscriber<>(s);
if (emitLast) {
- source.subscribe(new SampleTimedEmitLast<>(serial, period, unit, scheduler));
+ source.subscribe(new SampleTimedEmitLast<>(serial, period, unit, scheduler, onDropped));
} else {
- source.subscribe(new SampleTimedNoLast<>(serial, period, unit, scheduler));
+ source.subscribe(new SampleTimedNoLast<>(serial, period, unit, scheduler, onDropped));
}
}
@@ -58,6 +61,7 @@ abstract static class SampleTimedSubscriber extends AtomicReference implem
final long period;
final TimeUnit unit;
final Scheduler scheduler;
+ final Consumer onDropped;
final AtomicLong requested = new AtomicLong();
@@ -65,11 +69,12 @@ abstract static class SampleTimedSubscriber extends AtomicReference implem
Subscription upstream;
- SampleTimedSubscriber(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler) {
+ SampleTimedSubscriber(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) {
this.downstream = actual;
this.period = period;
this.unit = unit;
this.scheduler = scheduler;
+ this.onDropped = onDropped;
}
@Override
@@ -84,7 +89,17 @@ public void onSubscribe(Subscription s) {
@Override
public void onNext(T t) {
- lazySet(t);
+ T oldValue = getAndSet(t);
+ if (oldValue != null && onDropped != null) {
+ try {
+ onDropped.accept(oldValue);
+ } catch (Throwable throwable) {
+ Exceptions.throwIfFatal(throwable);
+ cancelTimer();
+ upstream.cancel();
+ downstream.onError(throwable);
+ }
+ }
}
@Override
@@ -125,7 +140,7 @@ void emit() {
BackpressureHelper.produced(requested, 1);
} else {
cancel();
- downstream.onError(new MissingBackpressureException("Couldn't emit value due to lack of requests!"));
+ downstream.onError(MissingBackpressureException.createDefault());
}
}
}
@@ -137,8 +152,8 @@ static final class SampleTimedNoLast extends SampleTimedSubscriber {
private static final long serialVersionUID = -7139995637533111443L;
- SampleTimedNoLast(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler) {
- super(actual, period, unit, scheduler);
+ SampleTimedNoLast(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) {
+ super(actual, period, unit, scheduler, onDropped);
}
@Override
@@ -158,8 +173,8 @@ static final class SampleTimedEmitLast extends SampleTimedSubscriber {
final AtomicInteger wip;
- SampleTimedEmitLast(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler) {
- super(actual, period, unit, scheduler);
+ SampleTimedEmitLast(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) {
+ super(actual, period, unit, scheduler, onDropped);
this.wip = new AtomicInteger(1);
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSequenceEqual.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSequenceEqual.java
index 0ba38698d6..8dadbe0161 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSequenceEqual.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSequenceEqual.java
@@ -300,7 +300,7 @@ public void onSubscribe(Subscription s) {
public void onNext(T t) {
if (sourceMode == QueueSubscription.NONE) {
if (!queue.offer(t)) {
- onError(new MissingBackpressureException());
+ onError(MissingBackpressureException.createDefault());
return;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSwitchMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSwitchMap.java
index 1c820570dc..4d4521c655 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSwitchMap.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSwitchMap.java
@@ -381,7 +381,7 @@ public void onNext(R t) {
SwitchMapSubscriber p = parent;
if (index == p.unique) {
if (fusionMode == QueueSubscription.NONE && !queue.offer(t)) {
- onError(new MissingBackpressureException("Queue full?!"));
+ onError(new QueueOverflowException());
return;
}
p.drain();
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java
index c12399e1d1..44499ec255 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java
@@ -16,6 +16,8 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.Consumer;
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;
@@ -32,44 +34,53 @@ public final class FlowableThrottleFirstTimed extends AbstractFlowableWithUps
final long timeout;
final TimeUnit unit;
final Scheduler scheduler;
+ final Consumer onDropped;
- public FlowableThrottleFirstTimed(Flowable source, long timeout, TimeUnit unit, Scheduler scheduler) {
+ public FlowableThrottleFirstTimed(Flowable source,
+ long timeout,
+ TimeUnit unit,
+ Scheduler scheduler,
+ Consumer onDropped) {
super(source);
this.timeout = timeout;
this.unit = unit;
this.scheduler = scheduler;
+ this.onDropped = onDropped;
}
@Override
protected void subscribeActual(Subscriber s) {
source.subscribe(new DebounceTimedSubscriber<>(
new SerializedSubscriber<>(s),
- timeout, unit, scheduler.createWorker()));
+ timeout, unit, scheduler.createWorker(),
+ onDropped));
}
static final class DebounceTimedSubscriber
extends AtomicLong
implements FlowableSubscriber, Subscription, Runnable {
-
private static final long serialVersionUID = -9102637559663639004L;
+
final Subscriber downstream;
final long timeout;
final TimeUnit unit;
final Scheduler.Worker worker;
-
+ final Consumer onDropped;
Subscription upstream;
-
final SequentialDisposable timer = new SequentialDisposable();
-
volatile boolean gate;
-
boolean done;
- DebounceTimedSubscriber(Subscriber actual, long timeout, TimeUnit unit, Worker worker) {
+ DebounceTimedSubscriber(Subscriber actual,
+ long timeout,
+ TimeUnit unit,
+ Worker worker,
+ Consumer onDropped) {
this.downstream = actual;
this.timeout = timeout;
this.unit = unit;
this.worker = worker;
+ this.onDropped = onDropped;
}
@Override
@@ -94,9 +105,10 @@ public void onNext(T t) {
downstream.onNext(t);
BackpressureHelper.produced(this, 1);
} else {
+ upstream.cancel();
done = true;
- cancel();
- downstream.onError(new MissingBackpressureException("Could not deliver value due to lack of requests"));
+ downstream.onError(MissingBackpressureException.createDefault());
+ worker.dispose();
return;
}
@@ -106,6 +118,16 @@ public void onNext(T t) {
}
timer.replace(worker.schedule(this, timeout, unit));
+ } else if (onDropped != null) {
+ try {
+ onDropped.accept(t);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ upstream.cancel();
+ done = true;
+ downstream.onError(ex);
+ worker.dispose();
+ }
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleLatest.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleLatest.java
index e6cb5ffdcc..e28e5f09df 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleLatest.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleLatest.java
@@ -19,9 +19,11 @@
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.exceptions.MissingBackpressureException;
+import io.reactivex.rxjava3.exceptions.*;
+import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
import io.reactivex.rxjava3.internal.util.BackpressureHelper;
+import io.reactivex.rxjava3.plugins.RxJavaPlugins;
/**
* Emits the next or latest item when the given time elapses.
@@ -44,19 +46,24 @@ public final class FlowableThrottleLatest extends AbstractFlowableWithUpstrea
final boolean emitLast;
+ final Consumer onDropped;
+
public FlowableThrottleLatest(Flowable source,
- long timeout, TimeUnit unit, Scheduler scheduler,
- boolean emitLast) {
+ long timeout, TimeUnit unit,
+ Scheduler scheduler,
+ boolean emitLast,
+ Consumer onDropped) {
super(source);
this.timeout = timeout;
this.unit = unit;
this.scheduler = scheduler;
this.emitLast = emitLast;
+ this.onDropped = onDropped;
}
@Override
protected void subscribeActual(Subscriber s) {
- source.subscribe(new ThrottleLatestSubscriber<>(s, timeout, unit, scheduler.createWorker(), emitLast));
+ source.subscribe(new ThrottleLatestSubscriber<>(s, timeout, unit, scheduler.createWorker(), emitLast, onDropped));
}
static final class ThrottleLatestSubscriber
@@ -79,6 +86,8 @@ static final class ThrottleLatestSubscriber
final AtomicLong requested;
+ final Consumer onDropped;
+
Subscription upstream;
volatile boolean done;
@@ -93,8 +102,10 @@ static final class ThrottleLatestSubscriber
boolean timerRunning;
ThrottleLatestSubscriber(Subscriber downstream,
- long timeout, TimeUnit unit, Scheduler.Worker worker,
- boolean emitLast) {
+ long timeout, TimeUnit unit,
+ Scheduler.Worker worker,
+ boolean emitLast,
+ Consumer onDropped) {
this.downstream = downstream;
this.timeout = timeout;
this.unit = unit;
@@ -102,6 +113,7 @@ static final class ThrottleLatestSubscriber
this.emitLast = emitLast;
this.latest = new AtomicReference<>();
this.requested = new AtomicLong();
+ this.onDropped = onDropped;
}
@Override
@@ -115,7 +127,17 @@ public void onSubscribe(Subscription s) {
@Override
public void onNext(T t) {
- latest.set(t);
+ T old = latest.getAndSet(t);
+ if (onDropped != null && old != null) {
+ try {
+ onDropped.accept(old);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ upstream.cancel();
+ error = ex;
+ done = true;
+ }
+ }
drain();
}
@@ -145,6 +167,22 @@ public void cancel() {
upstream.cancel();
worker.dispose();
if (getAndIncrement() == 0) {
+ clear();
+ }
+ }
+
+ void clear() {
+ if (onDropped != null) {
+ T v = latest.getAndSet(null);
+ if (v != null) {
+ try {
+ onDropped.accept(v);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ RxJavaPlugins.onError(ex);
+ }
+ }
+ } else {
latest.lazySet(null);
}
}
@@ -170,14 +208,27 @@ void drain() {
for (;;) {
if (cancelled) {
- latest.lazySet(null);
+ clear();
return;
}
boolean d = done;
+ Throwable error = this.error;
if (d && error != null) {
- latest.lazySet(null);
+ if (onDropped != null) {
+ T v = latest.getAndSet(null);
+ if (v != null) {
+ try {
+ onDropped.accept(v);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ error = new CompositeException(error, ex);
+ }
+ }
+ } else {
+ latest.lazySet(null);
+ }
downstream.onError(error);
worker.dispose();
return;
@@ -187,19 +238,31 @@ void drain() {
boolean empty = v == null;
if (d) {
- if (!empty && emitLast) {
+ if (!empty) {
v = latest.getAndSet(null);
- long e = emitted;
- if (e != requested.get()) {
- emitted = e + 1;
- downstream.onNext(v);
- downstream.onComplete();
+ if (emitLast) {
+ long e = emitted;
+ if (e != requested.get()) {
+ emitted = e + 1;
+ downstream.onNext(v);
+ downstream.onComplete();
+ } else {
+ tryDropAndSignalMBE(v);
+ }
} else {
- downstream.onError(new MissingBackpressureException(
- "Could not emit final value due to lack of requests"));
+ if (onDropped != null) {
+ try {
+ onDropped.accept(v);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ downstream.onError(ex);
+ worker.dispose();
+ return;
+ }
+ }
+ downstream.onComplete();
}
} else {
- latest.lazySet(null);
downstream.onComplete();
}
worker.dispose();
@@ -222,8 +285,7 @@ void drain() {
emitted = e + 1;
} else {
upstream.cancel();
- downstream.onError(new MissingBackpressureException(
- "Could not emit value due to lack of requests"));
+ tryDropAndSignalMBE(v);
worker.dispose();
return;
}
@@ -242,5 +304,18 @@ void drain() {
}
}
}
+
+ void tryDropAndSignalMBE(T valueToDrop) {
+ Throwable errorToSignal = MissingBackpressureException.createDefault();
+ if (onDropped != null) {
+ try {
+ onDropped.accept(valueToDrop);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ errorToSignal = new CompositeException(errorToSignal, ex);
+ }
+ }
+ downstream.onError(errorToSignal);
+ }
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableTimer.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableTimer.java
index dd33daea0f..db45fe2e98 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableTimer.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableTimer.java
@@ -78,7 +78,7 @@ public void run() {
downstream.onComplete();
} else {
lazySet(EmptyDisposable.INSTANCE);
- downstream.onError(new MissingBackpressureException("Can't deliver value due to lack of requests"));
+ downstream.onError(MissingBackpressureException.createDefault());
}
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableToList.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableToList.java
index 2b7c19e269..3f03d93112 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableToList.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableToList.java
@@ -46,7 +46,7 @@ protected void subscribeActual(Subscriber s) {
static final class ToListSubscriber>
extends DeferredScalarSubscription
- implements FlowableSubscriber, Subscription {
+ implements FlowableSubscriber {
private static final long serialVersionUID = -8134157938864266736L;
Subscription upstream;
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundary.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundary.java
index 1dcfd4e529..2aad69dcb5 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundary.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundary.java
@@ -248,7 +248,7 @@ void drain() {
} else {
SubscriptionHelper.cancel(upstream);
boundarySubscriber.dispose();
- errors.tryAddThrowableOrReport(new MissingBackpressureException("Could not deliver a window due to lack of requests"));
+ errors.tryAddThrowableOrReport(MissingBackpressureException.createDefault());
done = true;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundarySelector.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundarySelector.java
index 394315cd9b..d6bffae2e1 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundarySelector.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundarySelector.java
@@ -272,7 +272,7 @@ void drain() {
upstream.cancel();
startSubscriber.cancel();
resources.dispose();
- error.tryAddThrowableOrReport(new MissingBackpressureException(FlowableWindowTimed.missingBackpressureMessage(emitted)));
+ error.tryAddThrowableOrReport(FlowableWindowTimed.missingBackpressureMessage(emitted));
upstreamDone = true;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java
index 23ef916fef..39e7de59ad 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowTimed.java
@@ -213,7 +213,7 @@ void createFirstWindow() {
upstream.request(Long.MAX_VALUE);
} else {
upstream.cancel();
- downstream.onError(new MissingBackpressureException(missingBackpressureMessage(emitted)));
+ downstream.onError(missingBackpressureMessage(emitted));
cleanupResources();
upstreamCancelled = true;
@@ -282,7 +282,7 @@ else if (!isEmpty) {
cleanupResources();
upstreamCancelled = true;
- downstream.onError(new MissingBackpressureException(missingBackpressureMessage(emitted)));
+ downstream.onError(missingBackpressureMessage(emitted));
} else {
emitted++;
@@ -386,7 +386,7 @@ void createFirstWindow() {
upstream.request(Long.MAX_VALUE);
} else {
upstream.cancel();
- downstream.onError(new MissingBackpressureException(missingBackpressureMessage(emitted)));
+ downstream.onError(missingBackpressureMessage(emitted));
cleanupResources();
upstreamCancelled = true;
@@ -499,7 +499,7 @@ UnicastProcessor createNewWindow(UnicastProcessor window) {
cleanupResources();
upstreamCancelled = true;
- downstream.onError(new MissingBackpressureException(missingBackpressureMessage(emitted)));
+ downstream.onError(missingBackpressureMessage(emitted));
} else {
this.emitted = ++emitted;
@@ -584,7 +584,7 @@ void createFirstWindow() {
upstream.request(Long.MAX_VALUE);
} else {
upstream.cancel();
- downstream.onError(new MissingBackpressureException(missingBackpressureMessage(emitted)));
+ downstream.onError(missingBackpressureMessage(emitted));
cleanupResources();
upstreamCancelled = true;
@@ -654,7 +654,7 @@ void drain() {
}
} else {
upstream.cancel();
- Throwable ex = new MissingBackpressureException(missingBackpressureMessage(emitted));
+ Throwable ex = missingBackpressureMessage(emitted);
for (UnicastProcessor window : windows) {
window.onError(ex);
}
@@ -717,8 +717,8 @@ public void run() {
}
}
- static String missingBackpressureMessage(long index) {
- return "Unable to emit the next window (#" + index + ") due to lack of requests. Please make sure the downstream is ready to consume windows.";
+ static MissingBackpressureException missingBackpressureMessage(long index) {
+ return new MissingBackpressureException("Unable to emit the next window (#" + index + ") due to lack of requests. Please make sure the downstream is ready to consume windows.");
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/ConcatMapXMainSubscriber.java b/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/ConcatMapXMainSubscriber.java
index bd701de40c..03e4c568d6 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/ConcatMapXMainSubscriber.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/mixed/ConcatMapXMainSubscriber.java
@@ -18,7 +18,7 @@
import org.reactivestreams.Subscription;
import io.reactivex.rxjava3.core.FlowableSubscriber;
-import io.reactivex.rxjava3.exceptions.MissingBackpressureException;
+import io.reactivex.rxjava3.exceptions.*;
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
import io.reactivex.rxjava3.internal.util.*;
import io.reactivex.rxjava3.operators.QueueFuseable;
@@ -99,7 +99,7 @@ public final void onNext(T t) {
if (t != null) {
if (!queue.offer(t)) {
upstream.cancel();
- onError(new MissingBackpressureException("queue full?!"));
+ onError(new QueueOverflowException());
return;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableBufferExactBoundary.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableBufferExactBoundary.java
index 5892e092b9..bb65058908 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableBufferExactBoundary.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableBufferExactBoundary.java
@@ -43,7 +43,7 @@ protected void subscribeActual(Observer t) {
}
static final class BufferExactBoundaryObserver, B>
- extends QueueDrainObserver implements Observer, Disposable {
+ extends QueueDrainObserver implements Disposable {
final Supplier bufferSupplier;
final ObservableSource boundary;
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableDebounceTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableDebounceTimed.java
index 248d00ea23..f2db191229 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableDebounceTimed.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableDebounceTimed.java
@@ -19,6 +19,8 @@
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.core.Scheduler.Worker;
import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
import io.reactivex.rxjava3.observers.SerializedObserver;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
@@ -27,19 +29,20 @@ public final class ObservableDebounceTimed extends AbstractObservableWithUpst
final long timeout;
final TimeUnit unit;
final Scheduler scheduler;
+ final Consumer onDropped;
- public ObservableDebounceTimed(ObservableSource source, long timeout, TimeUnit unit, Scheduler scheduler) {
+ public ObservableDebounceTimed(ObservableSource source, long timeout, TimeUnit unit, Scheduler scheduler, Consumer onDropped) {
super(source);
this.timeout = timeout;
this.unit = unit;
this.scheduler = scheduler;
+ this.onDropped = onDropped;
}
@Override
public void subscribeActual(Observer t) {
source.subscribe(new DebounceTimedObserver<>(
- new SerializedObserver<>(t),
- timeout, unit, scheduler.createWorker()));
+ new SerializedObserver<>(t), timeout, unit, scheduler.createWorker(), onDropped));
}
static final class DebounceTimedObserver
@@ -48,20 +51,22 @@ static final class DebounceTimedObserver
final long timeout;
final TimeUnit unit;
final Scheduler.Worker worker;
+ final Consumer onDropped;
Disposable upstream;
- Disposable timer;
+ DebounceEmitter timer;
volatile long index;
boolean done;
- DebounceTimedObserver(Observer actual, long timeout, TimeUnit unit, Worker worker) {
+ DebounceTimedObserver(Observer actual, long timeout, TimeUnit unit, Worker worker, Consumer onDropped) {
this.downstream = actual;
this.timeout = timeout;
this.unit = unit;
this.worker = worker;
+ this.onDropped = onDropped;
}
@Override
@@ -80,15 +85,25 @@ public void onNext(T t) {
long idx = index + 1;
index = idx;
- Disposable d = timer;
- if (d != null) {
- d.dispose();
+ DebounceEmitter currentEmitter = timer;
+ if (currentEmitter != null) {
+ currentEmitter.dispose();
}
- DebounceEmitter de = new DebounceEmitter<>(t, idx, this);
- timer = de;
- d = worker.schedule(de, timeout, unit);
- de.setResource(d);
+ if (onDropped != null && currentEmitter != null) {
+ try {
+ onDropped.accept(timer.value);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ upstream.dispose();
+ downstream.onError(ex);
+ done = true;
+ }
+ }
+
+ DebounceEmitter newEmitter = new DebounceEmitter<>(t, idx, this);
+ timer = newEmitter;
+ newEmitter.setResource(worker.schedule(newEmitter, timeout, unit));
}
@Override
@@ -113,15 +128,13 @@ public void onComplete() {
}
done = true;
- Disposable d = timer;
+ DebounceEmitter d = timer;
if (d != null) {
d.dispose();
}
- @SuppressWarnings("unchecked")
- DebounceEmitter de = (DebounceEmitter)d;
- if (de != null) {
- de.run();
+ if (d != null) {
+ d.run();
}
downstream.onComplete();
worker.dispose();
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSampleTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSampleTimed.java
index f439339e15..f264b8e76d 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSampleTimed.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSampleTimed.java
@@ -18,6 +18,8 @@
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
import io.reactivex.rxjava3.observers.SerializedObserver;
@@ -25,24 +27,30 @@ public final class ObservableSampleTimed extends AbstractObservableWithUpstre
final long period;
final TimeUnit unit;
final Scheduler scheduler;
-
+ final Consumer onDropped;
final boolean emitLast;
- public ObservableSampleTimed(ObservableSource source, long period, TimeUnit unit, Scheduler scheduler, boolean emitLast) {
+ public ObservableSampleTimed(ObservableSource source,
+ long period,
+ TimeUnit unit,
+ Scheduler scheduler,
+ boolean emitLast,
+ Consumer onDropped) {
super(source);
this.period = period;
this.unit = unit;
this.scheduler = scheduler;
this.emitLast = emitLast;
+ this.onDropped = onDropped;
}
@Override
public void subscribeActual(Observer t) {
SerializedObserver serial = new SerializedObserver<>(t);
if (emitLast) {
- source.subscribe(new SampleTimedEmitLast<>(serial, period, unit, scheduler));
+ source.subscribe(new SampleTimedEmitLast<>(serial, period, unit, scheduler, onDropped));
} else {
- source.subscribe(new SampleTimedNoLast<>(serial, period, unit, scheduler));
+ source.subscribe(new SampleTimedNoLast<>(serial, period, unit, scheduler, onDropped));
}
}
@@ -54,16 +62,18 @@ abstract static class SampleTimedObserver extends AtomicReference implemen
final long period;
final TimeUnit unit;
final Scheduler scheduler;
+ final Consumer onDropped;
final AtomicReference timer = new AtomicReference<>();
Disposable upstream;
- SampleTimedObserver(Observer actual, long period, TimeUnit unit, Scheduler scheduler) {
+ SampleTimedObserver(Observer actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) {
this.downstream = actual;
this.period = period;
this.unit = unit;
this.scheduler = scheduler;
+ this.onDropped = onDropped;
}
@Override
@@ -79,7 +89,17 @@ public void onSubscribe(Disposable d) {
@Override
public void onNext(T t) {
- lazySet(t);
+ T oldValue = getAndSet(t);
+ if (oldValue != null && onDropped != null) {
+ try {
+ onDropped.accept(oldValue);
+ } catch (Throwable throwable) {
+ Exceptions.throwIfFatal(throwable);
+ cancelTimer();
+ upstream.dispose();
+ downstream.onError(throwable);
+ }
+ }
}
@Override
@@ -123,8 +143,8 @@ static final class SampleTimedNoLast extends SampleTimedObserver {
private static final long serialVersionUID = -7139995637533111443L;
- SampleTimedNoLast(Observer actual, long period, TimeUnit unit, Scheduler scheduler) {
- super(actual, period, unit, scheduler);
+ SampleTimedNoLast(Observer actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) {
+ super(actual, period, unit, scheduler, onDropped);
}
@Override
@@ -144,8 +164,8 @@ static final class SampleTimedEmitLast extends SampleTimedObserver {
final AtomicInteger wip;
- SampleTimedEmitLast(Observer actual, long period, TimeUnit unit, Scheduler scheduler) {
- super(actual, period, unit, scheduler);
+ SampleTimedEmitLast(Observer actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) {
+ super(actual, period, unit, scheduler, onDropped);
this.wip = new AtomicInteger(1);
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java
index 889c203dd7..6bf3b9f119 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java
@@ -19,6 +19,8 @@
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.core.Scheduler.Worker;
import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
import io.reactivex.rxjava3.observers.SerializedObserver;
@@ -26,20 +28,27 @@ public final class ObservableThrottleFirstTimed extends AbstractObservableWit
final long timeout;
final TimeUnit unit;
final Scheduler scheduler;
-
- public ObservableThrottleFirstTimed(ObservableSource source,
- long timeout, TimeUnit unit, Scheduler scheduler) {
+ final Consumer onDropped;
+
+ public ObservableThrottleFirstTimed(
+ ObservableSource source,
+ long timeout,
+ TimeUnit unit,
+ Scheduler scheduler,
+ Consumer onDropped) {
super(source);
this.timeout = timeout;
this.unit = unit;
this.scheduler = scheduler;
+ this.onDropped = onDropped;
}
@Override
public void subscribeActual(Observer t) {
source.subscribe(new DebounceTimedObserver<>(
new SerializedObserver<>(t),
- timeout, unit, scheduler.createWorker()));
+ timeout, unit, scheduler.createWorker(),
+ onDropped));
}
static final class DebounceTimedObserver
@@ -51,16 +60,21 @@ static final class DebounceTimedObserver
final long timeout;
final TimeUnit unit;
final Scheduler.Worker worker;
-
+ final Consumer onDropped;
Disposable upstream;
-
volatile boolean gate;
- DebounceTimedObserver(Observer actual, long timeout, TimeUnit unit, Worker worker) {
+ DebounceTimedObserver(
+ Observer actual,
+ long timeout,
+ TimeUnit unit,
+ Worker worker,
+ Consumer onDropped) {
this.downstream = actual;
this.timeout = timeout;
this.unit = unit;
this.worker = worker;
+ this.onDropped = onDropped;
}
@Override
@@ -83,6 +97,15 @@ public void onNext(T t) {
d.dispose();
}
DisposableHelper.replace(this, worker.schedule(this, timeout, unit));
+ } else if (onDropped != null) {
+ try {
+ onDropped.accept(t);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ upstream.dispose();
+ downstream.onError(ex);
+ worker.dispose();
+ }
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleLatest.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleLatest.java
index 8f0b2cc250..caf14d3a5e 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleLatest.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleLatest.java
@@ -18,7 +18,10 @@
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.*;
+import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
+import io.reactivex.rxjava3.plugins.RxJavaPlugins;
/**
* Emits the next or latest item when the given time elapses.
@@ -41,19 +44,24 @@ public final class ObservableThrottleLatest extends AbstractObservableWithUps
final boolean emitLast;
+ final Consumer onDropped;
+
public ObservableThrottleLatest(Observable source,
- long timeout, TimeUnit unit, Scheduler scheduler,
- boolean emitLast) {
+ long timeout, TimeUnit unit,
+ Scheduler scheduler,
+ boolean emitLast,
+ Consumer onDropped) {
super(source);
this.timeout = timeout;
this.unit = unit;
this.scheduler = scheduler;
this.emitLast = emitLast;
+ this.onDropped = onDropped;
}
@Override
protected void subscribeActual(Observer observer) {
- source.subscribe(new ThrottleLatestObserver<>(observer, timeout, unit, scheduler.createWorker(), emitLast));
+ source.subscribe(new ThrottleLatestObserver<>(observer, timeout, unit, scheduler.createWorker(), emitLast, onDropped));
}
static final class ThrottleLatestObserver
@@ -74,6 +82,8 @@ static final class ThrottleLatestObserver
final AtomicReference latest;
+ final Consumer onDropped;
+
Disposable upstream;
volatile boolean done;
@@ -86,14 +96,17 @@ static final class ThrottleLatestObserver
boolean timerRunning;
ThrottleLatestObserver(Observer downstream,
- long timeout, TimeUnit unit, Scheduler.Worker worker,
- boolean emitLast) {
+ long timeout, TimeUnit unit,
+ Scheduler.Worker worker,
+ boolean emitLast,
+ Consumer onDropped) {
this.downstream = downstream;
this.timeout = timeout;
this.unit = unit;
this.worker = worker;
this.emitLast = emitLast;
this.latest = new AtomicReference<>();
+ this.onDropped = onDropped;
}
@Override
@@ -106,7 +119,17 @@ public void onSubscribe(Disposable d) {
@Override
public void onNext(T t) {
- latest.set(t);
+ T old = latest.getAndSet(t);
+ if (onDropped != null && old != null) {
+ try {
+ onDropped.accept(old);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ upstream.dispose();
+ error = ex;
+ done = true;
+ }
+ }
drain();
}
@@ -129,6 +152,22 @@ public void dispose() {
upstream.dispose();
worker.dispose();
if (getAndIncrement() == 0) {
+ clear();
+ }
+ }
+
+ void clear() {
+ if (onDropped != null) {
+ T v = latest.getAndSet(null);
+ if (v != null) {
+ try {
+ onDropped.accept(v);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ RxJavaPlugins.onError(ex);
+ }
+ }
+ } else {
latest.lazySet(null);
}
}
@@ -158,14 +197,27 @@ void drain() {
for (;;) {
if (cancelled) {
- latest.lazySet(null);
+ clear();
return;
}
boolean d = done;
+ Throwable error = this.error;
if (d && error != null) {
- latest.lazySet(null);
+ if (onDropped != null) {
+ T v = latest.getAndSet(null);
+ if (v != null) {
+ try {
+ onDropped.accept(v);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ error = new CompositeException(error, ex);
+ }
+ }
+ } else {
+ latest.lazySet(null);
+ }
downstream.onError(error);
worker.dispose();
return;
@@ -175,9 +227,22 @@ void drain() {
boolean empty = v == null;
if (d) {
- v = latest.getAndSet(null);
- if (!empty && emitLast) {
- downstream.onNext(v);
+ if (!empty) {
+ v = latest.getAndSet(null);
+ if (emitLast) {
+ downstream.onNext(v);
+ } else {
+ if (onDropped != null) {
+ try {
+ onDropped.accept(v);
+ } catch (Throwable ex) {
+ Exceptions.throwIfFatal(ex);
+ downstream.onError(ex);
+ worker.dispose();
+ return;
+ }
+ }
+ }
}
downstream.onComplete();
worker.dispose();
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelFromPublisher.java b/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelFromPublisher.java
index 67c2ed4852..eb57bccea5 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelFromPublisher.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelFromPublisher.java
@@ -204,7 +204,7 @@ public void onNext(T t) {
if (sourceMode == QueueSubscription.NONE) {
if (!queue.offer(t)) {
upstream.cancel();
- onError(new MissingBackpressureException("Queue is full?"));
+ onError(new QueueOverflowException());
return;
}
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelJoin.java b/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelJoin.java
index 3152aaeab8..2852fda8c7 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelJoin.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelJoin.java
@@ -18,7 +18,7 @@
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.exceptions.MissingBackpressureException;
+import io.reactivex.rxjava3.exceptions.*;
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
import io.reactivex.rxjava3.internal.util.*;
import io.reactivex.rxjava3.operators.SimplePlainQueue;
@@ -153,7 +153,7 @@ public void onNext(JoinInnerSubscriber inner, T value) {
if (!q.offer(value)) {
cancelAll();
- Throwable mbe = new MissingBackpressureException("Queue full?!");
+ Throwable mbe = new QueueOverflowException();
if (errors.compareAndSet(null, mbe)) {
downstream.onError(mbe);
} else {
@@ -170,7 +170,7 @@ public void onNext(JoinInnerSubscriber inner, T value) {
if (!q.offer(value)) {
cancelAll();
- onError(new MissingBackpressureException("Queue full?!"));
+ onError(new QueueOverflowException());
return;
}
@@ -333,7 +333,7 @@ void onNext(JoinInnerSubscriber inner, T value) {
if (!q.offer(value)) {
inner.cancel();
- errors.tryAddThrowableOrReport(new MissingBackpressureException("Queue full?!"));
+ errors.tryAddThrowableOrReport(new QueueOverflowException());
done.decrementAndGet();
drainLoop();
return;
@@ -347,7 +347,7 @@ void onNext(JoinInnerSubscriber inner, T value) {
if (!q.offer(value)) {
inner.cancel();
- errors.tryAddThrowableOrReport(new MissingBackpressureException("Queue full?!"));
+ errors.tryAddThrowableOrReport(new QueueOverflowException());
done.decrementAndGet();
}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelRunOn.java b/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelRunOn.java
index e9c92a0c24..22f822db34 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelRunOn.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/parallel/ParallelRunOn.java
@@ -19,7 +19,7 @@
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.core.Scheduler.Worker;
-import io.reactivex.rxjava3.exceptions.MissingBackpressureException;
+import io.reactivex.rxjava3.exceptions.*;
import io.reactivex.rxjava3.internal.schedulers.SchedulerMultiWorkerSupport;
import io.reactivex.rxjava3.internal.schedulers.SchedulerMultiWorkerSupport.WorkerCallback;
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
@@ -148,7 +148,7 @@ public final void onNext(T t) {
}
if (!queue.offer(t)) {
upstream.cancel();
- onError(new MissingBackpressureException("Queue is full?!"));
+ onError(new QueueOverflowException());
return;
}
schedule();
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeout.java b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeout.java
index 609747d1d8..cc5b923727 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeout.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeout.java
@@ -113,11 +113,7 @@ public void onError(Throwable e) {
@Override
public void run() {
- Disposable d = get();
- if (d != DisposableHelper.DISPOSED && compareAndSet(d, DisposableHelper.DISPOSED)) {
- if (d != null) {
- d.dispose();
- }
+ if (DisposableHelper.dispose(this)) {
SingleSource other = this.other;
if (other == null) {
downstream.onError(new TimeoutException(timeoutMessage(timeout, unit)));
diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadWorker.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadWorker.java
index deba4948ba..764241e76e 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadWorker.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadWorker.java
@@ -26,7 +26,7 @@
* worker but doesn't perform task-tracking operations.
*
*/
-public class NewThreadWorker extends Scheduler.Worker implements Disposable {
+public class NewThreadWorker extends Scheduler.Worker {
private final ScheduledExecutorService executor;
volatile boolean disposed;
diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/TrampolineScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/TrampolineScheduler.java
index 7ed777f08f..04496482b0 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/TrampolineScheduler.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/TrampolineScheduler.java
@@ -62,7 +62,7 @@ public Disposable scheduleDirect(@NonNull Runnable run, long delay, TimeUnit uni
return EmptyDisposable.INSTANCE;
}
- static final class TrampolineWorker extends Scheduler.Worker implements Disposable {
+ static final class TrampolineWorker extends Scheduler.Worker {
final PriorityBlockingQueue queue = new PriorityBlockingQueue<>();
private final AtomicInteger wip = new AtomicInteger();
diff --git a/src/main/java/io/reactivex/rxjava3/internal/subscribers/QueueDrainSubscriber.java b/src/main/java/io/reactivex/rxjava3/internal/subscribers/QueueDrainSubscriber.java
index 145882e418..8fb7f55295 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/subscribers/QueueDrainSubscriber.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/subscribers/QueueDrainSubscriber.java
@@ -84,7 +84,7 @@ protected final void fastPathEmitMax(U value, boolean delayError, Disposable dis
}
} else {
dispose.dispose();
- s.onError(new MissingBackpressureException("Could not emit buffer due to lack of requests"));
+ s.onError(MissingBackpressureException.createDefault());
return;
}
} else {
@@ -118,7 +118,7 @@ protected final void fastPathOrderedEmitMax(U value, boolean delayError, Disposa
} else {
cancelled = true;
dispose.dispose();
- s.onError(new MissingBackpressureException("Could not emit buffer due to lack of requests"));
+ s.onError(MissingBackpressureException.createDefault());
return;
}
} else {
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 a9a2ab4901..fa0c500892 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/util/QueueDrainHelper.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/util/QueueDrainHelper.java
@@ -78,7 +78,7 @@ public static void drainMaxLoop(SimplePlainQueue q, Subscriber
* If the provided executor doesn't support any of the more specific standard Java executor
- * APIs, cancelling tasks scheduled by this scheduler can't be interrupted when they are
+ * APIs, tasks scheduled by this scheduler can't be interrupted when they are
* executing but only prevented from running prior to that. In addition, tasks scheduled with
* a time delay or periodically will use the {@link #single()} scheduler for the timed waiting
* before posting the actual task to the given executor.
@@ -327,13 +327,13 @@ public static Scheduler single() {
* {@link #from(Executor, boolean)} overload to enable task interruption via this wrapper.
*
* If the provided executor supports the standard Java {@link ExecutorService} API,
- * cancelling tasks scheduled by this scheduler can be cancelled/interrupted by calling
+ * tasks scheduled by this scheduler can be cancelled/interrupted by calling
* {@link io.reactivex.rxjava3.disposables.Disposable#dispose()}. In addition, tasks scheduled with
* a time delay or periodically will use the {@link #single()} scheduler for the timed waiting
* before posting the actual task to the given executor.
*
* If the provided executor supports the standard Java {@link ScheduledExecutorService} API,
- * cancelling tasks scheduled by this scheduler can be cancelled/interrupted by calling
+ * tasks scheduled by this scheduler can be cancelled/interrupted by calling
* {@link io.reactivex.rxjava3.disposables.Disposable#dispose()}. In addition, tasks scheduled with
* a time delay or periodically will use the provided executor. Note, however, if the provided
* {@code ScheduledExecutorService} instance is not single threaded, tasks scheduled
@@ -405,13 +405,13 @@ public static Scheduler from(@NonNull Executor executor) {
* before posting the actual task to the given executor.
*
* If the provided executor supports the standard Java {@link ExecutorService} API,
- * canceling tasks scheduled by this scheduler can be cancelled/interrupted by calling
+ * tasks scheduled by this scheduler can be cancelled/interrupted by calling
* {@link io.reactivex.rxjava3.disposables.Disposable#dispose()}. In addition, tasks scheduled with
* a time delay or periodically will use the {@link #single()} scheduler for the timed waiting
* before posting the actual task to the given executor.
*
* If the provided executor supports the standard Java {@link ScheduledExecutorService} API,
- * canceling tasks scheduled by this scheduler can be cancelled/interrupted by calling
+ * tasks scheduled by this scheduler can be cancelled/interrupted by calling
* {@link io.reactivex.rxjava3.disposables.Disposable#dispose()}. In addition, tasks scheduled with
* a time delay or periodically will use the provided executor. Note, however, if the provided
* {@code ScheduledExecutorService} instance is not single threaded, tasks scheduled
@@ -487,13 +487,13 @@ public static Scheduler from(@NonNull Executor executor, boolean interruptibleWo
* before posting the actual task to the given executor.
*
* If the provided executor supports the standard Java {@link ExecutorService} API,
- * canceling tasks scheduled by this scheduler can be cancelled/interrupted by calling
+ * tasks scheduled by this scheduler can be cancelled/interrupted by calling
* {@link io.reactivex.rxjava3.disposables.Disposable#dispose()}. In addition, tasks scheduled with
* a time delay or periodically will use the {@link #single()} scheduler for the timed waiting
* before posting the actual task to the given executor.
*
* If the provided executor supports the standard Java {@link ScheduledExecutorService} API,
- * canceling tasks scheduled by this scheduler can be cancelled/interrupted by calling
+ * tasks scheduled by this scheduler can be cancelled/interrupted by calling
* {@link io.reactivex.rxjava3.disposables.Disposable#dispose()}. In addition, tasks scheduled with
* a time delay or periodically will use the provided executor. Note, however, if the provided
* {@code ScheduledExecutorService} instance is not single threaded, tasks scheduled
diff --git a/src/main/java/io/reactivex/rxjava3/subjects/BehaviorSubject.java b/src/main/java/io/reactivex/rxjava3/subjects/BehaviorSubject.java
index 465ddcbebc..2b19ecdd26 100644
--- a/src/main/java/io/reactivex/rxjava3/subjects/BehaviorSubject.java
+++ b/src/main/java/io/reactivex/rxjava3/subjects/BehaviorSubject.java
@@ -59,9 +59,9 @@
*
* TestObserver<Integer> to1 = observable.test();
*
- * observable.onNext(1);
+ * subject.onNext(1);
* // this will "clear" the cache
- * observable.onNext(EMPTY);
+ * subject.onNext(EMPTY);
*
* TestObserver<Integer> to2 = observable.test();
*
diff --git a/src/test/java/io/reactivex/rxjava3/flowable/FlowableBackpressureTests.java b/src/test/java/io/reactivex/rxjava3/flowable/FlowableBackpressureTests.java
index 07ae9c0f27..975ef5c149 100644
--- a/src/test/java/io/reactivex/rxjava3/flowable/FlowableBackpressureTests.java
+++ b/src/test/java/io/reactivex/rxjava3/flowable/FlowableBackpressureTests.java
@@ -24,7 +24,7 @@
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.exceptions.MissingBackpressureException;
+import io.reactivex.rxjava3.exceptions.QueueOverflowException;
import io.reactivex.rxjava3.functions.*;
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
import io.reactivex.rxjava3.internal.util.BackpressureHelper;
@@ -475,7 +475,7 @@ public Integer apply(Integer v) {
int vc = ts.values().size();
assertTrue("10 < " + vc, vc <= 10);
- ts.assertError(MissingBackpressureException.class);
+ ts.assertError(QueueOverflowException.class);
}
@Test
diff --git a/src/test/java/io/reactivex/rxjava3/flowable/FlowableThrottleLastTests.java b/src/test/java/io/reactivex/rxjava3/flowable/FlowableThrottleLastTests.java
index 0d2dc343fc..c5c858c430 100644
--- a/src/test/java/io/reactivex/rxjava3/flowable/FlowableThrottleLastTests.java
+++ b/src/test/java/io/reactivex/rxjava3/flowable/FlowableThrottleLastTests.java
@@ -13,10 +13,14 @@
package io.reactivex.rxjava3.flowable;
-import static org.mockito.Mockito.inOrder;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
import java.util.concurrent.TimeUnit;
+import io.reactivex.rxjava3.core.Observer;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.functions.Action;
import org.junit.Test;
import org.mockito.InOrder;
import org.reactivestreams.Subscriber;
@@ -28,6 +32,72 @@
public class FlowableThrottleLastTests extends RxJavaTest {
+ @Test
+ public void throttleWithDroppedCallbackException() throws Throwable {
+ Subscriber subscriber = TestHelper.mockSubscriber();
+ Action whenDisposed = mock(Action.class);
+
+ TestScheduler s = new TestScheduler();
+ PublishProcessor o = PublishProcessor.create();
+ o.doOnCancel(whenDisposed)
+ .throttleLast(500, TimeUnit.MILLISECONDS, s, e-> {
+ if (e == 1) {
+ throw new TestException("forced");
+ }
+ })
+ .subscribe(subscriber);
+
+ // send events with simulated time increments
+ s.advanceTimeTo(0, TimeUnit.MILLISECONDS);
+ o.onNext(1); // skip
+ o.onNext(2); // deliver
+ s.advanceTimeTo(501, TimeUnit.MILLISECONDS);
+
+ InOrder inOrder = inOrder(subscriber);
+ inOrder.verify(subscriber).onError(any(TestException.class));
+ inOrder.verifyNoMoreInteractions();
+ verify(whenDisposed).run();
+ }
+
+ @Test
+ public void throttleWithDroppedCallback() {
+ Subscriber subscriber = TestHelper.mockSubscriber();
+ Observer