More cancellation options for third-party CompletionStage impl. Revisiting Promises combiner methods to return exceptional future instead of throwing exception

Author: vsilaev

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

@@ -25,6 +25,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.Callable;
@@ -46,7 +47,7 @@
  * @author vsilaev
  *
  */
-public class Promises {
+public final class Promises {
 
     private Promises() {}
 
@@ -703,8 +704,13 @@ private static <T> Promise<T> insufficientNumberOfArguments(int minResultCount,
         String message = String.format(
             "The number of futures supplied (%d) is less than a number of futures to await (%d)", size, minResultCount
         );
-        //return failure(new NoSuchElementException(message));
+        Exception ex = new NoSuchElementException(message);
+        //TODO: exceptional completion vs runtime exception on combined promise construction?
+        ex.fillInStackTrace();
+        return failure(ex);
+        /*
         throw new IllegalArgumentException(message);        
+        */
     }
 
     private static <V, T> RetryCallable<V, T> toRetryCallable(Callable<? extends V> callable) {
@@ -715,10 +721,8 @@ private static Duration duration(long startTime, long finishTime) {
         return Duration.ofNanos(finishTime - startTime);
     }
 
-    static final Object IGNORE = new Object();
-    
     @FunctionalInterface
-    static interface F3<T, U, V> {
+    private static interface F3<T, U, V> {
         void apply(T p1, U p2, V p3);
     }
 

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

@@ -85,10 +85,6 @@ protected static final class PositiveVerdict implements Verdict {
     private final DelayPolicy<? super T> backoff;
     private final DelayPolicy<? super T> timeout;
 
-    public static <T> RetryPolicy<T> create() {
-        return new RetryPolicy<>(-1);
-    }
-    
     @SafeVarargs
     public final RetryPolicy<T> retryOn(Class<? extends Throwable>... retryOnThrowables) {
         return retryOn(Arrays.asList(retryOnThrowables));

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

@@ -17,14 +17,18 @@
 
 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 {
 
@@ -68,25 +72,17 @@ static boolean cancelPromise(CompletionStage<?> promise, boolean mayInterruptIfR
             Future<?> future = (Future<?>) promise;
             return future.cancel(mayInterruptIfRunning);
         } else {
-            Method m = completeExceptionallyMethodOf(promise);
-            if (null != m) {
-                try {
-                    return (Boolean) m.invoke(promise, new CancellationException());
-                } catch (final ReflectiveOperationException ex) {
-                    return false;
-                }
-            } else {
-                return false;
-            }
-        }
-    }
-
-    private static Method completeExceptionallyMethodOf(CompletionStage<?> promise) {
-        try {
-            Class<?> clazz = promise.getClass();
-            return clazz.getMethod("completeExceptionally", Throwable.class);
-        } catch (ReflectiveOperationException | SecurityException ex) {
-            return null;
+            Stream<Function<Class<?>, ExceptionalCancellation>> options = Stream.of(
+               SharedFunctions::cancelInterruptibleMethodOf,     
+               SharedFunctions::cancelMethodOf,
+               SharedFunctions::completeExceptionallyMethodOf
+            );
+            
+            return options.map(f -> tryCancellation(f, promise, mayInterruptIfRunning))  
+                          .filter(Optional::isPresent)
+                          .map(Optional::get)
+                          .findFirst()
+                          .orElse(Boolean.FALSE);
         }
     }
 
@@ -113,6 +109,63 @@ static <T, E extends Throwable> T sneakyThrow(Throwable e) throws E {
         throw (E) e;
     }
 
+    
+    private static Optional<Boolean> tryCancellation(Function<Class<?>, ExceptionalCancellation> option, 
+                                                     CompletionStage<?> promise,
+                                                     boolean mayInterruptIfRunning) {
+        
+
+        return Optional.ofNullable( option.apply(promise.getClass()) )
+                       .map(SharedFunctions::uncheckedReflectionException) // TODO: False for reflective operation or runtime exception?
+                       .map(bf -> bf.apply(promise, mayInterruptIfRunning ? Boolean.TRUE : Boolean.FALSE))
+        ;      
+    }
+
+    private static ExceptionalCancellation completeExceptionallyMethodOf(Class<?> clazz) {
+        try {
+            Method m = clazz.getMethod("completeExceptionally", Throwable.class);
+            return (p, b) -> (Boolean)m.invoke(p, new CancellationException());
+        } catch (ReflectiveOperationException | SecurityException ex) {
+            return null;
+        }
+    }
+    
+    private static ExceptionalCancellation cancelInterruptibleMethodOf(Class<?> clazz) {
+        try {
+            Method m = clazz.getMethod("cancel", boolean.class);
+            return (p, b) -> (Boolean)m.invoke(p, b);
+        } catch (ReflectiveOperationException | SecurityException ex) {
+            return null;
+        }
+    }
+    
+    private static ExceptionalCancellation cancelMethodOf(Class<?> clazz) {
+        try {
+            Method m = clazz.getMethod("cancel");
+            return (p, b) -> (Boolean)m.invoke(p);
+        } catch (ReflectiveOperationException | SecurityException ex) {
+            return null;
+        }
+    }
+    
+    private static <T, U> Cancellation uncheckedReflectionException(ExceptionalCancellation original ) {
+        return (a, b) -> { 
+            try {
+                return original.apply(a, b);
+            } catch (ReflectiveOperationException ex) {
+                throw new RuntimeException(ex);
+            }
+        };
+    }
+    
     private static final BiFunction<Object, Object, Object> SELECT_FIRST  = (u, v) -> u;
     private static final BiFunction<Object, Object, Object> SELECT_SECOND = (u, v) -> v;
+    
+    @FunctionalInterface
+    private static interface Cancellation extends BiFunction<CompletionStage<?>, Boolean, Boolean> { }
+    
+    @FunctionalInterface
+    public interface ExceptionalCancellation {
+        Boolean apply(CompletionStage<?> p, Boolean b) throws ReflectiveOperationException;
+    }
 }

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

@@ -53,7 +53,7 @@ public static void main(final String[] argv) throws InterruptedException, Execut
 
         Promise<BigInteger> tryTyping = Promises.retry(
             J8Examples::tryCalc, executorService, 
-            RetryPolicy.<Number>create().withResultValidator(v -> v.intValue() > 0).withMaxRetries(2)
+            new RetryPolicy<Number>().withResultValidator(v -> v.intValue() > 0).withMaxRetries(2)
         );
         System.out.println( tryTyping.get() );