Add timeout and unit to TimeoutException message (#6234)

* Add timeout and unit to TimeoutException message

* New timeout message, add tests

Author: artem-zinnatullin

src/main/java/io/reactivex/internal/observers/BlockingMultiObserver.java

@@ -19,6 +19,8 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.util.*;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
+
 /**
  * A combined Observer that awaits the success or error signal via a CountDownLatch.
  * @param <T> the value type
@@ -148,7 +150,7 @@ public Throwable blockingGetError(long timeout, TimeUnit unit) {
                 BlockingHelper.verifyNonBlocking();
                 if (!await(timeout, unit)) {
                     dispose();
-                    throw ExceptionHelper.wrapOrThrow(new TimeoutException());
+                    throw ExceptionHelper.wrapOrThrow(new TimeoutException(timeoutMessage(timeout, unit)));
                 }
             } catch (InterruptedException ex) {
                 dispose();

src/main/java/io/reactivex/internal/observers/FutureObserver.java

@@ -23,6 +23,8 @@
 import io.reactivex.internal.util.BlockingHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
+
 /**
  * An Observer + Future that expects exactly one upstream value and provides it
  * via the (blocking) Future API.
@@ -92,7 +94,7 @@ public T get(long timeout, TimeUnit unit) throws InterruptedException, Execution
         if (getCount() != 0) {
             BlockingHelper.verifyNonBlocking();
             if (!await(timeout, unit)) {
-                throw new TimeoutException();
+                throw new TimeoutException(timeoutMessage(timeout, unit));
             }
         }
 

src/main/java/io/reactivex/internal/observers/FutureSingleObserver.java

@@ -22,6 +22,8 @@
 import io.reactivex.internal.util.BlockingHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
+
 /**
  * An Observer + Future that expects exactly one upstream value and provides it
  * via the (blocking) Future API.
@@ -91,7 +93,7 @@ public T get(long timeout, TimeUnit unit) throws InterruptedException, Execution
         if (getCount() != 0) {
             BlockingHelper.verifyNonBlocking();
             if (!await(timeout, unit)) {
-                throw new TimeoutException();
+                throw new TimeoutException(timeoutMessage(timeout, unit));
             }
         }
 

src/main/java/io/reactivex/internal/operators/completable/CompletableTimeout.java

@@ -20,6 +20,8 @@
 import io.reactivex.disposables.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
+
 public final class CompletableTimeout extends Completable {
 
     final CompletableSource source;
@@ -104,7 +106,7 @@ public void run() {
             if (once.compareAndSet(false, true)) {
                 set.clear();
                 if (other == null) {
-                    downstream.onError(new TimeoutException());
+                    downstream.onError(new TimeoutException(timeoutMessage(timeout, unit)));
                 } else {
                     other.subscribe(new DisposeObserver());
                 }

src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java

@@ -23,6 +23,8 @@
 import io.reactivex.internal.subscriptions.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
+
 public final class FlowableTimeoutTimed<T> extends AbstractFlowableWithUpstream<T, T> {
     final long timeout;
     final TimeUnit unit;
@@ -134,7 +136,7 @@ public void onTimeout(long idx) {
             if (compareAndSet(idx, Long.MAX_VALUE)) {
                 SubscriptionHelper.cancel(upstream);
 
-                downstream.onError(new TimeoutException());
+                downstream.onError(new TimeoutException(timeoutMessage(timeout, unit)));
 
                 worker.dispose();
             }

src/main/java/io/reactivex/internal/operators/observable/ObservableTimeoutTimed.java

@@ -21,6 +21,8 @@
 import io.reactivex.internal.disposables.*;
 import io.reactivex.plugins.RxJavaPlugins;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
+
 public final class ObservableTimeoutTimed<T> extends AbstractObservableWithUpstream<T, T> {
     final long timeout;
     final TimeUnit unit;
@@ -129,7 +131,7 @@ public void onTimeout(long idx) {
             if (compareAndSet(idx, Long.MAX_VALUE)) {
                 DisposableHelper.dispose(upstream);
 
-                downstream.onError(new TimeoutException());
+                downstream.onError(new TimeoutException(timeoutMessage(timeout, unit)));
 
                 worker.dispose();
             }

src/main/java/io/reactivex/internal/operators/single/SingleTimeout.java

@@ -21,6 +21,8 @@
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
+
 public final class SingleTimeout<T> extends Single<T> {
 
     final SingleSource<T> source;
@@ -45,7 +47,7 @@ public SingleTimeout(SingleSource<T> source, long timeout, TimeUnit unit, Schedu
     @Override
     protected void subscribeActual(final SingleObserver<? super T> observer) {
 
-        TimeoutMainObserver<T> parent = new TimeoutMainObserver<T>(observer, other);
+        TimeoutMainObserver<T> parent = new TimeoutMainObserver<T>(observer, other, timeout, unit);
         observer.onSubscribe(parent);
 
         DisposableHelper.replace(parent.task, scheduler.scheduleDirect(parent, timeout, unit));
@@ -66,6 +68,10 @@ static final class TimeoutMainObserver<T> extends AtomicReference<Disposable>
 
         SingleSource<? extends T> other;
 
+        final long timeout;
+
+        final TimeUnit unit;
+
         static final class TimeoutFallbackObserver<T> extends AtomicReference<Disposable>
         implements SingleObserver<T> {
 
@@ -92,9 +98,11 @@ public void onError(Throwable e) {
             }
         }
 
-        TimeoutMainObserver(SingleObserver<? super T> actual, SingleSource<? extends T> other) {
+        TimeoutMainObserver(SingleObserver<? super T> actual, SingleSource<? extends T> other, long timeout, TimeUnit unit) {
             this.downstream = actual;
             this.other = other;
+            this.timeout = timeout;
+            this.unit = unit;
             this.task = new AtomicReference<Disposable>();
             if (other != null) {
                 this.fallback = new TimeoutFallbackObserver<T>(actual);
@@ -112,7 +120,7 @@ public void run() {
                 }
                 SingleSource<? extends T> other = this.other;
                 if (other == null) {
-                    downstream.onError(new TimeoutException());
+                    downstream.onError(new TimeoutException(timeoutMessage(timeout, unit)));
                 } else {
                     this.other = null;
                     other.subscribe(fallback);

src/main/java/io/reactivex/internal/subscribers/FutureSubscriber.java

@@ -24,6 +24,8 @@
 import io.reactivex.internal.util.BlockingHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
+
 /**
  * A Subscriber + Future that expects exactly one upstream value and provides it
  * via the (blocking) Future API.
@@ -93,7 +95,7 @@ public T get(long timeout, TimeUnit unit) throws InterruptedException, Execution
         if (getCount() != 0) {
             BlockingHelper.verifyNonBlocking();
             if (!await(timeout, unit)) {
-                throw new TimeoutException();
+                throw new TimeoutException(timeoutMessage(timeout, unit));
             }
         }
 

src/main/java/io/reactivex/internal/util/ExceptionHelper.java

@@ -14,6 +14,7 @@
 package io.reactivex.internal.util;
 
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.exceptions.CompositeException;
@@ -121,6 +122,14 @@ public static <E extends Throwable> Exception throwIfThrowable(Throwable e) thro
         throw (E)e;
     }
 
+    public static String timeoutMessage(long timeout, TimeUnit unit) {
+        return "The source did not signal an event for "
+                + timeout
+                + " "
+                + unit.toString().toLowerCase()
+                + " and has been terminated.";
+    }
+    
     static final class Termination extends Throwable {
 
         private static final long serialVersionUID = -4649703670690200604L;

src/test/java/io/reactivex/internal/observers/BlockingMultiObserverTest.java

@@ -13,9 +13,11 @@
 
 package io.reactivex.internal.observers;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
 import static org.junit.Assert.*;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.junit.Test;
 
@@ -132,4 +134,17 @@ public void run() {
 
         assertTrue(bmo.blockingGetError(1, TimeUnit.MINUTES) instanceof TestException);
     }
+
+    @Test
+    public void blockingGetErrorTimedOut() {
+        final BlockingMultiObserver<Integer> bmo = new BlockingMultiObserver<Integer>();
+
+        try {
+            assertNull(bmo.blockingGetError(1, TimeUnit.NANOSECONDS));
+            fail("Should have thrown");
+        } catch (RuntimeException expected) {
+            assertEquals(TimeoutException.class, expected.getCause().getClass());
+            assertEquals(timeoutMessage(1, TimeUnit.NANOSECONDS), expected.getCause().getMessage());
+        }
+    }
 }

src/test/java/io/reactivex/internal/observers/FutureObserverTest.java

@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.observers;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
 import static org.junit.Assert.*;
 
 import java.util.*;
@@ -352,4 +353,14 @@ public void run() {
 
         assertEquals(1, fo.get().intValue());
     }
+
+    @Test
+    public void getTimedOut() throws Exception {
+        try {
+            fo.get(1, TimeUnit.NANOSECONDS);
+            fail("Should have thrown");
+        } catch (TimeoutException expected) {
+            assertEquals(timeoutMessage(1, TimeUnit.NANOSECONDS), expected.getMessage());
+        }
+    }
 }

src/test/java/io/reactivex/internal/observers/FutureSingleObserverTest.java

@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.observers;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
 import static org.junit.Assert.*;
 
 import java.util.concurrent.*;
@@ -89,8 +90,8 @@ public void timeout() throws Exception {
         try {
             f.get(100, TimeUnit.MILLISECONDS);
             fail("Should have thrown");
-        } catch (TimeoutException ex) {
-            // expected
+        } catch (TimeoutException expected) {
+            assertEquals(timeoutMessage(100, TimeUnit.MILLISECONDS), expected.getMessage());
         }
     }
 

src/test/java/io/reactivex/internal/operators/completable/CompletableTimeoutTest.java

@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.completable;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
 import static org.junit.Assert.*;
 
 import java.util.List;
@@ -40,7 +41,7 @@ public void timeoutException() throws Exception {
         .timeout(100, TimeUnit.MILLISECONDS, Schedulers.io())
         .test()
         .awaitDone(5, TimeUnit.SECONDS)
-        .assertFailure(TimeoutException.class);
+        .assertFailureAndMessage(TimeoutException.class, timeoutMessage(100, TimeUnit.MILLISECONDS));
     }
 
     @Test

src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTests.java

@@ -13,6 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import static io.reactivex.internal.util.ExceptionHelper.timeoutMessage;
 import static org.junit.Assert.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
@@ -82,26 +83,24 @@ public void shouldNotTimeoutIfSecondOnNextWithinTimeout() {
 
     @Test
     public void shouldTimeoutIfOnNextNotWithinTimeout() {
-        Subscriber<String> subscriber = TestHelper.mockSubscriber();
-        TestSubscriber<String> ts = new TestSubscriber<String>(subscriber);
+        TestSubscriber<String> subscriber = new TestSubscriber<String>();
 
-        withTimeout.subscribe(ts);
+        withTimeout.subscribe(subscriber);
 
         testScheduler.advanceTimeBy(TIMEOUT + 1, TimeUnit.SECONDS);
-        verify(subscriber).onError(any(TimeoutException.class));
-        ts.dispose();
+        subscriber.assertFailureAndMessage(TimeoutException.class, timeoutMessage(TIMEOUT, TIME_UNIT));
     }
 
     @Test
     public void shouldTimeoutIfSecondOnNextNotWithinTimeout() {
-        Subscriber<String> subscriber = TestHelper.mockSubscriber();
+        TestSubscriber<String> subscriber = new TestSubscriber<String>();
         TestSubscriber<String> ts = new TestSubscriber<String>(subscriber);
         withTimeout.subscribe(subscriber);
         testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
         underlyingSubject.onNext("One");
-        verify(subscriber).onNext("One");
+        subscriber.assertValue("One");
         testScheduler.advanceTimeBy(TIMEOUT + 1, TimeUnit.SECONDS);
-        verify(subscriber).onError(any(TimeoutException.class));
+        subscriber.assertFailureAndMessage(TimeoutException.class, timeoutMessage(TIMEOUT, TIME_UNIT), "One");
         ts.dispose();
     }
 
@@ -235,8 +234,7 @@ public void shouldTimeoutIfSynchronizedFlowableEmitFirstOnNextNotWithinTimeout()
         final CountDownLatch exit = new CountDownLatch(1);
         final CountDownLatch timeoutSetuped = new CountDownLatch(1);
 
-        final Subscriber<String> subscriber = TestHelper.mockSubscriber();
-        final TestSubscriber<String> ts = new TestSubscriber<String>(subscriber);
+        final TestSubscriber<String> subscriber = new TestSubscriber<String>();
 
         new Thread(new Runnable() {
 
@@ -258,16 +256,14 @@ public void subscribe(Subscriber<? super String> subscriber) {
                     }
 
                 }).timeout(1, TimeUnit.SECONDS, testScheduler)
-                        .subscribe(ts);
+                        .subscribe(subscriber);
             }
         }).start();
 
         timeoutSetuped.await();
         testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
 
-        InOrder inOrder = inOrder(subscriber);
-        inOrder.verify(subscriber, times(1)).onError(isA(TimeoutException.class));
-        inOrder.verifyNoMoreInteractions();
+        subscriber.assertFailureAndMessage(TimeoutException.class, timeoutMessage(1, TimeUnit.SECONDS));
 
         exit.countDown(); // exit the thread
     }
@@ -287,15 +283,12 @@ public void subscribe(Subscriber<? super String> subscriber) {
         TestScheduler testScheduler = new TestScheduler();
         Flowable<String> observableWithTimeout = never.timeout(1000, TimeUnit.MILLISECONDS, testScheduler);
 
-        Subscriber<String> subscriber = TestHelper.mockSubscriber();
-        TestSubscriber<String> ts = new TestSubscriber<String>(subscriber);
-        observableWithTimeout.subscribe(ts);
+        TestSubscriber<String> subscriber = new TestSubscriber<String>();
+        observableWithTimeout.subscribe(subscriber);
 
         testScheduler.advanceTimeBy(2000, TimeUnit.MILLISECONDS);
 
-        InOrder inOrder = inOrder(subscriber);
-        inOrder.verify(subscriber).onError(isA(TimeoutException.class));
-        inOrder.verifyNoMoreInteractions();
+        subscriber.assertFailureAndMessage(TimeoutException.class, timeoutMessage(1000, TimeUnit.MILLISECONDS));
 
         verify(s, times(1)).cancel();
     }
@@ -548,11 +541,13 @@ public void run() {
             if (ts.valueCount() != 0) {
                 if (ts.errorCount() != 0) {
                     ts.assertFailure(TimeoutException.class, 1);
+                    ts.assertErrorMessage(timeoutMessage(1, TimeUnit.SECONDS));
                 } else {
                     ts.assertValuesOnly(1);
                 }
             } else {
                 ts.assertFailure(TimeoutException.class);
+                ts.assertErrorMessage(timeoutMessage(1, TimeUnit.SECONDS));
             }
         }
     }