Better type bounds for Promises.retry(RetryCallable<..>...)

Author: vsilaev

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

@@ -449,45 +449,67 @@ public static <T> Promise<List<T>> atLeast(int minResultsCount, int maxErrorsCou
         }
     }
 
-    public static Promise<Void> retry(Runnable codeBlock, Executor executor, RetryPolicy<? super Void> retryPolicy) {
+    public static Promise<Void> retry(Runnable codeBlock, Executor executor, 
+                                      RetryPolicy<? super Void> retryPolicy) {
+        
         return retry(ctx -> { codeBlock.run(); }, executor, retryPolicy);
     }
 
-    public static Promise<Void> retry(RetryRunnable codeBlock, Executor executor, RetryPolicy<? super Void> retryPolicy) {
+    public static Promise<Void> retry(RetryRunnable codeBlock, Executor executor, 
+                                      RetryPolicy<? super Void> retryPolicy) {
+        
         return retry(ctx -> { codeBlock.run(ctx); return null; }, executor, retryPolicy.acceptNullResult());
     }
 
-    public static <T> Promise<T> retry(Callable<T> codeBlock, Executor executor, RetryPolicy<? super T> retryPolicy) {
+    public static <T> Promise<T> retry(Callable<T> codeBlock, Executor executor, 
+                                       RetryPolicy<? super T> retryPolicy) {
+        
         return retry(toRetryCallable(codeBlock), executor, retryPolicy);
     }
 
-    public static <T> Promise<T> retry(RetryCallable<T, T> codeBlock, Executor executor, RetryPolicy<? super T> retryPolicy) {
-        return invokePoller(
-            retryPolicy, 
-            (ctx, result, cancellation) -> pollValueOnce(codeBlock, executor, ctx, result, cancellation)
+    public static <C, T extends C> Promise<T> retry(RetryCallable<T, C> codeBlock, Executor executor, 
+                                                    RetryPolicy<? super C> retryPolicy) {
+        
+        // Need type on lambda param to please Eclipse compiler        
+        return invokePoller(retryPolicy, (RetryContext<C> ctx, 
+                                          CompletableFuture<T> result, 
+                                          Consumer<Promise<?>> cancellation) 
+                                          -> pollValueOnce(codeBlock, executor, ctx, result, cancellation)
         );
     }
 
-    public static <T> Promise<T> retryOptional(Callable<Optional<T>> codeBlock, Executor executor, RetryPolicy<? super T> retryPolicy) {
+    public static <T> Promise<T> retryOptional(Callable<Optional<T>> codeBlock, Executor executor, 
+                                               RetryPolicy<? super T> retryPolicy) {
+        
         return retryOptional(toRetryCallable(codeBlock), executor, retryPolicy);
     }
 
-    public static <T> Promise<T> retryOptional(RetryCallable<Optional<T>, T> codeBlock, Executor executor, RetryPolicy<? super T> retryPolicy) {
-        return retry(ctx -> codeBlock.call(ctx).orElse(null), executor, retryPolicy);
+    public static <C, T extends C> Promise<T> retryOptional(RetryCallable<Optional<T>, C> codeBlock, Executor executor, 
+                                                            RetryPolicy<? super C> retryPolicy) {
+        
+        // Need type on lambda param to please Eclipse compiler
+        return retry((RetryContext<C> ctx) -> codeBlock.call(ctx).orElse(null), executor, retryPolicy);
     }
 
-    public static <T> Promise<T> retryFuture(Callable<? extends CompletionStage<T>> invoker, RetryPolicy<? super T> retryPolicy) {
+    public static <T> Promise<T> retryFuture(Callable<? extends CompletionStage<T>> invoker, 
+                                             RetryPolicy<? super T> retryPolicy) {
+        
         return retryFuture(toRetryCallable(invoker), retryPolicy);
     }
 
-    public static <T> Promise<T> retryFuture(RetryCallable<? extends CompletionStage<T>, T> invoker, RetryPolicy<? super T> retryPolicy) {
-        return invokePoller(
-            retryPolicy, 
-            (ctx, result, cancellation) -> pollFutureOnce(invoker, ctx, result, cancellation, null)
+    public static <C, T extends C> Promise<T> retryFuture(RetryCallable<? extends CompletionStage<T>, C> futureFactory, 
+                                                          RetryPolicy<? super C> retryPolicy) {
+        
+        return invokePoller(retryPolicy, (RetryContext<C> ctx, 
+                                          CompletableFuture<T> result, 
+                                          Consumer<Promise<?>> cancellation) 
+                                          -> pollFutureOnce(futureFactory, ctx, result, cancellation, null)
         );
     }
 
-    private static <T> Promise<T> invokePoller(RetryPolicy<? super T> retryPolicy, F3<RetryContext<T>, CompletableFuture<T>, Consumer<Promise<?>>> initiator) {
+    private static <C, T extends C> Promise<T> invokePoller(RetryPolicy<? super C> retryPolicy, 
+                                                            F3<RetryContext<C>, CompletableFuture<T>, Consumer<Promise<?>>> initiator) {
+        
         final CompletableFuture<T> result = new CompletableFuture<>();
         final AtomicReference<Promise<?>> callPromiseRef = new AtomicReference<>();
         // Cleanup latest timeout on completion;
@@ -506,16 +528,16 @@ private static <T> Promise<T> invokePoller(RetryPolicy<? super T> retryPolicy, F
             }
         };
 
-        RetryContext<T> ctx = RetryContext.initial(retryPolicy);     
+        RetryContext<C> ctx = RetryContext.initial(retryPolicy);     
         initiator.apply(ctx, result, cancellation);
         return new CompletableFutureWrapper<>(result);
     }
 
-    private static <T> void pollValueOnce(RetryCallable<T, T> codeBlock, 
-                                          Executor executor, 
-                                          RetryContext<T> ctx, 
-                                          CompletableFuture<T> result, 
-                                          Consumer<Promise<?>> cancellation) {
+    private static <C, T extends C> void pollValueOnce(RetryCallable<T, C> codeBlock, 
+                                                       Executor executor, 
+                                                       RetryContext<C> ctx, 
+                                                       CompletableFuture<T> result, 
+                                                       Consumer<Promise<?>> cancellation) {
 
         // Promise may be cancelled outside of polling
         if (result.isDone()) {
@@ -533,12 +555,12 @@ private static <T> void pollValueOnce(RetryCallable<T, T> codeBlock,
                             result.complete(value);
                         } else {
                             long finishTime = System.nanoTime();
-                            RetryContext<T> nextCtx = ctx.nextRetry(duration(startTime, finishTime), value);
+                            RetryContext<C> nextCtx = ctx.nextRetry(duration(startTime, finishTime), value);
                             pollValueOnce(codeBlock, executor, nextCtx, result, cancellation);
                         }
                     } catch (Exception ex) {
                         long finishTime = System.nanoTime();
-                        RetryContext<T> nextCtx = ctx.nextRetry(duration(startTime, finishTime), ex);
+                        RetryContext<C> nextCtx = ctx.nextRetry(duration(startTime, finishTime), ex);
                         pollValueOnce(codeBlock, executor, nextCtx, result, cancellation);
                     }
                 };
@@ -564,11 +586,11 @@ private static <T> void pollValueOnce(RetryCallable<T, T> codeBlock,
         }
     }
 
-    private static <T> void pollFutureOnce(RetryCallable<? extends CompletionStage<T>, T> invoker, 
-                                           RetryContext<T> ctx, 
-                                           CompletableFuture<T> result,
-                                           Consumer<Promise<?>> cancellation,
-                                           Promise<?> prev) {
+    private static <C, T extends C> void pollFutureOnce(RetryCallable<? extends CompletionStage<T>, C> futureFactory, 
+                                                        RetryContext<C> ctx, 
+                                                        CompletableFuture<T> result,
+                                                        Consumer<Promise<?>> cancellation,
+                                                        Promise<?> prev) {
         // Promise may be cancelled outside of polling
         if (result.isDone()) {
             return;
@@ -581,7 +603,7 @@ private static <T> void pollFutureOnce(RetryCallable<? extends CompletionStage<T
 
                 Promise<? extends T> target;
                 try {
-                    target = Promises.from(invoker.call(ctx));
+                    target = Promises.from(futureFactory.call(ctx));
                 } catch (Exception ex) {
                     target = Promises.failure(ex);
                 }
@@ -595,18 +617,18 @@ private static <T> void pollFutureOnce(RetryCallable<? extends CompletionStage<T
                         result.complete(value);
                     } else {
                         long finishTime = System.nanoTime();
-                        RetryContext<T> nextCtx = ctx.nextRetry(
+                        RetryContext<C> nextCtx = ctx.nextRetry(
                             duration(startTime, finishTime), unwrapCompletionException(ex)
                         );
                         boolean callLater = isRecursive.get() && Thread.currentThread() == invokerThread;
                         if (callLater) {
                             // Call after minimal possible delay
                             callLater(
                                 p, Duration.ofNanos(1), cancellation, 
-                                () -> pollFutureOnce(invoker, nextCtx, result, cancellation, p)
+                                () -> pollFutureOnce(futureFactory, nextCtx, result, cancellation, p)
                             );
                         } else {
-                            pollFutureOnce(invoker, nextCtx, result, cancellation, p);
+                            pollFutureOnce(futureFactory, nextCtx, result, cancellation, p);
                         }
                     }
                 });

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

@@ -15,6 +15,7 @@
  */
 package net.tascalate.concurrent;
 
+import java.math.BigInteger;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.concurrent.CompletionStage;
@@ -26,9 +27,19 @@
 import net.tascalate.concurrent.decorators.ExtendedPromiseDecorator;
 
 public class J8Examples {
+    
+    static BigInteger tryCalc(RetryContext<Number> ctx) {
+        return BigInteger.ONE;
+    }
 
     public static void main(final String[] argv) throws InterruptedException, ExecutionException {
         final TaskExecutorService executorService = TaskExecutors.newFixedThreadPool(6);
+        
+        Promise<BigInteger> tryTyping = Promises.retry(
+            J8Examples::tryCalc, executorService, 
+            RetryPolicy.<Number>create().withResultValidator(v -> v.intValue() > 0).withMaxRetries(2)
+        );
+        System.out.println( tryTyping.get() );
 
         Promise<Object> k = CompletableTask.supplyAsync(() -> produceStringSlow("-ABC"), executorService);
         //Promise<Object> k = CompletableTask.complete("ABC", executorService);
@@ -70,18 +81,6 @@ public static void main(final String[] argv) throws InterruptedException, Execut
         //p.cancel(true);
         p.get();
 
-        Promise<String> retry1 = Promises.retry(
-            () -> "ABC",
-            executorService, RetryPolicy.DEFAULT
-        );
-        
-        Promise<String> retry2 = Promises.retry(
-            ctx -> "ABC",
-            executorService, RetryPolicy.DEFAULT.rejectNullResult()
-        );
-
-        System.out.println(retry1.join() + " vs " + retry2.join());
-
         Promise<String> pollerFuture = Promises.retryFuture( 
             ctx -> pollingFutureMethod(ctx, executorService),
             RetryPolicy.DEFAULT