diff --git a/src/main/java/io/reactivex/rxjava3/core/Flowable.java b/src/main/java/io/reactivex/rxjava3/core/Flowable.java index 4c4236f29fa..f866ed5302d 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Flowable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Flowable.java @@ -14729,7 +14729,6 @@ public final Flowable sample(long period, @NonNull TimeUnit unit, @NonNull Sc *
You specify which {@code Scheduler} this operator will use.
* * - *

History: 2.0.5 - experimental * @param period * the sampling rate * @param unit @@ -14747,13 +14746,14 @@ public final Flowable sample(long period, @NonNull TimeUnit unit, @NonNull Sc * @see ReactiveX operators documentation: Sample * @see RxJava wiki: Backpressure * @see #throttleLast(long, TimeUnit, Scheduler) - * @since 2.1 + * @since 3.1.6 - Experimental */ @CheckReturnValue @NonNull @BackpressureSupport(BackpressureKind.ERROR) @SchedulerSupport(SchedulerSupport.CUSTOM) - public final Flowable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast, @NonNull Consumer onDropped) { + @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"); @@ -17172,11 +17172,13 @@ public final Flowable throttleFirst(long skipDuration, @NonNull TimeUnit unit * @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"); @@ -17285,12 +17287,14 @@ public final Flowable throttleLast(long intervalDuration, @NonNull TimeUnit u * @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 - public final Flowable throttleLast(long intervalDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) { + @Experimental + public final Flowable throttleLast(long intervalDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) { return sample(intervalDuration, unit, scheduler, false, onDropped); } diff --git a/src/main/java/io/reactivex/rxjava3/core/Observable.java b/src/main/java/io/reactivex/rxjava3/core/Observable.java index f371fda2593..4216d1c7706 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Observable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Observable.java @@ -12133,49 +12133,51 @@ public final Observable sample(long period, @NonNull TimeUnit unit, @NonNull /** * 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}. + * 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. *

- * + * *

*
Scheduler:
*
You specify which {@code Scheduler} this operator will use.
*
* + *

History: 2.0.5 - experimental * @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 onDropped - * called with the current entry when it has been replaced by a new one + * @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. * @return the new {@code Observable} instance - * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null} + * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} * @see ReactiveX operators documentation: Sample * @see #throttleLast(long, TimeUnit, Scheduler) + * @since 2.1 */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.CUSTOM) @NonNull - public final Observable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) { + 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"); - Objects.requireNonNull(onDropped, "onDropped is null"); - return RxJavaPlugins.onAssembly(new ObservableSampleTimed<>(this, period, unit, scheduler, false, onDropped)); + 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} - * and optionally emit the very last upstream item when the upstream completes. + * within periodic time intervals, where the intervals are defined on a particular {@link Scheduler}. *

- * + * *

*
Scheduler:
*
You specify which {@code Scheduler} this operator will use.
*
* - *

History: 2.0.5 - experimental * @param period * the sampling rate * @param unit @@ -12186,19 +12188,23 @@ public final Observable sample(long period, @NonNull TimeUnit unit, @NonNull * 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} + * @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 2.1 + * @since 3.1.6 - Experimental */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.CUSTOM) @NonNull - public final Observable sample(long period, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean emitLast) { + @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"); - return RxJavaPlugins.onAssembly(new ObservableSampleTimed<>(this, period, unit, scheduler, emitLast, null)); + Objects.requireNonNull(onDropped, "onDropped is null"); + return RxJavaPlugins.onAssembly(new ObservableSampleTimed<>(this, period, unit, scheduler, emitLast, onDropped)); } /** @@ -14225,10 +14231,12 @@ public final Observable throttleFirst(long skipDuration, @NonNull TimeUnit un * @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"); @@ -14293,12 +14301,14 @@ public final Observable throttleLast(long intervalDuration, @NonNull TimeUnit * @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 - public final Observable throttleLast(long intervalDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) { - return sample(intervalDuration, unit, scheduler, onDropped); + @Experimental + public final Observable throttleLast(long intervalDuration, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer onDropped) { + return sample(intervalDuration, unit, scheduler, false, onDropped); } /** 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 59dce502019..40551f4a8ed 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 @@ -32,9 +32,9 @@ public final class FlowableSampleTimed extends AbstractFlowableWithUpstream onDropped; + final Consumer onDropped; - public FlowableSampleTimed(Flowable source, long period, TimeUnit unit, Scheduler scheduler, boolean emitLast, Consumer onDropped) { + public FlowableSampleTimed(Flowable source, long period, TimeUnit unit, Scheduler scheduler, boolean emitLast, Consumer onDropped) { super(source); this.period = period; this.unit = unit; @@ -61,7 +61,7 @@ abstract static class SampleTimedSubscriber extends AtomicReference implem final long period; final TimeUnit unit; final Scheduler scheduler; - final Consumer onDropped; + final Consumer onDropped; final AtomicLong requested = new AtomicLong(); @@ -69,7 +69,7 @@ abstract static class SampleTimedSubscriber extends AtomicReference implem Subscription upstream; - SampleTimedSubscriber(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { + SampleTimedSubscriber(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { this.downstream = actual; this.period = period; this.unit = unit; @@ -152,7 +152,7 @@ static final class SampleTimedNoLast extends SampleTimedSubscriber { private static final long serialVersionUID = -7139995637533111443L; - SampleTimedNoLast(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { + SampleTimedNoLast(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { super(actual, period, unit, scheduler, onDropped); } @@ -173,7 +173,7 @@ static final class SampleTimedEmitLast extends SampleTimedSubscriber { final AtomicInteger wip; - SampleTimedEmitLast(Subscriber actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { + 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/observable/ObservableSampleTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableSampleTimed.java index b2025060b9e..f264b8e76d4 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 @@ -22,13 +22,12 @@ 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; public final class ObservableSampleTimed extends AbstractObservableWithUpstream { final long period; final TimeUnit unit; final Scheduler scheduler; - final Consumer onDropped; + final Consumer onDropped; final boolean emitLast; public ObservableSampleTimed(ObservableSource source, @@ -36,7 +35,7 @@ public ObservableSampleTimed(ObservableSource source, TimeUnit unit, Scheduler scheduler, boolean emitLast, - Consumer onDropped) { + Consumer onDropped) { super(source); this.period = period; this.unit = unit; @@ -63,13 +62,13 @@ abstract static class SampleTimedObserver extends AtomicReference implemen final long period; final TimeUnit unit; final Scheduler scheduler; - final Consumer onDropped; + final Consumer onDropped; final AtomicReference timer = new AtomicReference<>(); Disposable upstream; - SampleTimedObserver(Observer actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { + SampleTimedObserver(Observer actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { this.downstream = actual; this.period = period; this.unit = unit; @@ -144,7 +143,7 @@ static final class SampleTimedNoLast extends SampleTimedObserver { private static final long serialVersionUID = -7139995637533111443L; - SampleTimedNoLast(Observer actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { + SampleTimedNoLast(Observer actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { super(actual, period, unit, scheduler, onDropped); } @@ -165,7 +164,7 @@ static final class SampleTimedEmitLast extends SampleTimedObserver { final AtomicInteger wip; - SampleTimedEmitLast(Observer actual, long period, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { + 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/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java b/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java index 21d8905ded2..ba377dbf639 100644 --- a/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java +++ b/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java @@ -437,8 +437,8 @@ public void checkParallelFlowable() { addOverride(new ParamOverride(Observable.class, 0, ParamMode.ANY, "sample", Long.TYPE, TimeUnit.class)); addOverride(new ParamOverride(Observable.class, 0, ParamMode.ANY, "sample", Long.TYPE, TimeUnit.class, Boolean.TYPE)); addOverride(new ParamOverride(Observable.class, 0, ParamMode.ANY, "sample", Long.TYPE, TimeUnit.class, Scheduler.class)); - addOverride(new ParamOverride(Observable.class, 0, ParamMode.ANY, "sample", Long.TYPE, TimeUnit.class, Scheduler.class, Consumer.class)); addOverride(new ParamOverride(Observable.class, 0, ParamMode.ANY, "sample", Long.TYPE, TimeUnit.class, Scheduler.class, Boolean.TYPE)); + addOverride(new ParamOverride(Observable.class, 0, ParamMode.ANY, "sample", Long.TYPE, TimeUnit.class, Scheduler.class, Boolean.TYPE, Consumer.class)); // negative time is considered as zero time addOverride(new ParamOverride(Observable.class, 0, ParamMode.ANY, "takeLast", Long.TYPE, TimeUnit.class));