Fix onCancel in ConfigurableDependentPromise; making wrappers more smart in decorators; new DelayPolicies.

Author: vsilaev

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

@@ -67,7 +67,7 @@ public CompletableFuture<T> toCompletableFuture() {
     }
 
     @Override
-    protected <U> Promise<U> wrap(CompletionStage<U> original) {
+    protected <U> Promise<U> wrapNew(CompletionStage<U> original) {
         return new CompletableFutureWrapper<>((CompletableFuture<U>)original);
     }
 

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

@@ -78,7 +78,7 @@ public Promise<T> completeAsync(Supplier<? extends T> supplier, Executor executo
     }
 
     public CompletablePromise<T> copy() {
-        CompletablePromise<T> result = wrap(new CompletableFuture<>());
+        CompletablePromise<T> result = wrapNew(new CompletableFuture<>());
         if (isDone()) {
             try {
                 result.complete(join());
@@ -125,7 +125,7 @@ public CompletableFuture<T> toCompletableFuture() {
     }
 
     @Override
-    protected <U> CompletablePromise<U> wrap(CompletionStage<U> original) {
+    protected <U> CompletablePromise<U> wrapNew(CompletionStage<U> original) {
         return new CompletablePromise<>((CompletableFuture<U>)original);
     }
 
@@ -135,7 +135,7 @@ public MinimalCompletionStage(CompletionStage<T> delegate) {
         }
 
         @Override
-        protected <U> CompletionStage<U> wrap(CompletionStage<U> original) {
+        protected <U> CompletionStage<U> wrapNew(CompletionStage<U> original) {
             return new MinimalCompletionStage<>(original);
         }
 

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

@@ -142,7 +142,7 @@ public EnforcedExecutorBoundPromise(Promise<T> delegate) {
                 }
 
                 @Override
-                protected <U> Promise<U> wrap(CompletionStage<U> original) {
+                protected <U> Promise<U> wrapNew(CompletionStage<U> original) {
                     return new EnforcedExecutorBoundPromise<>((Promise<U>)original);
                 }
 

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

@@ -123,7 +123,7 @@ public T join() {
     }
 
     @Override
-    protected <U> Promise<U> wrap(CompletionStage<U> original) {
+    protected final <U> Promise<U> wrapNew(CompletionStage<U> original) {
         return Promises.from(original);
     }
 
@@ -164,7 +164,7 @@ public Promise<T> raw() {
         }
 
         @Override
-        protected <U> Promise<U> wrap(CompletionStage<U> original) {
+        protected <U> Promise<U> wrapNew(CompletionStage<U> original) {
             return new StrictPromise<>((Promise<U>)original);
         }
 

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

@@ -17,7 +17,6 @@
 
 import static net.tascalate.concurrent.SharedFunctions.cancelPromise;
 import static net.tascalate.concurrent.SharedFunctions.iif;
-import static net.tascalate.concurrent.SharedFunctions.whenCancel;
 
 import java.time.Duration;
 import java.util.Arrays;
@@ -116,11 +115,17 @@ private static <U> DependentPromise<U> doWrap(Promise<U> original,
         }
 
         return new ConfigurableDependentPromise<>(
-            noOrigins ? original : whenCancel(original, () -> cancelPromises(cancellableOrigins, true)), 
+            noOrigins ? original : original.onCancel(() -> cancelPromises(cancellableOrigins, true)), 
             defaultEnlistOptions, cancellableOrigins
         );
     }
 
+    @Override
+    public DependentPromise<T> onCancel(Runnable action) {
+        delegate.onCancel(action);
+        return this;
+    }
+    
     // All delay overloads delegate to these methods
     @Override
     public DependentPromise<T> delay(Duration duration, boolean delayOnError) {

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

@@ -31,6 +31,8 @@
 import net.tascalate.concurrent.delays.ExponentialDelayPolicy;
 import net.tascalate.concurrent.delays.FirstRetryNoDelayPolicy;
 import net.tascalate.concurrent.delays.FixedIntervalDelayPolicy;
+import net.tascalate.concurrent.delays.OnFailureNoDelayPolicy;
+import net.tascalate.concurrent.delays.OnSuccessNoDelayPolicy;
 import net.tascalate.concurrent.delays.ProportionalRandomDelayPolicy;
 import net.tascalate.concurrent.delays.UniformRandomDelayPolicy;
 
@@ -132,6 +134,14 @@ default DelayPolicy<T> withFirstRetryNoDelay() {
         return new FirstRetryNoDelayPolicy<>(this);
     }
 
+    default DelayPolicy<T> withOnFailureNoDelay() {
+        return new OnFailureNoDelayPolicy<>(this);
+    }
+    
+    default DelayPolicy<T> withOnSuccessNoDelay() {
+        return new OnSuccessNoDelayPolicy<>(this);
+    }
+    
     default <D extends DelayPolicy<T>> D withCusomizer(Function<? super DelayPolicy<T>, D> fn) {
         return fn.apply(this);
     }

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

@@ -88,12 +88,7 @@ public static <U> DependentPromise<U> from(Promise<U> source, Set<PromiseOrigin>
     }
 
     @Override
-    default DependentPromise<T> onCancel(Runnable action) {
-        // Safe here regardless of origins used
-        // Max is self-origin, but whenComplete used 
-        // when self is completed already
-        return SharedFunctions.whenCancel(this, action);
-    }
+    DependentPromise<T> onCancel(Runnable action);
 
     @Override
     default DependentPromise<T> defaultAsyncOn(Executor executor) {

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

@@ -19,7 +19,6 @@
 import static net.tascalate.concurrent.SharedFunctions.failure;
 import static net.tascalate.concurrent.SharedFunctions.selectFirst;
 import static net.tascalate.concurrent.SharedFunctions.unwrapExecutionException;
-import static net.tascalate.concurrent.SharedFunctions.whenCancel;
 import static net.tascalate.concurrent.SharedFunctions.wrapCompletionException;
 
 import java.time.Duration;
@@ -101,7 +100,17 @@ default boolean isCompletedExceptionally() {
     }
 
     default Promise<T> onCancel(Runnable action) {
-        return whenCancel(this, action);
+        if (isCancelled()) {
+            action.run();
+        } else if (!isDone()) {
+            exceptionally(__ -> {
+               if (isCancelled()) {
+                   action.run();
+               }
+               return null;
+            });
+        }
+        return this;
     }
 
     default Promise<T> delay(long timeout, TimeUnit unit) {

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

@@ -18,11 +18,13 @@
 import java.util.EnumSet;
 import java.util.Set;
 
+import static java.util.Collections.unmodifiableSet;
+
 public enum PromiseOrigin {
     THIS, PARAM;
 
-    public static Set<PromiseOrigin> ALL = EnumSet.of(THIS, PARAM);
-    public static Set<PromiseOrigin> NONE = EnumSet.noneOf(PromiseOrigin.class);
-    public static Set<PromiseOrigin> THIS_ONLY = EnumSet.of(THIS);
-    public static Set<PromiseOrigin> PARAM_ONLY = EnumSet.of(PARAM);
+    public static final Set<PromiseOrigin> ALL        = unmodifiableSet(EnumSet.of(THIS, PARAM));
+    public static final Set<PromiseOrigin> NONE       = unmodifiableSet(EnumSet.noneOf(PromiseOrigin.class));
+    public static final Set<PromiseOrigin> THIS_ONLY  = unmodifiableSet(EnumSet.of(THIS));
+    public static final Set<PromiseOrigin> PARAM_ONLY = unmodifiableSet(EnumSet.of(PARAM));
 }

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

@@ -68,22 +68,7 @@ static <T> CompletionStage<T> failure(Function<? super T, Throwable> errorSuppli
         result.completeExceptionally(errorSupplier.apply(value));
         return result;
     }
-    
-    static <T, D extends Promise<T>> D whenCancel(D promise, Runnable action) {
-        if (promise.isCancelled()) {
-            action.run();
-        } else if (promise.isDone()) {
-            //
-        } else {
-            promise.whenComplete((r, e) -> {
-               if (promise.isCancelled()) {
-                   action.run();
-               }
-            });
-        }
-        return promise;        
-    }
-    
+
     static boolean cancelPromise(CompletionStage<?> promise, boolean mayInterruptIfRunning) {
         if (promise instanceof Future) {
             Future<?> future = (Future<?>) promise;