Remove ThrowableWrapper, fix attempt at flaky tests

Author: akarnokd

src/main/java/io/reactivex/rxjava4/core/CompletionStageDisposable.java

@@ -21,9 +21,8 @@
 import java.util.function.Consumer;
 
 import io.reactivex.rxjava4.annotations.NonNull;
-import io.reactivex.rxjava4.disposables.*;
-import io.reactivex.rxjava4.exceptions.ThrowableWrapper;
-import io.reactivex.rxjava4.internal.util.*;
+import io.reactivex.rxjava4.disposables.Disposable;
+import io.reactivex.rxjava4.internal.util.ExceptionHelper;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 
 /**
@@ -88,14 +87,14 @@ public CompletionStageDisposable(@NonNull CompletionStage<T> stage, @NonNull Dis
      * <p>
      * Rethrows any original unchecked exceptions as is.
      * @throws CancellationException if the computation was cancelled
-     * @throws ThrowableWrapper if the original exception was a checked exception
+     * @throws CompletionException if the original exception was a checked exception
      */
     public void await() {
         state.lazySet(true);
         try {
             stage.toCompletableFuture().join();
         } catch (CompletionException ce) {
-            throw ExceptionHelper.wrapOrThrow(ce.getCause());
+            throw ExceptionHelper.unwrapOrThrow(ce);
         }
     }
 

src/main/java/io/reactivex/rxjava4/exceptions/Exceptions.java

@@ -13,6 +13,8 @@
 
 package io.reactivex.rxjava4.exceptions;
 
+import java.util.concurrent.CompletionException;
+
 import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.internal.util.ExceptionHelper;
 
@@ -28,7 +30,7 @@ private Exceptions() {
     }
     /**
      * Convenience method to throw a {@code RuntimeException} and {@code Error} directly
-     * or wrap any other exception type into a {@link ThrowableWrapper}.
+     * or wrap any other exception type into a {@link CompletionException}.
      * @param t the exception to throw directly or wrapped
      * @return because {@code propagate} itself throws an exception or error, this is a sort of phantom return
      *         value; {@code propagate} does not actually return anything

src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java

@@ -74,7 +74,9 @@ public void run() {
         }
 
         void interrupedSleep(InterruptedException ex) {
-            waiter.completeExceptionally(ex);
+            if (!isDisposed()) {
+                waiter.completeExceptionally(ex);
+            }
         }
 
         @Override

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java

@@ -18,8 +18,8 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;
 
-import io.reactivex.rxjava4.annotations.Nullable;
-import io.reactivex.rxjava4.exceptions.*;
+import io.reactivex.rxjava4.annotations.*;
+import io.reactivex.rxjava4.exceptions.CompositeException;
 
 /**
  * Terminal atomics for Throwable containers.
@@ -33,19 +33,39 @@ private ExceptionHelper() {
 
     /**
      * If the provided Throwable is an Error this method
-     * throws it, otherwise returns a RuntimeException wrapping the error
+     * throws it, otherwise returns a CompletionException wrapping the error
      * if that error is a checked exception.
      * @param error the error to wrap or throw
      * @return the (wrapped) error
      */
-    public static RuntimeException wrapOrThrow(Throwable error) {
+    @NonNull
+    public static RuntimeException wrapOrThrow(@NonNull Throwable error) {
         if (error instanceof Error err) {
             throw err;
         }
         if (error instanceof RuntimeException rte) {
             return rte;
         }
-        return new ThrowableWrapper(error);
+        return new CompletionException("You forgot to unwrap me!", error);
+    }
+    /**
+     * Unwraps a {@link CompletionException} and rethrows its {@link Error}
+     * or {@link RuntimeException} inside it, or returns it as is if
+     * the {@code CompletionException} holds a checked exception.
+     * @param error the error to unwrap and rethrow its cause if possible
+     * @return the {@code error} if it has a checked exception cause
+     * @since 4.0.0
+     */
+    @NonNull
+    public static RuntimeException unwrapOrThrow(@NonNull CompletionException error) {
+        var cause = error.getCause();
+        if (cause instanceof Error err) {
+            throw err;
+        }
+        if (cause instanceof RuntimeException rte) {
+            return rte;
+        }
+        return error;
     }
 
     /**
@@ -184,19 +204,19 @@ public static <T> T nullCheck(T value, String prefix) {
     }
 
     /**
-     * Unwraps both throwables if they are wrapped into a {@link CompletionException} or
-     * {@link ThrowableWrapper}, then if both are present, add {@code b} as suppressed to {@code a}
+     * Unwraps both throwables if they are wrapped into a {@link CompletionException},
+     * then if both are present, add {@code b} as suppressed to {@code a}
      * and return a; return b otherwise
      * @param main the first throwable
      * @param secondary the second throwable
      * @return the unwrapped and combined throwable or null if both where
      */
     @Nullable
     public static Throwable unwrapAndCombine(@Nullable Throwable main, @Nullable Throwable secondary) {
-        if (main instanceof CompletionException || main instanceof ThrowableWrapper) {
+        if (main instanceof CompletionException) {
             main = main.getCause();
         }
-        if (secondary instanceof CompletionException || secondary instanceof ThrowableWrapper) {
+        if (secondary instanceof CompletionException) {
             secondary = secondary.getCause();
         }
         if (main != null && secondary != null && main != secondary) {
@@ -209,12 +229,12 @@ public static Throwable unwrapAndCombine(@Nullable Throwable main, @Nullable Thr
     }
 
     /**
-     * Unwraps the given {@link CompletionException} or {@link ThrowableWrapper}
+     * Unwraps the given {@link CompletionException}.
      * @param t the possible throwable to unwrap
      * @return the unwrapped Throwable
      */
     public static Throwable unwrap(@Nullable Throwable t) {
-        if (t instanceof CompletionException || t instanceof ThrowableWrapper) {
+        if (t instanceof CompletionException) {
             t = t.getCause();
         }
         return t;

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

@@ -22,7 +22,7 @@
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.exceptions.*;
-import io.reactivex.rxjava4.internal.util.BackpressureHelper;
+import io.reactivex.rxjava4.internal.util.*;
 
 /**
  * Runs a generator callback on a virtual thread backed by a Worker of the given scheduler
@@ -105,7 +105,7 @@ public Void call() {
                 } catch (Throwable ex) {
                     Exceptions.throwIfFatal(ex);
                     if (ex != STOP && !cancelled) {
-                        downstream.onError(ThrowableWrapper.unwrap(ex));
+                        downstream.onError(ExceptionHelper.unwrap(ex));
                     }
                     return null;
                 }

src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java

@@ -22,8 +22,8 @@
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.core.Scheduler.Worker;
 import io.reactivex.rxjava4.disposables.*;
-import io.reactivex.rxjava4.exceptions.*;
-import io.reactivex.rxjava4.internal.util.BackpressureHelper;
+import io.reactivex.rxjava4.exceptions.Exceptions;
+import io.reactivex.rxjava4.internal.util.*;
 import io.reactivex.rxjava4.operators.SpscArrayQueue;
 
 public final class FlowableVirtualTransformExecutor<T, R> extends Flowable<R> {
@@ -206,7 +206,7 @@ public Void call() {
                         if (d && empty) {
                             var ex = error;
                             if (ex != null) {
-                                downstream.onError(ThrowableWrapper.unwrap(ex));
+                                downstream.onError(ExceptionHelper.unwrap(ex));
                             } else {
                                 downstream.onComplete();
                             }
@@ -234,7 +234,7 @@ public Void call() {
                     Exceptions.throwIfFatal(ex);
                     if (ex != STOP && !cancelled) {
                         upstream.cancel();
-                        downstream.onError(ThrowableWrapper.unwrap(ex));
+                        downstream.onError(ExceptionHelper.unwrap(ex));
                     }
                     return null;
                 }

src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java

@@ -29,7 +29,7 @@ public class StreamableForEachTest extends StreamableBaseTest {
 
     @Test
     public void forEachCheckedCrash() {
-        var ex = assertThrows(ThrowableWrapper.class, () -> {
+        var ex = assertThrows(CompletionException.class, () -> {
             Streamable.just(1)
             .forEach(_ -> {
                 throw new Exception("test");
@@ -58,7 +58,7 @@ public void forEachUncheckedCrash() {
     @Test
     public void forEachExecCheckedCrash() throws Throwable {
         withCachedExecutor(exec -> {
-            var ex = assertThrows(ThrowableWrapper.class, () -> {
+            var ex = assertThrows(CompletionException.class, () -> {
                 Streamable.just(1)
                 .forEach(_ -> {
                     throw new Exception("test");
@@ -90,7 +90,7 @@ public void forEachExecUncheckedCrash() throws Throwable {
     @Test
     public void forEachBiCheckedCrash() throws Throwable {
         withVirtual(exec -> {
-            var ex = assertThrows(ThrowableWrapper.class, () -> {
+            var ex = assertThrows(CompletionException.class, () -> {
                 Streamable.just(1)
                 .forEach((_, _) -> {
                     throw new Exception("test");

src/test/java/io/reactivex/rxjava4/exceptions/ThrowableWrapperTest.java

@@ -91,12 +91,12 @@ public void unwrapAndCombine6() {
     @Test
     public void unwrapAndCombine7() {
         var te = new TestException();
-        assertSame(te, ExceptionHelper.unwrapAndCombine(new ThrowableWrapper(te), null));
+        assertSame(te, ExceptionHelper.unwrapAndCombine(new CompletionException(te), null));
     }
 
     @Test
     public void unwrapAndCombine8() {
         var te = new TestException();
-        assertSame(te, ExceptionHelper.unwrapAndCombine(null, new ThrowableWrapper(te)));
+        assertSame(te, ExceptionHelper.unwrapAndCombine(null, new CompletionException(te)));
     }
 }