Fix the fix for suppressed exceptions in Promises.try* (reverting)

Author: vsilaev

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

@@ -163,14 +163,14 @@ public Promise<T> exceptionallyAsync(Function<Throwable, ? extends T> fn, Execut
         return addCallbacks(
             internalCreateCompletionStage(executor), 
             Function.identity(), 
-            e -> {
+            failure -> {
                 try {
-                    return fn.apply(e);
-                } catch (Throwable t) {
-                    if (t != e) {
-                        t.addSuppressed(e);
+                    return fn.apply(failure);
+                } catch (Throwable e) {
+                    if (e != failure) {
+                        e.addSuppressed(failure);
                     }
-                    return forwardException(t);
+                    return forwardException(e);
                 }
             }, 
             executor

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

@@ -16,6 +16,7 @@
 package net.tascalate.concurrent;
 
 import static net.tascalate.concurrent.SharedFunctions.cancelPromise;
+import static net.tascalate.concurrent.SharedFunctions.wrapCompletionException;
 import static net.tascalate.concurrent.SharedFunctions.iif;
 
 import java.time.Duration;
@@ -162,8 +163,8 @@ public static <T, R extends AsyncCloseable> Promise<T> tryApplyEx(Promise<R> p,
                            // So resource.close() is never cancellable
                            return resource.close().thenCompose(__ -> failure(onAction));
                        } catch (Throwable onClose) {
-                           onClose.addSuppressed(onAction);
-                           return failure(onClose);
+                           onAction.addSuppressed(onClose);
+                           return failure(onAction);
                        }
                    }
 
@@ -193,8 +194,8 @@ public static <T, R extends AutoCloseable> Promise<T> tryCompose(Promise<R> p,
                            resource.close();
                            return failure(onAction);
                        } catch (Exception onClose) {
-                           onClose.addSuppressed(onAction);
-                           return failure(onClose);
+                           onAction.addSuppressed(onClose);
+                           return failure(onAction);
                        }
                    }
 
@@ -204,10 +205,7 @@ public static <T, R extends AutoCloseable> Promise<T> tryCompose(Promise<R> p,
                            resource.close();
                            result.complete(actionResult, actionException);
                        } catch (Throwable onClose) {
-                           if (null != actionException) {
-                               onClose.addSuppressed(actionException);
-                           }
-                           result.failure(onClose);
+                           completeExceptionally(result, actionException, onClose);
                        }
                    });
                    return result.onCancel(() -> cancelPromise(action, true));
@@ -231,8 +229,8 @@ public static <T, R extends AsyncCloseable> Promise<T> tryComposeEx(Promise<R> p
                            // So resource.close() is never cancellable
                            return resource.close().thenCompose(__ -> failure(onAction));
                        } catch (Throwable onClose) {
-                           onClose.addSuppressed(onAction);
-                           return failure(onClose);
+                           onAction.addSuppressed(onClose);
+                           return failure(onAction);
                        }                               
                    }
 
@@ -244,17 +242,11 @@ public static <T, R extends AsyncCloseable> Promise<T> tryComposeEx(Promise<R> p
                                if (null == onClose) {
                                    result.complete(actionResult, actionException);
                                } else {
-                                   if (null != actionException) {
-                                       onClose.addSuppressed(actionException);
-                                   }
-                                   result.failure(onClose);
+                                   completeExceptionally(result, actionException, onClose);
                                }
                            });
                        } catch (Throwable onClose) {
-                           if (null != actionException) {
-                               onClose.addSuppressed(actionException);
-                           }
-                           result.failure(onClose);
+                           completeExceptionally(result, actionException, onClose);
                        }
                    });
                    return result.onCancel(() -> cancelPromise(action, true));
@@ -300,7 +292,7 @@ private static <T, A, R> Promise<R> partitioned1(Iterator<? extends T> values,
             parallelStep1(values, batchSize, spawner, downstream)
             .dependent()
             .thenApply(downstream.finisher().compose(IndexedStep::payload), true)
-            .as(onCloseSource(null != source? source : values))
+            .asʹ(maybeClosingSource(null != source? source : values))
             .unwrap();
     }
 
@@ -314,7 +306,7 @@ private static <T, A, R> Promise<R> partitioned2(Iterator<? extends T> values,
             parallelStep2(values, batchSize, spawner, downstream, downstreamExecutor)
             .dependent()
             .thenApplyAsync(downstream.finisher().compose(IndexedStep::payload), downstreamExecutor, true)
-            .as(onCloseSource(null != source? source : values))
+            .asʹ(maybeClosingSource(null != source? source : values))
             .unwrap();
     }
 
@@ -416,21 +408,17 @@ private static <T, A, R> A accumulate(List<T> vals, boolean initial, A current,
         return initial ? insertion : downstream.combiner().apply(current, insertion);
     }
 
-    private static <T> Function<Promise<T>, Promise<T>> onCloseSource(Object source) {
+    private static <T> Function<DependentPromise<T>, DependentPromise<T>> maybeClosingSource(Object source) {
         if (source instanceof AutoCloseable) {
-            return p -> p.dependent().whenComplete((r, e) -> {
-                try (AutoCloseable o = (AutoCloseable)source) {
-                    
-                } catch (RuntimeException | Error ex) {
-                    if (null != e) {
-                        ex.addSuppressed(e);
-                    }
-                    throw ex;
+            return p -> p.whenComplete((r, e) -> {
+                try {
+                    ((AutoCloseable)source).close();
                 } catch (Exception ex) {
                     if (null != e) {
-                        ex.addSuppressed(e);
+                        e.addSuppressed(ex);
+                    } else {
+                        throw wrapCompletionException(ex);
                     }
-                    throw new CompletionException(ex);
                 }
             }, true);
         } else {
@@ -1205,6 +1193,15 @@ private static <T> Promise<T> insufficientNumberOfArguments(int minResultCount,
         */
     }
 
+    private static <T> void completeExceptionally(CompletableFutureWrapper<T> result, Throwable actionException, Throwable onClose) {
+        if (null != actionException) {
+            actionException.addSuppressed(onClose);
+            result.failure(actionException);
+        } else {
+            result.failure(onClose);
+        }        
+    }
+    
     private static <K, T> List<? extends CompletionStage<? extends T>> 
         collectKeyedResults(Map<K, T> result, 
                             Map<? extends K, ? extends CompletionStage<? extends T>> promises) {