Promise.delay methods; refactoring timeout-related stuff to the separate internal class

Author: vsilaev

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

@@ -371,7 +371,7 @@ private <R, U> Promise<U> doApplyToEitherAsync(CompletionStage<? extends R> firs
             if (failure == null) {
                 nextStage.onSuccess(result);
             } else {
-                nextStage.onError(wrapException(failure));
+                nextStage.onError(Promises.wrapException(failure));
             }
         };
         // only the first result is accepted by completion stage,
@@ -403,15 +403,7 @@ private static <R> Function<R, Void> runnableAsFunction(Runnable action) {
     }
 
     private static <U> U forwardException(Throwable e) {
-        throw wrapException(e);
-    }
-
-    private static CompletionException wrapException(Throwable e) {
-        if (e instanceof CompletionException) {
-            return (CompletionException) e;
-        } else {
-            return new CompletionException(e);
-        }
+        throw Promises.wrapException(e);
     }
 
     private static ExecutionException rewrapExecutionException(ExecutionException ex) {

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

@@ -15,9 +15,12 @@
  */
 package net.tascalate.concurrent;
 
+import java.time.Duration;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Executor;
 import java.util.concurrent.RunnableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 /**
@@ -141,7 +144,26 @@ public static <U> Promise<U> supplyAsync(Supplier<U> supplier, Executor executor
         executor.execute(result);
         return result;
     }
-
+    
+    public static <T> Promise<T> await(CompletionStage<? extends T> stage, Executor executor) {
+        return await(stage, executor, false);
+    }
+    
+    public static <T> Promise<T> await(CompletionStage<? extends T> stage, Executor executor, boolean enlistOrigin) {
+        return asyncOn(executor).dependent()
+               .thenCombineAsync(
+                   stage, (u, v) -> v, enlistOrigin ? PromiseOrigin.PARAM_ONLY : PromiseOrigin.NONE
+               );
+    }
+    
+    public static Promise<Duration> delay(long timeout, TimeUnit unit, Executor executor) {
+        return delay(Timeouts.toDuration(timeout, unit), executor);
+    }
+    
+    public static Promise<Duration> delay(Duration duration, Executor executor) {
+        return await(Timeouts.delay(duration), executor, true);
+    }
+    
     @Override
     Runnable setupTransition(Callable<T> code) {
         throw new UnsupportedOperationException();

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

@@ -149,6 +149,34 @@ private static <U> DependentPromise<U> doWrap(Promise<U> original, CompletionSta
         return result;
     }
 
+    public DependentPromise<T> delay(long timeout, TimeUnit unit) {
+        return delay(timeout, unit, true);
+    }
+    
+    public DependentPromise<T> delay(long timeout, TimeUnit unit, boolean delayOnError) {
+        return delay(timeout, unit, delayOnError, false);
+    }
+    
+    public DependentPromise<T> delay(long timeout, TimeUnit unit, boolean delayOnError, boolean enlistOrigin) {
+        return delay(Timeouts.toDuration(timeout, unit), delayOnError, enlistOrigin);
+    }
+    
+    public DependentPromise<T> delay(Duration duration) {
+        return delay(duration, true);
+    }
+    
+    public DependentPromise<T> delay(Duration duration, boolean delayOnError) {
+        return delay(duration, delayOnError, false);
+    }
+    
+    public DependentPromise<T> delay(Duration duration, boolean delayOnError, boolean enlistOrigin) {
+        CompletablePromise<?> delayed = new CompletablePromise<>();
+        whenComplete(Timeouts.configureDelay(this, delayed, duration, delayOnError));
+        // Use *async to execute on default "this" executor
+        return thenCombineAsync(
+            delayed, (r, d) -> r, enlistOrigin ? PromiseOrigin.ALL : PromiseOrigin.PARAM_ONLY
+        );
+    }
 
     public DependentPromise<T> orTimeout(long timeout, TimeUnit unit) {
         return orTimeout(timeout, unit, true);
@@ -159,7 +187,7 @@ public DependentPromise<T> orTimeout(long timeout, TimeUnit unit, boolean cancel
     }
 
     public DependentPromise<T> orTimeout(long timeout, TimeUnit unit, boolean cancelOnTimeout, boolean enlistOrigin) {
-        return orTimeout(Promises.toDuration(timeout, unit), cancelOnTimeout, enlistOrigin);
+        return orTimeout(Timeouts.toDuration(timeout, unit), cancelOnTimeout, enlistOrigin);
     }
 
     public DependentPromise<T> orTimeout(Duration duration) {
@@ -171,11 +199,11 @@ public DependentPromise<T> orTimeout(Duration duration, boolean cancelOnTimeout)
     }
 
     public DependentPromise<T> orTimeout(Duration duration, boolean cancelOnTimeout, boolean enlistOrigin) {
-        Promise<T> onTimeout = Promises.failAfter(duration);
+        Promise<T> onTimeout = Timeouts.failAfter(duration);
         // Use *async to execute on default "this" executor
         DependentPromise<T> result = this
             .applyToEitherAsync(onTimeout, Function.identity(), enlistOrigin ? PromiseOrigin.ALL : PromiseOrigin.PARAM_ONLY);
-        result.whenComplete(Promises.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
+        result.whenComplete(Timeouts.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
         return result;
     }
 
@@ -188,7 +216,7 @@ public DependentPromise<T> onTimeout(T value, long timeout, TimeUnit unit, boole
     }
 
     public DependentPromise<T> onTimeout(T value, long timeout, TimeUnit unit, boolean cancelOnTimeout, boolean enlistOrigin) {
-        return onTimeout(value, Promises.toDuration(timeout, unit), cancelOnTimeout, enlistOrigin);
+        return onTimeout(value, Timeouts.toDuration(timeout, unit), cancelOnTimeout, enlistOrigin);
     }
 
     public DependentPromise<T> onTimeout(T value, Duration duration) {
@@ -208,11 +236,11 @@ public DependentPromise<T> onTimeout(Supplier<? extends T> supplier, long timeou
     }
 
     public DependentPromise<T> onTimeout(Supplier<? extends T> supplier, long timeout, TimeUnit unit, boolean cancelOnTimeout) {
-        return onTimeout(supplier, Promises.toDuration(timeout, unit), false);
+        return onTimeout(supplier, Timeouts.toDuration(timeout, unit), false);
     }
 
     public DependentPromise<T> onTimeout(Supplier<? extends T> supplier, long timeout, TimeUnit unit, boolean cancelOnTimeout, boolean enlistOrigin) {
-        return onTimeout(supplier, Promises.toDuration(timeout, unit), enlistOrigin);
+        return onTimeout(supplier, Timeouts.toDuration(timeout, unit), enlistOrigin);
     }
 
     public DependentPromise<T> onTimeout(Supplier<? extends T> supplier, Duration duration) {
@@ -227,7 +255,7 @@ public DependentPromise<T> onTimeout(Supplier<? extends T> supplier, Duration du
         Function<T, Supplier<? extends T>> valueToSupplier = v -> () -> v;
 
         // timeout converted to supplier
-        Promise<Supplier<? extends T>> onTimeout = Promises
+        Promise<Supplier<? extends T>> onTimeout = Timeouts
             .delay(duration)
             .dependent()
             .thenApply(d -> supplier, true);
@@ -238,7 +266,7 @@ public DependentPromise<T> onTimeout(Supplier<? extends T> supplier, Duration du
             // Use *async to execute on default "this" executor
             .applyToEitherAsync(onTimeout, Supplier::get,  PromiseOrigin.ALL);
 
-        result.whenComplete(Promises.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
+        result.whenComplete(Timeouts.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
         return result;
     }
 

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

@@ -59,26 +59,48 @@ default T getNow(Supplier<? extends T> valueIfAbsent) {
             return valueIfAbsent.get();
         }
     }
+
+    default Promise<T> delay(long timeout, TimeUnit unit) {
+        return delay(timeout, unit, true);
+    }
+    
+    default Promise<T> delay(long timeout, TimeUnit unit, boolean delayOnError) {
+        return delay(Timeouts.toDuration(timeout, unit), delayOnError);
+    }
+    
+    default Promise<T> delay(Duration duration) {
+        return delay(duration, true);
+    }
+    
+    default Promise<T> delay(Duration duration, boolean delayOnError) {
+        CompletablePromise<?> delayed = new CompletablePromise<>();
+        whenComplete(Timeouts.configureDelay(this, delayed, duration, delayOnError));
+        // Use *async to execute on default "this" executor
+        return dependent().thenCombineAsync(
+            delayed, (r, d) -> r, PromiseOrigin.PARAM_ONLY
+        );
+    }
+
 
     default Promise<T> orTimeout(long timeout, TimeUnit unit) {
         return orTimeout(timeout, unit, true);
     }
 
     default Promise<T> orTimeout(long timeout, TimeUnit unit, boolean cancelOnTimeout) {
-        return orTimeout(Promises.toDuration(timeout, unit), cancelOnTimeout);
+        return orTimeout(Timeouts.toDuration(timeout, unit), cancelOnTimeout);
     }
 
     default Promise<T> orTimeout(Duration duration) {
         return orTimeout(duration, true);
     }
 
     default Promise<T> orTimeout(Duration duration, boolean cancelOnTimeout) {
-        Promise<T> onTimeout = Promises.failAfter(duration);
+        Promise<T> timeout = Timeouts.failAfter(duration);
         // Use *async to execute on default "this" executor
         Promise<T> result = dependent()
-            .applyToEitherAsync(onTimeout, Function.identity(), PromiseOrigin.PARAM_ONLY);
+            .applyToEitherAsync(timeout, Function.identity(), PromiseOrigin.PARAM_ONLY);
 
-        result.whenComplete(Promises.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
+        result.whenComplete(Timeouts.timeoutsCleanup(this, timeout, cancelOnTimeout));
         return result;
     }
 
@@ -87,7 +109,7 @@ default Promise<T> onTimeout(T value, long timeout, TimeUnit unit) {
     }
 
     default Promise<T> onTimeout(T value, long timeout, TimeUnit unit, boolean cancelOnTimeout) {
-        return onTimeout(value, Promises.toDuration(timeout, unit));
+        return onTimeout(value, Timeouts.toDuration(timeout, unit));
     }
 
     default Promise<T> onTimeout(T value, Duration duration) {
@@ -103,7 +125,7 @@ default Promise<T> onTimeout(Supplier<? extends T> supplier, long timeout, TimeU
     }
 
     default Promise<T> onTimeout(Supplier<? extends T> supplier, long timeout, TimeUnit unit, boolean cancelOnTimeout) {
-        return onTimeout(supplier, Promises.toDuration(timeout, unit), cancelOnTimeout);
+        return onTimeout(supplier, Timeouts.toDuration(timeout, unit), cancelOnTimeout);
     }
 
     default Promise<T> onTimeout(Supplier<? extends T> supplier, Duration duration) {
@@ -114,7 +136,7 @@ default Promise<T> onTimeout(Supplier<? extends T> supplier, Duration duration,
         Function<T, Supplier<? extends T>> valueToSupplier = v -> () -> v;
 
         // timeout converted to supplier
-        Promise<Supplier<? extends T>> onTimeout = Promises
+        Promise<Supplier<? extends T>> timeout = Timeouts
             .delay(duration)
             .dependent()
             .thenApply(d -> supplier, true);
@@ -123,9 +145,9 @@ default Promise<T> onTimeout(Supplier<? extends T> supplier, Duration duration,
             // resolved value converted to supplier
             .thenApply(valueToSupplier, false)
             // Use *async to execute on default "this" executor
-            .applyToEitherAsync(onTimeout, Supplier::get, PromiseOrigin.ALL);
+            .applyToEitherAsync(timeout, Supplier::get, PromiseOrigin.ALL);
 
-        result.whenComplete(Promises.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
+        result.whenComplete(Timeouts.timeoutsCleanup(this, timeout, cancelOnTimeout));
         return result;
     }