Fix for exceptional handling of Promise.orTimeout

Author: vsilaev

src/main/java/net/tascalate/concurrent/ConfigurableDependentPromise.java

@@ -18,8 +18,6 @@
 import static net.tascalate.concurrent.LinkedCompletion.FutureCompletion;
 import static net.tascalate.concurrent.SharedFunctions.cancelPromise;
 import static net.tascalate.concurrent.SharedFunctions.selectFirst;
-import static net.tascalate.concurrent.SharedFunctions.supply;
-import static net.tascalate.concurrent.SharedFunctions.timeout;
 
 import java.time.Duration;
 import java.util.Arrays;
@@ -154,11 +152,12 @@ public DependentPromise<T> orTimeout(Duration duration, boolean cancelOnTimeout)
 
     @Override
     public DependentPromise<T> orTimeout(Duration duration, boolean cancelOnTimeout, boolean enlistOrigin) {
-        Promise<? extends Reference<T>> onTimeout = Timeouts.delayed(null, duration);
+        Promise<? extends Try<T>> onTimeout = Timeouts.delayed(null, duration);
         DependentPromise<T> result = 
-        this.thenApply(Reference::new, enlistOrigin)
+        this.thenApply(Try::success, enlistOrigin)
+            .exceptionally(Try::failure, true)
             // Use *Async to execute on default "this" executor
-            .applyToEitherAsync(onTimeout, v -> Reference.getOrElse(v, timeout(duration)), PromiseOrigin.ALL);
+            .applyToEitherAsync(onTimeout, v -> Try.doneOrTimeout(v, duration), PromiseOrigin.ALL);
 
         result.whenComplete(Timeouts.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
         return result;
@@ -171,9 +170,12 @@ public DependentPromise<T> onTimeout(T value, Duration duration, boolean cancelO
 
     @Override
     public DependentPromise<T> onTimeout(T value, Duration duration, boolean cancelOnTimeout, boolean enlistOrigin) {
-        Promise<T> onTimeout = Timeouts.delayed(value, duration);
-        // Use *Async to execute on default "this" executor
-        DependentPromise<T> result = applyToEitherAsync(onTimeout, Function.identity(), enlistParamOrAll(enlistOrigin));
+        Promise<Try<T>> onTimeout = Timeouts.delayed(Try.success(value), duration);
+        DependentPromise<T> result = 
+        this.thenApply(Try::success, enlistOrigin)
+            .exceptionally(Try::failure, true)
+            // Use *Async to execute on default "this" executor
+            .applyToEitherAsync(onTimeout, Try::done, PromiseOrigin.ALL);
 
         result.whenComplete(Timeouts.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
         return result;
@@ -187,15 +189,15 @@ public DependentPromise<T> onTimeout(Supplier<? extends T> supplier, Duration du
 
     @Override
     public DependentPromise<T> onTimeout(Supplier<? extends T> supplier, Duration duration, boolean cancelOnTimeout, boolean enlistOrigin) {
-        Function<T, Supplier<? extends T>> valueToSupplier = v -> supply(v);
-        
         // timeout converted to supplier
-        Promise<Supplier<? extends T>> onTimeout = Timeouts.delayed(supplier, duration);
+        Promise<Supplier<Try<T>>> onTimeout = Timeouts.delayed(Try.with(supplier), duration);
 
         DependentPromise<T> result = 
-        this.thenApply(valueToSupplier, enlistOrigin) 
+        this.thenApply(Try::success, enlistOrigin)
+            .exceptionally(Try::failure, true)
+            .thenApply(SharedFunctions::supply, true)
             // Use *Async to execute on default "this" executor
-            .applyToEitherAsync(onTimeout, Supplier::get,  PromiseOrigin.ALL);
+            .applyToEitherAsync(onTimeout, s -> s.get().done(),  PromiseOrigin.ALL);
 
         result.whenComplete(Timeouts.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
         return result;
@@ -715,10 +717,6 @@ private CompletionStage<?>[] originAndParam(CompletionStage<?> param, Set<Promis
     private boolean defaultEnlistOrigin() {
         return defaultEnlistOptions.contains(PromiseOrigin.THIS);
     }
-    
-    private static Set<PromiseOrigin> enlistParamOrAll(boolean enlistThis) {
-        return enlistThis ? PromiseOrigin.ALL : PromiseOrigin.PARAM_ONLY;
-    }
 
     static void cancelPromises(CompletionStage<?>[] promises, boolean mayInterruptIfRunning) {
         if (null != promises) {

src/main/java/net/tascalate/concurrent/Promise.java

@@ -16,8 +16,6 @@
 package net.tascalate.concurrent;
 
 import static net.tascalate.concurrent.SharedFunctions.selectFirst;
-import static net.tascalate.concurrent.SharedFunctions.supply;
-import static net.tascalate.concurrent.SharedFunctions.timeout;
 import static net.tascalate.concurrent.SharedFunctions.unwrapExecutionException;
 import static net.tascalate.concurrent.SharedFunctions.wrapCompletionException;
 
@@ -52,7 +50,7 @@
 public interface Promise<T> extends Future<T>, CompletionStage<T> {
 
     default T getNow(T valueIfAbsent) throws CancellationException, CompletionException {
-        return getNow(supply(valueIfAbsent));
+        return getNow(SharedFunctions.supply(valueIfAbsent));
     }
 
     default T getNow(Supplier<? extends T> valueIfAbsent) throws CancellationException, CompletionException {
@@ -125,12 +123,13 @@ default Promise<T> orTimeout(Duration duration) {
     }
 
     default Promise<T> orTimeout(Duration duration, boolean cancelOnTimeout) {
-        Promise<? extends Reference<T>> onTimeout = Timeouts.delayed(null, duration);
+        Promise<? extends Try<T>> onTimeout = Timeouts.delayed(null, duration);
         Promise<T> result = 
         this.dependent()
-            .thenApply(Reference::new, false)
+            .thenApply(Try::success, false)
+            .exceptionally(Try::failure, true)
             // Use *Async to execute on default "this" executor
-            .applyToEitherAsync(onTimeout, v -> Reference.getOrElse(v, timeout(duration)), PromiseOrigin.ALL);
+            .applyToEitherAsync(onTimeout, v -> Try.doneOrTimeout(v, duration), PromiseOrigin.ALL);
 
         result.whenComplete(Timeouts.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
         return result.raw();
@@ -150,11 +149,13 @@ default Promise<T> onTimeout(T value, Duration duration) {
 
     default Promise<T> onTimeout(T value, Duration duration, boolean cancelOnTimeout) {
         // timeout converted to supplier
-        Promise<T> onTimeout = Timeouts.delayed(value, duration);
-        // Use *Async to execute on default "this" executor
+        Promise<Try<T>> onTimeout = Timeouts.delayed(Try.success(value), duration);
         Promise<T> result = 
         this.dependent()
-            .applyToEitherAsync(onTimeout, Function.identity(), PromiseOrigin.PARAM_ONLY);
+            .thenApply(Try::success, false)
+            .exceptionally(Try::failure, true)
+            // Use *Async to execute on default "this" executor
+            .applyToEitherAsync(onTimeout, Try::done, PromiseOrigin.ALL);
 
         result.whenComplete(Timeouts.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
         return result.raw();
@@ -173,17 +174,17 @@ default Promise<T> onTimeout(Supplier<? extends T> supplier, Duration duration)
     }
 
     default Promise<T> onTimeout(Supplier<? extends T> supplier, Duration duration, boolean cancelOnTimeout) {
-        Function<T, Supplier<? extends T>> valueToSupplier = v -> supply(v);
-        
         // timeout converted to supplier
-        Promise<Supplier<? extends T>> onTimeout = Timeouts.delayed(supplier, duration);
+        Promise<Supplier<Try<T>>> onTimeout = Timeouts.delayed(Try.with(supplier), duration);
 
         Promise<T> result = 
         this.dependent()
-            // resolved value converted to supplier
-            .thenApply(valueToSupplier, false)
+            // resolved value converted to supplier of Try
+            .thenApply(Try::success, false)
+            .exceptionally(Try::failure, true)
+            .thenApply(SharedFunctions::supply, true)
             // Use *Async to execute on default "this" executor
-            .applyToEitherAsync(onTimeout, Supplier::get, PromiseOrigin.ALL);
+            .applyToEitherAsync(onTimeout, s -> s.get().done(), PromiseOrigin.ALL);
 
         result.whenComplete(Timeouts.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
         return result.raw();

src/main/java/net/tascalate/concurrent/Reference.java

@@ -16,20 +16,17 @@
 package net.tascalate.concurrent;
 
 import java.lang.reflect.Method;
-import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
-import java.util.concurrent.TimeoutException;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
-
 class SharedFunctions {
 
     private SharedFunctions() {}
@@ -100,10 +97,6 @@ static <T> Supplier<T> supply(T value) {
         return () -> value;
     }
 
-    static Supplier<TimeoutException> timeout(Duration duration) {
-        return supply(new TimeoutException("Timeout after " + duration));        
-    }
-    
     @SuppressWarnings("unchecked")
     static <T, E extends Throwable> T sneakyThrow(Throwable e) throws E {
         throw (E) e;

src/main/java/net/tascalate/concurrent/SharedFunctions.java

@@ -15,6 +15,10 @@
  */
 package net.tascalate.concurrent;
 
+import java.time.Duration;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
 abstract class Try<R> {
 
     abstract R done();
@@ -55,7 +59,24 @@ static <R> Try<R> success(R result) {
     static <R> Try<R> failure(Throwable error) {
         return new Failure<R>(error);
     }
- 
+    
+    static <R> Supplier<Try<R>> with(Supplier<? extends R> supplier) {
+        return () -> of(supplier);
+    }
+    
+    static <R> Try<R> of(Supplier<? extends R> supplier) {
+        try {
+            return success(supplier.get());
+        } catch (Throwable ex) {
+            return failure(ex);
+        }
+    }
+    
+    static <R> R doneOrTimeout(Try<R> result, Duration duration) {
+        Try<R> checkedResult = null != result ? result: failure(new TimeoutException("Timeout after " + duration));
+        return checkedResult.done();        
+    }
+    
     @SuppressWarnings("unchecked")
     static <R> Try<R> nothing() {
         return (Try<R>)NOTHING;

src/main/java/net/tascalate/concurrent/Try.java

@@ -131,7 +131,7 @@ public static void main(final String[] argv) throws InterruptedException, Execut
                 System.out.println("After delay: " + v);
                 return v;
             }, true)
-            .onTimeout(123456789, Duration.ofMillis(2000))
+            .onTimeout(() -> 123456789, Duration.ofMillis(2000))
             .thenAcceptAsync(J8Examples::onComplete)
             .get(); 
 

src/test/java/net/tascalate/concurrent/J8Examples.java

@@ -3,6 +3,7 @@
 import static org.junit.Assert.assertTrue;
 
 import java.time.Duration;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.After;
@@ -35,13 +36,17 @@ public void testExceptionalOrTimeout() {
                 else
                     System.out.println("ERROR: " + err.getMessage());
              });
-        p.join();
+        try {
+            p.join();
+        } catch (CompletionException ex) {
+            assertTrue(ex.getCause() instanceof IllegalStateException);
+        }
     }
 
     private String doTask () throws Exception {
         TimeUnit.SECONDS.sleep(3);
         if (null != System.out)
-            throw new Exception("my error");
+            throw new IllegalStateException("my error");
         return "executed ok";
     }