Handle exceptions in AbstractCompletableTask as in CompletableFuture (wrapping & propagation); Optimizing Promises to use only Promise.dependent() instead of ad-hoc decorators.

Author: vsilaev

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

@@ -296,15 +296,27 @@ public Promise<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> act
         addCallbacks(
             nextStage, 
             result -> {
-                action.accept(result, null);
-                return result;
+                try {
+                    action.accept(result, null);
+                    return result;
+                } catch (Throwable e) {
+                    // CompletableFuture wraps exception here
+                    // Copying this behavior                    
+                    return forwardException(e);
+                }
             }, 
+            // exceptions are handled in regular way
             failure -> {
                 try {
                     action.accept(null, failure);
                     return forwardException(failure);
                 } catch (Throwable e) {
-                    return forwardException(e);
+                    // CompletableFuture does not override exception here
+                    // unlike as in handle[Async](BiFunction)
+                    // Preserve this behavior, but let us add at least 
+                    // suppressed exception
+                    failure.addSuppressed(e);
+                    return forwardException(failure);
                 }
             }, 
             executor
@@ -317,9 +329,27 @@ public <U> Promise<U> handleAsync(BiFunction<? super T, Throwable, ? extends U>
         AbstractCompletableTask<U> nextStage = internalCreateCompletionStage(executor);
         addCallbacks(
             nextStage, 
-            result -> fn.apply(result, null),
-            // exceptions are treated as success
-            error -> fn.apply(null, error), executor
+            result -> {
+                try {
+                    return fn.apply(result, null);
+                } catch (Throwable e) {
+                    // CompletableFuture wraps exception here
+                    // Copying this behavior
+                    return forwardException(e);
+                }
+            },
+            // exceptions are handled in regular way
+            failure -> {
+                try {
+                    return fn.apply(null, failure);
+                } catch (Throwable e) {
+                    // CompletableFuture handle[Async](BiFunction)
+                    // allows to overwrite exception for resulting stage.
+                    // Copying this behavior
+                    return forwardException(e);
+                }
+            },
+            executor
         );
         return nextStage;
     }
@@ -371,7 +401,7 @@ private <R, U> Promise<U> doApplyToEitherAsync(CompletionStage<? extends R> firs
             if (failure == null) {
                 nextStage.onSuccess(result);
             } else {
-                nextStage.onError(Promises.wrapException(failure));
+                nextStage.onError(forwardException(failure));
             }
         };
         // only the first result is accepted by completion stage,

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

@@ -27,9 +27,7 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiConsumer;
 import java.util.function.Consumer;
-import java.util.function.Function;
 import java.util.function.Supplier;
 
 /**
@@ -54,9 +52,9 @@ private Promises() {}
      *   a successfully resolved {@link Promise} with a value provided
      */
     public static <T> Promise<T> success(T value) {
-        CompletablePromise<T> result = new CompletablePromise<>();
-        result.onSuccess(value);
-        return result;
+        return new CompletablePromise<>(
+            CompletableFuture.completedFuture(value)
+        );
     }
 
     /**
@@ -69,9 +67,9 @@ public static <T> Promise<T> success(T value) {
      *   a faulty resolved {@link Promise} with an exception provided
      */    
     public static <T> Promise<T> failure(Throwable exception) {
-        CompletablePromise<T> result = new CompletablePromise<>();
-        result.onFailure(exception);
-        return result;
+        CompletableFuture<T> delegate = new CompletableFuture<>();
+        delegate.completeExceptionally(exception);
+        return new CompletablePromise<>(delegate);
     }
 
     /**
@@ -92,35 +90,7 @@ public static <T> Promise<T> from(CompletionStage<T> stage) {
             return new CompletablePromise<>((CompletableFuture<T>)stage);
         }
 
-        CompletablePromise<T> result = createLinkedPromise(stage);
-        stage.whenComplete(handler(result::onSuccess, result::onFailure));
-        return result;
-    }
-
-    static <T, R> Promise<R> from(CompletionStage<? extends T> stage, 
-                                  Function<? super T, ? extends R> resultConverter,
-                                  Function<? super Throwable, ? extends Throwable> errorConverter) {
-        
-        CompletablePromise<R> result = createLinkedPromise(stage);
-        stage.whenComplete(handler(
-            acceptConverted(result::onSuccess, resultConverter),
-            acceptConverted(result::onFailure, errorConverter)
-        ));
-        return result;
-    }
-
-    private static <T, R> CompletablePromise<R> createLinkedPromise(CompletionStage<? extends T> stage) {
-        return new CompletablePromise<R>() {
-            @Override
-            public boolean cancel(boolean mayInterruptIfRunning) {
-                if (super.cancel(mayInterruptIfRunning)) {
-                    cancelPromise(stage, mayInterruptIfRunning);
-                    return true;
-                } else {
-                    return false;
-                }
-            }
-        };
+        return COMPLETED_DEPENDENT_PROMISE.thenCombine(stage, (u, v) -> v, PromiseOrigin.PARAM_ONLY).raw();
     }
 
     /**
@@ -425,7 +395,10 @@ public static <T> Promise<List<T>> atLeast(int minResultsCount, int maxErrorsCou
         } else if (minResultsCount == 0) {
             return success(Collections.emptyList());
         } else if (size == 1) {
-            return from(promises.get(0), Collections::singletonList, Function.<Throwable> identity());
+            return from(promises.get(0))
+                   .dependent()
+                   .thenApply((T r) -> Collections.singletonList(r), true)
+                   .raw();
         } else {
             return new AggregatingPromise<>(minResultsCount, maxErrorsCount, cancelRemaining, promises);
         }
@@ -541,6 +514,14 @@ private static <T> void pollOnce(Callable<Optional<? extends T>> codeBlock,
             resultPromise.onFailure(ctx.asFailure());
         }
     }
+    
+    public static Throwable unwrapException(Throwable ex) {
+        Throwable nested = ex;
+        while (nested instanceof CompletionException) {
+            nested = nested.getCause();
+        }
+        return null == nested ? ex : nested;
+    }
 
     static CompletionException wrapException(Throwable e) {
         if (e instanceof CompletionException) {
@@ -551,34 +532,23 @@ static CompletionException wrapException(Throwable e) {
     }
 
     private static <T> Promise<T> unwrap(CompletionStage<List<T>> original, boolean unwrapException) {
-        return from(
-            original,
-            c -> c.stream().filter(Objects::nonNull).findFirst().get(),
-            unwrapException ? 
-                ex -> ex instanceof MultitargetException ? ((MultitargetException)ex).getFirstException().get() : ex
-                :
-                Function.identity()
-        );
-    }
-
-    private static <T> BiConsumer<T, ? super Throwable> handler(Consumer<? super T> onResult, Consumer<? super Throwable> onError) {
-        return (r, e) -> {
-            if (null != e) {
-                onError.accept(e);
-            } else {
-                try {
-                    onResult.accept(r);
-                } catch (Exception ex) {
-                    onError.accept(ex);
+        return from(original)
+            .dependent()
+            .handle((r, e) -> {
+                if (null != e) {
+                    if (unwrapException) {
+                        Throwable targetException = unwrapException(e);
+                        if (targetException instanceof MultitargetException) {
+                            throw wrapException( ((MultitargetException)targetException).getFirstException().get() );
+                        } 
+                    }
+                    throw wrapException(e);
+                } else {
+                    return r.stream().filter(Objects::nonNull).findFirst().get();
                 }
-            }
-        };
+            }, true)
+            .raw();
     }
-
-    private static <T, U> Consumer<? super T> acceptConverted(Consumer<? super U> target, Function<? super T, ? extends U> converter) {
-        return t -> target.accept(converter.apply(t));
-    }
-
 
     private static class ObjectRef<T> {
         private final T reference;
@@ -593,4 +563,5 @@ T dereference() {
     }
 
     private static final Object IGNORE = new Object();
+    private static final DependentPromise<Object> COMPLETED_DEPENDENT_PROMISE = success(null).dependent();
 }