diff --git a/docs/Operator-Matrix.md b/docs/Operator-Matrix.md
index 81e1f7878d..10427533cc 100644
--- a/docs/Operator-Matrix.md
+++ b/docs/Operator-Matrix.md
@@ -58,7 +58,7 @@ Operator | |||||
<a name='delay'></a>`delay`||||||
<a name='delaySubscription'></a>`delaySubscription`||||||
-<a name='dematerialize'></a>`dematerialize`||||| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
+<a name='dematerialize'></a>`dematerialize`||||| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
<a name='distinct'></a>`distinct`||| <sup title='At most one item, always distinct.'>([39](#notes-39))</sup>| <sup title='At most one item, always distinct.'>([39](#notes-39))</sup>| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
<a name='distinctUntilChanged'></a>`distinctUntilChanged`||| <sup title='At most one item, always distinct.'>([39](#notes-39))</sup>| <sup title='At most one item, always distinct.'>([39](#notes-39))</sup>| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
<a name='doAfterNext'></a>`doAfterNext`||| <sup title='Different terminology. Use doAfterSuccess().'>([40](#notes-40))</sup>| <sup title='Different terminology. Use doAfterSuccess().'>([40](#notes-40))</sup>| <sup title='Always empty.'>([2](#notes-2))</sup>|
@@ -212,10 +212,10 @@ Operator | || <sup title='At most one item signaled so no subsequent items to work with.'>([36](#notes-36))</sup>| <sup title='At most one item signaled so no subsequent items to work with.'>([36](#notes-36))</sup>| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
<a name='throttleLatest'></a>`throttleLatest`||| <sup title='At most one item signaled so no subsequent items to work with.'>([36](#notes-36))</sup>| <sup title='At most one item signaled so no subsequent items to work with.'>([36](#notes-36))</sup>| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
<a name='throttleWithTimeout'></a>`throttleWithTimeout`||| <sup title='At most one item signaled so no subsequent items to work with.'>([36](#notes-36))</sup>| <sup title='At most one item signaled so no subsequent items to work with.'>([36](#notes-36))</sup>| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
-<a name='timeInterval'></a>`timeInterval`||||||
+<a name='timeInterval'></a>`timeInterval`||||| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
<a name='timeout'></a>`timeout`||||||
<a name='timer'></a>`timer`||||||
-<a name='timestamp'></a>`timestamp`||||| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
+<a name='timestamp'></a>`timestamp`||||| <sup title='Always empty thus no items to work with.'>([37](#notes-37))</sup>|
<a name='to'></a>`to`||||||
<a name='toCompletionStage'></a>`toCompletionStage`| <sup title='Use firstStage.'>([98](#notes-98))</sup>| <sup title='Use firstStage.'>([98](#notes-98))</sup>||||
<a name='toFlowable'></a>`toFlowable`| <sup title='Would be no-op.'>([99](#notes-99))</sup>|||||
@@ -237,7 +237,7 @@ Operator | |||| <sup title='Use merge().'>([108](#notes-108))</sup>|
<a name='zipArray'></a>`zipArray`||||| <sup title='Use mergeArray().'>([109](#notes-109))</sup>|
<a name='zipWith'></a>`zipWith`||||| <sup title='Use mergeWith().'>([110](#notes-110))</sup>|
-<a name='total'></a>**237 operators** | **215** | **209** | **108** | **93** | **76** |
+<a name='total'></a>**237 operators** | **215** | **209** | **111** | **95** | **76** |
#### Notes
<a name='notes-1'></a><sup>1</sup> Use [`contains()`](#contains).<br/>
@@ -372,25 +372,19 @@ Operator | 
18. Maybe.concatMapSingle()
19. Single.concatMapSingle()
-20. Maybe.dematerialize()
-21. Maybe.doOnLifecycle()
-22. Single.doOnLifecycle()
-23. Completable.doOnLifecycle()
-24. Single.mergeArray()
-25. Single.mergeArrayDelayError()
-26. Single.ofType()
-27. Completable.onErrorReturn()
-28. Completable.onErrorReturnItem()
-29. Maybe.safeSubscribe()
-30. Single.safeSubscribe()
-31. Completable.safeSubscribe()
-32. Completable.sequenceEqual()
-33. Maybe.startWith()
-34. Single.startWith()
-35. Maybe.timeInterval()
-36. Single.timeInterval()
-37. Completable.timeInterval()
-38. Maybe.timestamp()
-39. Single.timestamp()
-40. Maybe.toFuture()
-41. Completable.toFuture()
+20. Maybe.doOnLifecycle()
+21. Single.doOnLifecycle()
+22. Completable.doOnLifecycle()
+23. Single.mergeArray()
+24. Single.mergeArrayDelayError()
+25. Single.ofType()
+26. Completable.onErrorReturn()
+27. Completable.onErrorReturnItem()
+28. Maybe.safeSubscribe()
+29. Single.safeSubscribe()
+30. Completable.safeSubscribe()
+31. Completable.sequenceEqual()
+32. Maybe.startWith()
+33. Single.startWith()
+34. Maybe.toFuture()
+35. Completable.toFuture()
diff --git a/src/main/java/io/reactivex/rxjava3/core/Maybe.java b/src/main/java/io/reactivex/rxjava3/core/Maybe.java
index 12ab3678e2..03e73abe65 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Maybe.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Maybe.java
@@ -34,7 +34,7 @@
import io.reactivex.rxjava3.internal.util.ErrorMode;
import io.reactivex.rxjava3.observers.TestObserver;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
-import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.schedulers.*;
/**
* The {@code Maybe} class represents a deferred computation and emission of a single value, no value at all or an exception.
@@ -4991,6 +4991,232 @@ public final <U> Maybe<T> takeUntil(@NonNull Publisher<U> other) {
return RxJavaPlugins.onAssembly(new MaybeTakeUntilPublisher<>(this, other));
}
+ /**
+ * Measures the time (in milliseconds) between the subscription and success item emission
+ * of the current {@code Maybe} and signals it as a tuple ({@link Timed})
+ * success value.
+ * <p>
+ * <img width="640" height="352" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.timeInterval.png" alt="">
+ * <p>
+ * If the current {@code Maybe} is empty or fails, the resulting {@code Maybe} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link Single#timeInterval()}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timeInterval} uses the {@code computation} {@link Scheduler}
+ * for determining the current time upon subscription and upon receiving the
+ * success item from the current {@code Maybe}.</dd>
+ * </dl>
+ * @return the new {@code Maybe} instance
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.COMPUTATION)
+ public final Maybe<Timed<T>> timeInterval() {
+ return timeInterval(TimeUnit.MILLISECONDS, Schedulers.computation());
+ }
+
+ /**
+ * Measures the time (in milliseconds) between the subscription and success item emission
+ * of the current {@code Maybe} and signals it as a tuple ({@link Timed})
+ * success value.
+ * <p>
+ * <img width="640" height="355" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.timeInterval.s.png" alt="">
+ * <p>
+ * If the current {@code Maybe} is empty or fails, the resulting {@code Maybe} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link Single#timeInterval(Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timeInterval} uses the provided {@link Scheduler}
+ * for determining the current time upon subscription and upon receiving the
+ * success item from the current {@code Maybe}.</dd>
+ * </dl>
+ * @param scheduler the {@code Scheduler} used for providing the current time
+ * @return the new {@code Maybe} instance
+ * @throws NullPointerException if {@code scheduler} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ public final Maybe<Timed<T>> timeInterval(@NonNull Scheduler scheduler) {
+ return timeInterval(TimeUnit.MILLISECONDS, scheduler);
+ }
+
+ /**
+ * Measures the time between the subscription and success item emission
+ * of the current {@code Maybe} and signals it as a tuple ({@link Timed})
+ * success value.
+ * <p>
+ * <img width="640" height="352" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.timeInterval.png" alt="">
+ * <p>
+ * If the current {@code Maybe} is empty or fails, the resulting {@code Maybe} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link Single#timeInterval(TimeUnit)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timeInterval} uses the {@code computation} {@link Scheduler}
+ * for determining the current time upon subscription and upon receiving the
+ * success item from the current {@code Maybe}.</dd>
+ * </dl>
+ * @param unit the time unit for measurement
+ * @return the new {@code Maybe} instance
+ * @throws NullPointerException if {@code unit} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.COMPUTATION)
+ public final Maybe<Timed<T>> timeInterval(@NonNull TimeUnit unit) {
+ return timeInterval(unit, Schedulers.computation());
+ }
+
+ /**
+ * Measures the time between the subscription and success item emission
+ * of the current {@code Maybe} and signals it as a tuple ({@link Timed})
+ * success value.
+ * <p>
+ * <img width="640" height="355" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.timeInterval.s.png" alt="">
+ * <p>
+ * If the current {@code Maybe} is empty or fails, the resulting {@code Maybe} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link Single#timeInterval(TimeUnit, Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timeInterval} uses the provided {@link Scheduler}
+ * for determining the current time upon subscription and upon receiving the
+ * success item from the current {@code Maybe}.</dd>
+ * </dl>
+ * @param unit the time unit for measurement
+ * @param scheduler the {@code Scheduler} used for providing the current time
+ * @return the new {@code Maybe} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ public final Maybe<Timed<T>> timeInterval(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ return RxJavaPlugins.onAssembly(new MaybeTimeInterval<>(this, unit, scheduler, true));
+ }
+
+ /**
+ * Combines the success value from the current {@code Maybe} with the current time (in milliseconds) of
+ * its reception, using the {@code computation} {@link Scheduler} as time source,
+ * then signals them as a {@link Timed} instance.
+ * <p>
+ * <img width="640" height="352" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.timestamp.png" alt="">
+ * <p>
+ * If the current {@code Maybe} is empty or fails, the resulting {@code Maybe} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link Single#timestamp()}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timestamp} uses the {@code computation} {@code Scheduler}
+ * for determining the current time upon receiving the
+ * success item from the current {@code Maybe}.</dd>
+ * </dl>
+ * @return the new {@code Maybe} instance
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.COMPUTATION)
+ public final Maybe<Timed<T>> timestamp() {
+ return timestamp(TimeUnit.MILLISECONDS, Schedulers.computation());
+ }
+
+ /**
+ * Combines the success value from the current {@code Maybe} with the current time (in milliseconds) of
+ * its reception, using the given {@link Scheduler} as time source,
+ * then signals them as a {@link Timed} instance.
+ * <p>
+ * <img width="640" height="355" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.timestamp.s.png" alt="">
+ * <p>
+ * If the current {@code Maybe} is empty or fails, the resulting {@code Maybe} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link Single#timestamp(Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timestamp} uses the provided {@code Scheduler}
+ * for determining the current time upon receiving the
+ * success item from the current {@code Maybe}.</dd>
+ * </dl>
+ * @param scheduler the {@code Scheduler} used for providing the current time
+ * @return the new {@code Maybe} instance
+ * @throws NullPointerException if {@code scheduler} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ public final Maybe<Timed<T>> timestamp(@NonNull Scheduler scheduler) {
+ return timestamp(TimeUnit.MILLISECONDS, scheduler);
+ }
+
+ /**
+ * Combines the success value from the current {@code Maybe} with the current time of
+ * its reception, using the {@code computation} {@link Scheduler} as time source,
+ * then signals it as a {@link Timed} instance.
+ * <p>
+ * <img width="640" height="352" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.timestamp.png" alt="">
+ * <p>
+ * If the current {@code Maybe} is empty or fails, the resulting {@code Maybe} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link Single#timestamp(TimeUnit)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timestamp} uses the {@code computation} {@code Scheduler},
+ * for determining the current time upon receiving the
+ * success item from the current {@code Maybe}.</dd>
+ * </dl>
+ * @param unit the time unit for measurement
+ * @return the new {@code Maybe} instance
+ * @throws NullPointerException if {@code unit} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.COMPUTATION)
+ public final Maybe<Timed<T>> timestamp(@NonNull TimeUnit unit) {
+ return timestamp(unit, Schedulers.computation());
+ }
+
+ /**
+ * Combines the success value from the current {@code Maybe} with the current time of
+ * its reception, using the given {@link Scheduler} as time source,
+ * then signals it as a {@link Timed} instance.
+ * <p>
+ * <img width="640" height="355" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.timestamp.s.png" alt="">
+ * <p>
+ * If the current {@code Maybe} is empty or fails, the resulting {@code Maybe} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link Single#timestamp(TimeUnit, Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timestamp} uses the provided {@code Scheduler},
+ * which is used for determining the current time upon receiving the
+ * success item from the current {@code Maybe}.</dd>
+ * </dl>
+ * @param unit the time unit for measurement
+ * @param scheduler the {@code Scheduler} used for providing the current time
+ * @return the new {@code Maybe} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ public final Maybe<Timed<T>> timestamp(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ return RxJavaPlugins.onAssembly(new MaybeTimeInterval<>(this, unit, scheduler, false));
+ }
+
/**
* Returns a {@code Maybe} that mirrors the current {@code Maybe} but applies a timeout policy for each emitted
* item. If the next item isn't emitted within the specified timeout duration starting from its predecessor,
diff --git a/src/main/java/io/reactivex/rxjava3/core/Single.java b/src/main/java/io/reactivex/rxjava3/core/Single.java
index a3b536b1f7..697c1355b5 100644
--- a/src/main/java/io/reactivex/rxjava3/core/Single.java
+++ b/src/main/java/io/reactivex/rxjava3/core/Single.java
@@ -36,7 +36,7 @@
import io.reactivex.rxjava3.internal.util.ErrorMode;
import io.reactivex.rxjava3.observers.TestObserver;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
-import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.schedulers.*;
/**
* The {@code Single} class implements the Reactive Pattern for a single value response.
@@ -4137,6 +4137,232 @@ public final Single<T> subscribeOn(@NonNull Scheduler scheduler) {
return RxJavaPlugins.onAssembly(new SingleSubscribeOn<>(this, scheduler));
}
+ /**
+ * Measures the time (in milliseconds) between the subscription and success item emission
+ * of the current {@code Single} and signals it as a tuple ({@link Timed})
+ * success value.
+ * <p>
+ * <img width="640" height="466" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.timeInterval.png" alt="">
+ * <p>
+ * If the current {@code Single} fails, the resulting {@code Single} will
+ * pass along the signal to the downstream. To measure the time to error,
+ * use {@link #materialize()} and apply {@link #timeInterval()}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timeInterval} uses the {@code computation} {@link Scheduler}
+ * for determining the current time upon subscription and upon receiving the
+ * success item from the current {@code Single}.</dd>
+ * </dl>
+ * @return the new {@code Single} instance
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.COMPUTATION)
+ public final Single<Timed<T>> timeInterval() {
+ return timeInterval(TimeUnit.MILLISECONDS, Schedulers.computation());
+ }
+
+ /**
+ * Measures the time (in milliseconds) between the subscription and success item emission
+ * of the current {@code Single} and signals it as a tuple ({@link Timed})
+ * success value.
+ * <p>
+ * <img width="640" height="463" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.timeInterval.s.png" alt="">
+ * <p>
+ * If the current {@code Single} fails, the resulting {@code Single} will
+ * pass along the signal to the downstream. To measure the time to error,
+ * use {@link #materialize()} and apply {@link #timeInterval(Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timeInterval} uses the provided {@link Scheduler}
+ * for determining the current time upon subscription and upon receiving the
+ * success item from the current {@code Single}.</dd>
+ * </dl>
+ * @param scheduler the {@code Scheduler} used for providing the current time
+ * @return the new {@code Single} instance
+ * @throws NullPointerException if {@code scheduler} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ public final Single<Timed<T>> timeInterval(@NonNull Scheduler scheduler) {
+ return timeInterval(TimeUnit.MILLISECONDS, scheduler);
+ }
+
+ /**
+ * Measures the time between the subscription and success item emission
+ * of the current {@code Single} and signals it as a tuple ({@link Timed})
+ * success value.
+ * <p>
+ * <img width="640" height="466" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.timeInterval.png" alt="">
+ * <p>
+ * If the current {@code Single} fails, the resulting {@code Single} will
+ * pass along the signals to the downstream. To measure the time to error,
+ * use {@link #materialize()} and apply {@link #timeInterval(TimeUnit, Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timeInterval} uses the {@code computation} {@link Scheduler}
+ * for determining the current time upon subscription and upon receiving the
+ * success item from the current {@code Single}.</dd>
+ * </dl>
+ * @param unit the time unit for measurement
+ * @return the new {@code Single} instance
+ * @throws NullPointerException if {@code unit} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.COMPUTATION)
+ public final Single<Timed<T>> timeInterval(@NonNull TimeUnit unit) {
+ return timeInterval(unit, Schedulers.computation());
+ }
+
+ /**
+ * Measures the time between the subscription and success item emission
+ * of the current {@code Single} and signals it as a tuple ({@link Timed})
+ * success value.
+ * <p>
+ * <img width="640" height="463" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.timeInterval.s.png" alt="">
+ * <p>
+ * If the current {@code Single} is empty or fails, the resulting {@code Single} will
+ * pass along the signals to the downstream. To measure the time to termination,
+ * use {@link #materialize()} and apply {@link #timeInterval(TimeUnit, Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timeInterval} uses the provided {@link Scheduler}
+ * for determining the current time upon subscription and upon receiving the
+ * success item from the current {@code Single}.</dd>
+ * </dl>
+ * @param unit the time unit for measurement
+ * @param scheduler the {@code Scheduler} used for providing the current time
+ * @return the new {@code Single} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ public final Single<Timed<T>> timeInterval(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ return RxJavaPlugins.onAssembly(new SingleTimeInterval<>(this, unit, scheduler, true));
+ }
+
+ /**
+ * Combines the success value from the current {@code Single} with the current time (in milliseconds) of
+ * its reception, using the {@code computation} {@link Scheduler} as time source,
+ * then signals them as a {@link Timed} instance.
+ * <p>
+ * <img width="640" height="465" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.timestamp.png" alt="">
+ * <p>
+ * If the current {@code Single} is empty or fails, the resulting {@code Single} will
+ * pass along the signals to the downstream. To get the timestamp of the error,
+ * use {@link #materialize()} and apply {@link #timestamp()}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timestamp} uses the {@code computation} {@code Scheduler}
+ * for determining the current time upon receiving the
+ * success item from the current {@code Single}.</dd>
+ * </dl>
+ * @return the new {@code Single} instance
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.COMPUTATION)
+ public final Single<Timed<T>> timestamp() {
+ return timestamp(TimeUnit.MILLISECONDS, Schedulers.computation());
+ }
+
+ /**
+ * Combines the success value from the current {@code Single} with the current time (in milliseconds) of
+ * its reception, using the given {@link Scheduler} as time source,
+ * then signals them as a {@link Timed} instance.
+ * <p>
+ * <img width="640" height="465" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.timestamp.s.png" alt="">
+ * <p>
+ * If the current {@code Single} is empty or fails, the resulting {@code Single} will
+ * pass along the signals to the downstream. To get the timestamp of the error,
+ * use {@link #materialize()} and apply {@link #timestamp(Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timestamp} uses the provided {@code Scheduler}
+ * for determining the current time upon receiving the
+ * success item from the current {@code Single}.</dd>
+ * </dl>
+ * @param scheduler the {@code Scheduler} used for providing the current time
+ * @return the new {@code Single} instance
+ * @throws NullPointerException if {@code scheduler} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ public final Single<Timed<T>> timestamp(@NonNull Scheduler scheduler) {
+ return timestamp(TimeUnit.MILLISECONDS, scheduler);
+ }
+
+ /**
+ * Combines the success value from the current {@code Single} with the current time of
+ * its reception, using the {@code computation} {@link Scheduler} as time source,
+ * then signals it as a {@link Timed} instance.
+ * <p>
+ * <img width="640" height="465" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.timestamp.png" alt="">
+ * <p>
+ * If the current {@code Single} is empty or fails, the resulting {@code Single} will
+ * pass along the signals to the downstream. To get the timestamp of the error,
+ * use {@link #materialize()} and apply {@link #timestamp(TimeUnit)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timestamp} uses the {@code computation} {@code Scheduler},
+ * for determining the current time upon receiving the
+ * success item from the current {@code Single}.</dd>
+ * </dl>
+ * @param unit the time unit for measurement
+ * @return the new {@code Single} instance
+ * @throws NullPointerException if {@code unit} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.COMPUTATION)
+ public final Single<Timed<T>> timestamp(@NonNull TimeUnit unit) {
+ return timestamp(unit, Schedulers.computation());
+ }
+
+ /**
+ * Combines the success value from the current {@code Single} with the current time of
+ * its reception, using the given {@link Scheduler} as time source,
+ * then signals it as a {@link Timed} instance.
+ * <p>
+ * <img width="640" height="465" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.timestamp.s.png" alt="">
+ * <p>
+ * If the current {@code Single} is empty or fails, the resulting {@code Single} will
+ * pass along the signals to the downstream. To get the timestamp of the error,
+ * use {@link #materialize()} and apply {@link #timestamp(TimeUnit, Scheduler)}.
+ * <dl>
+ * <dt><b>Scheduler:</b></dt>
+ * <dd>{@code timestamp} uses the provided {@code Scheduler},
+ * which is used for determining the current time upon receiving the
+ * success item from the current {@code Single}.</dd>
+ * </dl>
+ * @param unit the time unit for measurement
+ * @param scheduler the {@code Scheduler} used for providing the current time
+ * @return the new {@code Single} instance
+ * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null}
+ * @since 3.0.0
+ */
+ @CheckReturnValue
+ @NonNull
+ @SchedulerSupport(SchedulerSupport.CUSTOM)
+ public final Single<Timed<T>> timestamp(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
+ Objects.requireNonNull(unit, "unit is null");
+ Objects.requireNonNull(scheduler, "scheduler is null");
+ return RxJavaPlugins.onAssembly(new SingleTimeInterval<>(this, unit, scheduler, false));
+ }
+
/**
* Returns a {@code Single} that emits the item emitted by the current {@code Single} until a {@link CompletableSource} terminates. Upon
* termination of {@code other}, this will emit a {@link CancellationException} rather than go to
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimeInterval.java b/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimeInterval.java
new file mode 100644
index 0000000000..af30ee3a1d
--- /dev/null
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimeInterval.java
@@ -0,0 +1,105 @@
+/**
+ * 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.internal.operators.maybe;
+
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
+import io.reactivex.rxjava3.schedulers.Timed;
+
+/**
+ * Measures the time between subscription and the success item emission
+ * from the upstream and emits this as a {@link Timed} success value.
+ * @param <T> the element type of the sequence
+ * @since 3.0.0
+ */
+public final class MaybeTimeInterval<T> extends Maybe<Timed<T>> {
+
+ final MaybeSource<T> source;
+
+ final TimeUnit unit;
+
+ final Scheduler scheduler;
+
+ final boolean start;
+
+ public MaybeTimeInterval(MaybeSource<T> source, TimeUnit unit, Scheduler scheduler, boolean start) {
+ this.source = source;
+ this.unit = unit;
+ this.scheduler = scheduler;
+ this.start = start;
+ }
+
+ @Override
+ protected void subscribeActual(@NonNull MaybeObserver<? super @NonNull Timed<T>> observer) {
+ source.subscribe(new TimeIntervalMaybeObserver<>(observer, unit, scheduler, start));
+ }
+
+ static final class TimeIntervalMaybeObserver<T> implements MaybeObserver<T>, Disposable {
+
+ final MaybeObserver<? super Timed<T>> downstream;
+
+ final TimeUnit unit;
+
+ final Scheduler scheduler;
+
+ final long startTime;
+
+ Disposable upstream;
+
+ TimeIntervalMaybeObserver(MaybeObserver<? super Timed<T>> downstream, TimeUnit unit, Scheduler scheduler, boolean start) {
+ this.downstream = downstream;
+ this.unit = unit;
+ this.scheduler = scheduler;
+ this.startTime = start ? scheduler.now(unit) : 0L;
+ }
+
+ @Override
+ public void onSubscribe(@NonNull Disposable d) {
+ if (DisposableHelper.validate(this.upstream, d)) {
+ this.upstream = d;
+
+ downstream.onSubscribe(this);
+ }
+ }
+
+ @Override
+ public void onSuccess(@NonNull T t) {
+ downstream.onSuccess(new Timed<>(t, scheduler.now(unit) - startTime, unit));
+ }
+
+ @Override
+ public void onError(@NonNull Throwable e) {
+ downstream.onError(e);
+ }
+
+ @Override
+ public void onComplete() {
+ downstream.onComplete();
+ }
+
+ @Override
+ public void dispose() {
+ upstream.dispose();
+ }
+
+ @Override
+ public boolean isDisposed() {
+ return upstream.isDisposed();
+ }
+ }
+}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeInterval.java b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeInterval.java
new file mode 100644
index 0000000000..2eac3bbc3a
--- /dev/null
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeInterval.java
@@ -0,0 +1,100 @@
+/**
+ * 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.internal.operators.single;
+
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
+import io.reactivex.rxjava3.schedulers.Timed;
+
+/**
+ * Measures the time between subscription and the success item emission
+ * from the upstream and emits this as a {@link Timed} success value.
+ * @param <T> the element type of the sequence
+ * @since 3.0.0
+ */
+public final class SingleTimeInterval<T> extends Single<Timed<T>> {
+
+ final SingleSource<T> source;
+
+ final TimeUnit unit;
+
+ final Scheduler scheduler;
+
+ final boolean start;
+
+ public SingleTimeInterval(SingleSource<T> source, TimeUnit unit, Scheduler scheduler, boolean start) {
+ this.source = source;
+ this.unit = unit;
+ this.scheduler = scheduler;
+ this.start = start;
+ }
+
+ @Override
+ protected void subscribeActual(@NonNull SingleObserver<? super @NonNull Timed<T>> observer) {
+ source.subscribe(new TimeIntervalSingleObserver<>(observer, unit, scheduler, start));
+ }
+
+ static final class TimeIntervalSingleObserver<T> implements SingleObserver<T>, Disposable {
+
+ final SingleObserver<? super Timed<T>> downstream;
+
+ final TimeUnit unit;
+
+ final Scheduler scheduler;
+
+ final long startTime;
+
+ Disposable upstream;
+
+ TimeIntervalSingleObserver(SingleObserver<? super Timed<T>> downstream, TimeUnit unit, Scheduler scheduler, boolean start) {
+ this.downstream = downstream;
+ this.unit = unit;
+ this.scheduler = scheduler;
+ this.startTime = start ? scheduler.now(unit) : 0L;
+ }
+
+ @Override
+ public void onSubscribe(@NonNull Disposable d) {
+ if (DisposableHelper.validate(this.upstream, d)) {
+ this.upstream = d;
+
+ downstream.onSubscribe(this);
+ }
+ }
+
+ @Override
+ public void onSuccess(@NonNull T t) {
+ downstream.onSuccess(new Timed<>(t, scheduler.now(unit) - startTime, unit));
+ }
+
+ @Override
+ public void onError(@NonNull Throwable e) {
+ downstream.onError(e);
+ }
+
+ @Override
+ public void dispose() {
+ upstream.dispose();
+ }
+
+ @Override
+ public boolean isDisposed() {
+ return upstream.isDisposed();
+ }
+ }
+}
diff --git a/src/main/java/io/reactivex/rxjava3/schedulers/Timed.java b/src/main/java/io/reactivex/rxjava3/schedulers/Timed.java
index 814b223dcf..a7b6cc46e7 100644
--- a/src/main/java/io/reactivex/rxjava3/schedulers/Timed.java
+++ b/src/main/java/io/reactivex/rxjava3/schedulers/Timed.java
@@ -33,10 +33,10 @@ public final class Timed<T> {
* @param value the value to hold
* @param time the time to hold
* @param unit the time unit, not null
- * @throws NullPointerException if unit is {@code null}
+ * @throws NullPointerException if {@code value} or {@code unit} is {@code null}
*/
public Timed(@NonNull T value, long time, @NonNull TimeUnit unit) {
- this.value = value;
+ this.value = Objects.requireNonNull(value, "value is null");
this.time = time;
this.unit = Objects.requireNonNull(unit, "unit is null");
}
@@ -89,7 +89,7 @@ public boolean equals(Object other) {
@Override
public int hashCode() {
- int h = value != null ? value.hashCode() : 0;
+ int h = value.hashCode();
h = h * 31 + (int)((time >>> 31) ^ time);
h = h * 31 + unit.hashCode();
return h;
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimeIntervalTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimeIntervalTest.java
new file mode 100644
index 0000000000..30b770aed4
--- /dev/null
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimeIntervalTest.java
@@ -0,0 +1,111 @@
+/**
+ * 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.internal.operators.maybe;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.observers.TestObserver;
+import io.reactivex.rxjava3.schedulers.*;
+import io.reactivex.rxjava3.subjects.MaybeSubject;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class MaybeTimeIntervalTest {
+
+ @Test
+ public void just() {
+ Maybe.just(1)
+ .timeInterval()
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void empty() {
+ Maybe.empty()
+ .timeInterval()
+ .test()
+ .assertResult();
+ }
+
+ @Test
+ public void error() {
+ Maybe.error(new TestException())
+ .timeInterval()
+ .test()
+ .assertFailure(TestException.class);
+ }
+
+ @Test
+ public void justSeconds() {
+ Maybe.just(1)
+ .timeInterval(TimeUnit.SECONDS)
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void justScheduler() {
+ Maybe.just(1)
+ .timeInterval(Schedulers.single())
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void justSecondsScheduler() {
+ Maybe.just(1)
+ .timeInterval(TimeUnit.SECONDS, Schedulers.single())
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void doubleOnSubscribe() {
+ TestHelper.checkDoubleOnSubscribeMaybe(m -> m.timeInterval());
+ }
+
+ @Test
+ public void dispose() {
+ TestHelper.checkDisposed(MaybeSubject.create().timeInterval());
+ }
+
+ @Test
+ public void timeInfo() {
+ TestScheduler scheduler = new TestScheduler();
+
+ MaybeSubject<Integer> ms = MaybeSubject.create();
+
+ TestObserver<Timed<Integer>> to = ms
+ .timeInterval(scheduler)
+ .test();
+
+ scheduler.advanceTimeBy(1000, TimeUnit.MILLISECONDS);
+
+ ms.onSuccess(1);
+
+ to.assertResult(new Timed<>(1, 1000L, TimeUnit.MILLISECONDS));
+ }
+}
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimestampTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimestampTest.java
new file mode 100644
index 0000000000..1ce230f8d0
--- /dev/null
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeTimestampTest.java
@@ -0,0 +1,111 @@
+/**
+ * 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.internal.operators.maybe;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.observers.TestObserver;
+import io.reactivex.rxjava3.schedulers.*;
+import io.reactivex.rxjava3.subjects.MaybeSubject;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class MaybeTimestampTest {
+
+ @Test
+ public void just() {
+ Maybe.just(1)
+ .timestamp()
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void empty() {
+ Maybe.empty()
+ .timestamp()
+ .test()
+ .assertResult();
+ }
+
+ @Test
+ public void error() {
+ Maybe.error(new TestException())
+ .timestamp()
+ .test()
+ .assertFailure(TestException.class);
+ }
+
+ @Test
+ public void justSeconds() {
+ Maybe.just(1)
+ .timestamp(TimeUnit.SECONDS)
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void justScheduler() {
+ Maybe.just(1)
+ .timestamp(Schedulers.single())
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void justSecondsScheduler() {
+ Maybe.just(1)
+ .timestamp(TimeUnit.SECONDS, Schedulers.single())
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void doubleOnSubscribe() {
+ TestHelper.checkDoubleOnSubscribeMaybe(m -> m.timestamp());
+ }
+
+ @Test
+ public void dispose() {
+ TestHelper.checkDisposed(MaybeSubject.create().timestamp());
+ }
+
+ @Test
+ public void timeInfo() {
+ TestScheduler scheduler = new TestScheduler();
+
+ MaybeSubject<Integer> ms = MaybeSubject.create();
+
+ TestObserver<Timed<Integer>> to = ms
+ .timestamp(scheduler)
+ .test();
+
+ scheduler.advanceTimeBy(1000, TimeUnit.MILLISECONDS);
+
+ ms.onSuccess(1);
+
+ to.assertResult(new Timed<>(1, 1000L, TimeUnit.MILLISECONDS));
+ }
+}
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeIntervalTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeIntervalTest.java
new file mode 100644
index 0000000000..09dd26eeba
--- /dev/null
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleTimeIntervalTest.java
@@ -0,0 +1,103 @@
+/**
+ * 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.internal.operators.single;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.observers.TestObserver;
+import io.reactivex.rxjava3.schedulers.*;
+import io.reactivex.rxjava3.subjects.SingleSubject;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class SingleTimeIntervalTest {
+
+ @Test
+ public void just() {
+ Single.just(1)
+ .timestamp()
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void error() {
+ Single.error(new TestException())
+ .timestamp()
+ .test()
+ .assertFailure(TestException.class);
+ }
+
+ @Test
+ public void justSeconds() {
+ Single.just(1)
+ .timestamp(TimeUnit.SECONDS)
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void justScheduler() {
+ Single.just(1)
+ .timestamp(Schedulers.single())
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void justSecondsScheduler() {
+ Single.just(1)
+ .timestamp(TimeUnit.SECONDS, Schedulers.single())
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void doubleOnSubscribe() {
+ TestHelper.checkDoubleOnSubscribeSingle(m -> m.timestamp());
+ }
+
+ @Test
+ public void dispose() {
+ TestHelper.checkDisposed(SingleSubject.create().timestamp());
+ }
+
+ @Test
+ public void timeInfo() {
+ TestScheduler scheduler = new TestScheduler();
+
+ SingleSubject<Integer> ss = SingleSubject.create();
+
+ TestObserver<Timed<Integer>> to = ss
+ .timestamp(scheduler)
+ .test();
+
+ scheduler.advanceTimeBy(1000, TimeUnit.MILLISECONDS);
+
+ ss.onSuccess(1);
+
+ to.assertResult(new Timed<>(1, 1000L, TimeUnit.MILLISECONDS));
+ }
+}
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleTimestampTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleTimestampTest.java
new file mode 100644
index 0000000000..a282c0345e
--- /dev/null
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleTimestampTest.java
@@ -0,0 +1,103 @@
+/**
+ * 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.internal.operators.single;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.observers.TestObserver;
+import io.reactivex.rxjava3.schedulers.*;
+import io.reactivex.rxjava3.subjects.SingleSubject;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class SingleTimestampTest {
+
+ @Test
+ public void just() {
+ Single.just(1)
+ .timeInterval()
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void error() {
+ Single.error(new TestException())
+ .timeInterval()
+ .test()
+ .assertFailure(TestException.class);
+ }
+
+ @Test
+ public void justSeconds() {
+ Single.just(1)
+ .timeInterval(TimeUnit.SECONDS)
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void justScheduler() {
+ Single.just(1)
+ .timeInterval(Schedulers.single())
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void justSecondsScheduler() {
+ Single.just(1)
+ .timeInterval(TimeUnit.SECONDS, Schedulers.single())
+ .test()
+ .assertValueCount(1)
+ .assertNoErrors()
+ .assertComplete();
+ }
+
+ @Test
+ public void doubleOnSubscribe() {
+ TestHelper.checkDoubleOnSubscribeSingle(m -> m.timeInterval());
+ }
+
+ @Test
+ public void dispose() {
+ TestHelper.checkDisposed(SingleSubject.create().timeInterval());
+ }
+
+ @Test
+ public void timeInfo() {
+ TestScheduler scheduler = new TestScheduler();
+
+ SingleSubject<Integer> ss = SingleSubject.create();
+
+ TestObserver<Timed<Integer>> to = ss
+ .timeInterval(scheduler)
+ .test();
+
+ scheduler.advanceTimeBy(1000, TimeUnit.MILLISECONDS);
+
+ ss.onSuccess(1);
+
+ to.assertResult(new Timed<>(1, 1000L, TimeUnit.MILLISECONDS));
+ }
+}
diff --git a/src/test/java/io/reactivex/rxjava3/internal/util/OperatorMatrixGenerator.java b/src/test/java/io/reactivex/rxjava3/internal/util/OperatorMatrixGenerator.java
index 60b5a11fb6..a0ba3e52e4 100644
--- a/src/test/java/io/reactivex/rxjava3/internal/util/OperatorMatrixGenerator.java
+++ b/src/test/java/io/reactivex/rxjava3/internal/util/OperatorMatrixGenerator.java
@@ -458,6 +458,7 @@ static String findNotes(String clazzName, String operatorName) {
" C throttleLatest Always empty thus no items to work with.",
" MS throttleWithTimeout At most one item signaled so no subsequent items to work with.",
" C throttleWithTimeout Always empty thus no items to work with.",
+ " C timeInterval Always empty thus no items to work with.",
" C timestamp Always empty thus no items to work with.",
"FO toCompletionStage Use [`firstStage`](#firstStage), [`lastStage`](#lastStage) or [`singleStage`](#singleStage).",
"F toFlowable Would be no-op.",
diff --git a/src/test/java/io/reactivex/rxjava3/schedulers/TimedTest.java b/src/test/java/io/reactivex/rxjava3/schedulers/TimedTest.java
index d7a5eb752e..ce117c2fb7 100644
--- a/src/test/java/io/reactivex/rxjava3/schedulers/TimedTest.java
+++ b/src/test/java/io/reactivex/rxjava3/schedulers/TimedTest.java
@@ -39,7 +39,7 @@ public void hashCodeOf() {
assertEquals(TimeUnit.SECONDS.hashCode() + 31 * (5 + 31 * 1), t1.hashCode());
- Timed<Integer> t2 = new Timed<>(null, 5, TimeUnit.SECONDS);
+ Timed<Integer> t2 = new Timed<>(0, 5, TimeUnit.SECONDS);
assertEquals(TimeUnit.SECONDS.hashCode() + 31 * (5 + 31 * 0), t2.hashCode());
}
diff --git a/src/test/java/io/reactivex/rxjava3/validators/JavadocWording.java b/src/test/java/io/reactivex/rxjava3/validators/JavadocWording.java
index d351e7809d..e22fdd9521 100644
--- a/src/test/java/io/reactivex/rxjava3/validators/JavadocWording.java
+++ b/src/test/java/io/reactivex/rxjava3/validators/JavadocWording.java
@@ -153,12 +153,12 @@ public void maybeDocRefersToMaybeTypes() throws Exception {
jdx = 0;
for (;;) {
int idx = m.javadoc.indexOf("Single", jdx);
- if (idx >= 0) {
+ if (idx >= 0 && m.javadoc.indexOf("Single#", jdx) != idx) {
int j = m.javadoc.indexOf("#toSingle", jdx);
int k = m.javadoc.indexOf("{@code Single", jdx);
if (!m.signature.contains("Single") && (j + 3 != idx && k + 7 != idx)) {
e.append("java.lang.RuntimeException: Maybe doc mentions Single but not in the signature\r\n at io.reactivex.rxjava3.core.")
- .append("Maybe(Maybe.java:").append(m.javadocLine + lineNumber(m.javadoc, idx) - 1).append(")\r\n\r\n");
+ .append("Maybe.method(Maybe.java:").append(m.javadocLine + lineNumber(m.javadoc, idx) - 1).append(")\r\n\r\n");
}
jdx = idx + 6;
} else {