DelayPolicy subclasses; fix cancellation in orTimeout/onTimeout; support service-call timeout in polling

Author: vsilaev

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

@@ -0,0 +1,73 @@
+/**
+ * Original work: copyright 2013 Tomasz Nurkiewicz
+ * https://github.com/nurkiewicz/async-retry
+ * 
+ * This class is based on the work create by Tomasz Nurkiewicz 
+ * under the Apache License, Version 2.0. Please see 
+ * https://github.com/nurkiewicz/async-retry/blob/master/src/main/java/com/nurkiewicz/asyncretry/backoff/Backoff.java 
+ * 
+ * Modified work: copyright 2015-2017 Valery Silaev (http://vsilaev.com)
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package net.tascalate.concurrent;
+
+import net.tascalate.concurrent.delays.BoundedMaxDelayPolicy;
+import net.tascalate.concurrent.delays.BoundedMinDelayPolicy;
+import net.tascalate.concurrent.delays.FirstRetryNoDelayPolicy;
+import net.tascalate.concurrent.delays.FixedIntervalDelayPolicy;
+import net.tascalate.concurrent.delays.ProportionalRandomDelayPolicy;
+import net.tascalate.concurrent.delays.UniformRandomDelayPolicy;
+
+public interface DelayPolicy {
+    public static final DelayPolicy DEFAULT = new FirstRetryNoDelayPolicy(new FixedIntervalDelayPolicy());
+    public static final DelayPolicy INVALID_DELAY = ctx -> -1;
+    
+    abstract public long delayMillis(RetryContext context);
+    
+    default DelayPolicy withUniformJitter() {
+        return new UniformRandomDelayPolicy(this);
+    }
+
+    default DelayPolicy withUniformJitter(long range) {
+        return new UniformRandomDelayPolicy(this, range);
+    }
+
+    default DelayPolicy withProportionalJitter() {
+        return new ProportionalRandomDelayPolicy(this);
+    }
+
+    default DelayPolicy withProportionalJitter(double multiplier) {
+        return new ProportionalRandomDelayPolicy(this, multiplier);
+    }
+
+    default DelayPolicy withMinDelay(long minDelayMillis) {
+        return new BoundedMinDelayPolicy(this, minDelayMillis);
+    }
+
+    default DelayPolicy withMinDelay() {
+        return new BoundedMinDelayPolicy(this);
+    }
+
+    default DelayPolicy withMaxDelay(long maxDelayMillis) {
+        return new BoundedMaxDelayPolicy(this, maxDelayMillis);
+    }
+
+    default DelayPolicy withMaxDelay() {
+        return new BoundedMaxDelayPolicy(this);
+    }
+
+    default DelayPolicy withFirstRetryNoDelay() {
+        return new FirstRetryNoDelayPolicy(this);
+    }
+}

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

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2015-2017 Valery Silaev (http://vsilaev.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package net.tascalate.concurrent;
 
 import java.lang.reflect.Array;
@@ -159,16 +174,14 @@ public DependentPromise<T> orTimeout(Duration duration, boolean cancelOnTimeout,
         Promise<T> onTimeout = Promises.failAfter(duration);
         // Use *async to execute on default "this" executor
         return applyToEitherAsync(
-            onTimeout, 
-            v -> {
-                if (cancelOnTimeout) {
-                    cancel(true);
-                }
-                onTimeout.cancel(true); 
-                return v; 
-            }, 
+            onTimeout, Function.identity(), 
             enlistOrigin ? PromiseOrigin.ALL : PromiseOrigin.PARAM_ONLY
-        );        
+        ).whenComplete((v, e) -> {
+            if (cancelOnTimeout) {
+                cancel(true);
+            }
+            onTimeout.cancel(true); 
+        }, true);        
     }
 
     public DependentPromise<T> onTimeout(T value, long timeout, TimeUnit unit) {
@@ -225,17 +238,13 @@ public DependentPromise<T> onTimeout(Supplier<T> supplier, Duration duration, bo
             // resolved value converted to supplier
             .thenApply(valueToSupplier, enlistOrigin)
             // Use *async to execute on default "this" executor
-            .applyToEitherAsync(
-                onTimeout, 
-                s -> {
-                   if (cancelOnTimeout) {
-                       cancel(true);
-                   }
-                   onTimeout.cancel(true); 
-                   return s.get(); 
-                }, 
-                PromiseOrigin.ALL
-             );        
+            .applyToEitherAsync(onTimeout, Supplier::get,  PromiseOrigin.ALL)
+            .whenComplete((v, e) -> {
+                if (cancelOnTimeout) {
+                    cancel(true);
+                }
+                onTimeout.cancel(true);                 
+            }, true);        
     }
 
     public <U> DependentPromise<U> thenApply(Function<? super T, ? extends U> fn) {

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

@@ -75,16 +75,15 @@ default Promise<T> orTimeout(Duration duration) {
     default Promise<T> orTimeout(Duration duration, boolean cancelOnTimeout) {
         Promise<T> onTimeout = Promises.failAfter(duration);
         // Use *async to execute on default "this" executor
-        return Promises.dependent(this).applyToEitherAsync(
-            onTimeout, 
-            v -> {
+        return Promises.dependent(this)
+            .applyToEitherAsync(onTimeout, Function.identity(), PromiseOrigin.PARAM_ONLY)
+            .whenComplete((v, e) -> {
                 if (cancelOnTimeout) {
                     cancel(true);
                 }
                 onTimeout.cancel(true); 
-                return v; 
             }, 
-            PromiseOrigin.PARAM_ONLY
+            true
         );
     }
 
@@ -126,17 +125,13 @@ default Promise<T> onTimeout(Supplier<T> supplier, Duration duration, boolean ca
             // resolved value converted to supplier
             .thenApply(valueToSupplier, false)
             // Use *async to execute on default "this" executor
-            .applyToEitherAsync(
-                onTimeout, 
-                s -> {
-                   if (cancelOnTimeout) {
-                       cancel(true);
-                   }
-                   onTimeout.cancel(true); 
-                   return s.get(); 
-                }, 
-                PromiseOrigin.ALL
-             );
+            .applyToEitherAsync(onTimeout, Supplier::get, PromiseOrigin.ALL)
+            .whenComplete((v, e) -> {
+                if (cancelOnTimeout) {
+                    cancel(true);
+                }
+                onTimeout.cancel(true);
+            }, true);
     }
 
     public <U> Promise<U> thenApply(Function<? super T, ? extends U> fn);

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

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2015-2017 Valery Silaev (http://vsilaev.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package net.tascalate.concurrent;
 
 import java.util.EnumSet;

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

@@ -136,12 +136,9 @@ public static <T> DependentPromise<T> dependent(Promise<T> stage) {
         return DependentPromise.from(stage);
     }
 
-    public static Promise<Void> task(Executor executor) {
-        return CompletableTask.asyncOn(executor);
-    }
-    
     public static <T> Promise<T> task(CompletionStage<T> stage, Executor executor) {
-        return dependent(task(executor)).thenCombineAsync(stage, (u, v) -> v, PromiseOrigin.PARAM_ONLY);
+        return dependent(CompletableTask.asyncOn(executor))
+               .thenCombineAsync(stage, (u, v) -> v, PromiseOrigin.PARAM_ONLY);
     }
 
     private static <T, R> CompletablePromise<R> createLinkedPromise(CompletionStage<T> stage) {
@@ -531,7 +528,11 @@ public static <T> Promise<T> failAfter(long delay, TimeUnit timeUnit) {
 
 
     public static <T> Promise<T> poll(Callable<? extends T> codeBlock, Executor executor, RetryPolicy retryPolicy) {
-        return pollOptional(() -> Optional.ofNullable(codeBlock.call()), executor, retryPolicy);
+        Promise<ObjectRef<T>> wrappedResult = pollOptional(
+            () -> Optional.of(new ObjectRef<T>( codeBlock.call() )), 
+            executor, retryPolicy
+        );
+        return dependent(wrappedResult).thenApply(ObjectRef::dereference, true);
     }
 
     public static <T> Promise<T> pollOptional(Callable<Optional<? extends T>> codeBlock, Executor executor, RetryPolicy retryPolicy) {
@@ -560,8 +561,8 @@ private static <T> void pollOnce(Callable<Optional<? extends T>> codeBlock,
             return;
         }
 
-        long executionDelayMillis = ctx.executionDelayMillis();
-        if (executionDelayMillis >= 0) {
+        RetryPolicy.Outcome answer = ctx.shouldContinue();
+        if (answer.shouldExecute()) {
             Runnable doCall = () -> {
                 long startTime = System.currentTimeMillis();
                 try {
@@ -570,35 +571,43 @@ private static <T> void pollOnce(Callable<Optional<? extends T>> codeBlock,
                         resultPromise.onSuccess(result.get());
                     } else {
                         long finishTime = System.currentTimeMillis();
-                        RetryContext nextCtx = ctx.getNextRetry(finishTime - startTime);
+                        RetryContext nextCtx = ctx.nextRetry(finishTime - startTime);
                         pollOnce(codeBlock, executor, nextCtx, resultPromise, callPromiseRef);
                     }
                 } catch (Exception ex) {
                     long finishTime = System.currentTimeMillis();
-                    RetryContext nextCtx = ctx.getNextRetry(finishTime - startTime, ex);
+                    RetryContext nextCtx = ctx.nextRetry(finishTime - startTime, ex);
                     pollOnce(codeBlock, executor, nextCtx, resultPromise, callPromiseRef);
                 }
             };
 
             Promise<?> callPromise;
-            if (executionDelayMillis > 0) {
+            Promise<?> finalPromise;
+            long backoffDelayMillis = answer.backoffDelayMillis();
+            if (backoffDelayMillis > 0) {
                 // Timeout itself
-                Promise<?> timeout = delay(Duration.ofMillis(executionDelayMillis));
-                // Call should be done via CompletableTask to let it be interruptible
-                callPromise = dependent(task(executor)).runAfterBothAsync(timeout, doCall, PromiseOrigin.PARAM_ONLY);
+                Promise<?> backoff = delay(Duration.ofMillis(backoffDelayMillis));
+                // Call should be done via CompletableTask to let it be interruptible                
+                finalPromise = CompletableTask.asyncOn(executor).runAfterBothAsync(backoff, doCall);
+                callPromise  = backoff; // Canceling timeout will cancel the chain above 
             } else {
                 // Immediately send to executor
-                callPromise = CompletableTask.runAsync(doCall, executor); 
+                callPromise = finalPromise = CompletableTask.runAsync(doCall, executor); 
             }
             callPromiseRef.set(callPromise);
             // If result promise is cancelled after callPromise was set need to stop;
             if (resultPromise.isDone()) {
                 callPromise.cancel(true);
-                return;
+            } else if (answer.hasTimeout()) {
+                // Restrict execution time of the final promise
+                // Timeout should be a sum of policy timeout and policy backoff 
+                // while timer is set up immediately 
+                long totalTimeout = Math.max(0, answer.timeoutDelayMillis()) + Math.max(0, answer.backoffDelayMillis());
+                finalPromise.orTimeout(Duration.ofMillis(totalTimeout), true);
             }
 
         } else {
-            resultPromise.onFailure(ctx.getLastThrowable());
+            resultPromise.onFailure(ctx.asFailure());
         }
     }
 
@@ -667,4 +676,16 @@ public Thread newThread(Runnable r) {
             return result;
         }
     });
+    
+    private static class ObjectRef<T> {
+        private final T reference;
+        
+        ObjectRef(T reference) {
+            this.reference = reference;
+        }
+        
+        T dereference() {
+            return reference;
+        }
+    }
 }