From 50fbd806115aa08d9c06f0399f83deb47ba2abb2 Mon Sep 17 00:00:00 2001
From: Jerry Lee <oldratlee@gmail.com>
Date: Wed, 12 Jun 2024 12:56:37 +0800
Subject: [PATCH] WIP

---
 .../cffu/CompletableFutureUtils.java          | 305 ++++++++++++++++++
 1 file changed, 305 insertions(+)

diff --git a/cffu-core/src/main/java/io/foldright/cffu/CompletableFutureUtils.java b/cffu-core/src/main/java/io/foldright/cffu/CompletableFutureUtils.java
index 655239e0..268a14d8 100644
--- a/cffu-core/src/main/java/io/foldright/cffu/CompletableFutureUtils.java
+++ b/cffu-core/src/main/java/io/foldright/cffu/CompletableFutureUtils.java
@@ -27,6 +27,197 @@
  * @author Jerry Lee (oldratlee at gmail dot com)
  */
 public final class CompletableFutureUtils {
+    ////////////////////////////////////////////////////////////////////////////////
+    //# multi-run methods
+    ////////////////////////////////////////////////////////////////////////////////
+
+    /**
+     * Returns a new CompletableFuture that is asynchronously completed
+     * by tasks running in the CompletableFuture's default asynchronous execution facility
+     * after runs the given actions.
+     *
+     * @param actions the actions to run before completing the returned CompletableFuture
+     * @return the new CompletableFuture
+     * @see #allOf(CompletionStage[])
+     * @see CompletableFuture#runAsync(Runnable)
+     */
+    public static CompletableFuture<Void> mRunAsync(Runnable... actions) {
+        return mRunAsync(AsyncPoolHolder.ASYNC_POOL, actions);
+    }
+
+    /**
+     * Returns a new CompletableFuture that is asynchronously completed
+     * by tasks running in the supplied Executor after runs the given actions.
+     *
+     * @param executor the executor to use for asynchronous execution
+     * @param actions  the actions to run before completing the returned CompletableFuture
+     * @return the new CompletableFuture
+     * @see #allOf(CompletionStage[])
+     * @see CompletableFuture#runAsync(Runnable, Executor)
+     */
+    public static CompletableFuture<Void> mRunAsync(Executor executor, Runnable... actions) {
+        CompletableFuture<?>[] cfs = new CompletableFuture[actions.length];
+        for (int i = 0; i < actions.length; i++) {
+            cfs[i] = CompletableFuture.runAsync(actions[i], executor);
+        }
+        return CompletableFuture.allOf(cfs);
+    }
+
+    /**
+     * Returns a new CompletableFuture that is asynchronously completed
+     * by tasks running in the CompletableFuture's default asynchronous execution facility
+     * after runs the given actions.
+     * <p>
+     * This method is the same as {@link #mRunAsync(Runnable...)} except for the fast-fail behavior.
+     *
+     * @param actions the actions to run before completing the returned CompletableFuture
+     * @return the new CompletableFuture
+     * @see #allOfFastFail(CompletionStage[])
+     * @see CompletableFuture#runAsync(Runnable)
+     */
+    public static CompletableFuture<Void> mRunFastFailAsync(Runnable... actions) {
+        return mRunFastFailAsync(AsyncPoolHolder.ASYNC_POOL, actions);
+    }
+
+    /**
+     * Returns a new CompletableFuture that is asynchronously completed
+     * by tasks running in the supplied Executor after runs the given actions.
+     * <p>
+     * This method is the same as {@link #mRunAsync(Executor, Runnable...)} except for the fast-fail behavior.
+     *
+     * @param executor the executor to use for asynchronous execution
+     * @param actions  the actions to run before completing the returned CompletableFuture
+     * @return the new CompletableFuture
+     * @see #allOfFastFail(CompletionStage[])
+     * @see CompletableFuture#runAsync(Runnable, Executor)
+     */
+    public static CompletableFuture<Void> mRunFastFailAsync(Executor executor, Runnable... actions) {
+        CompletableFuture<?>[] cfs = new CompletableFuture[actions.length];
+        for (int i = 0; i < actions.length; i++) {
+            cfs[i] = CompletableFuture.runAsync(actions[i], executor);
+        }
+        return allOfFastFail(cfs);
+    }
+
+    /**
+     * Returns a new CompletableFuture that is asynchronously completed
+     * by tasks running in the CompletableFuture's default asynchronous execution facility
+     * with the values obtained by calling the given Suppliers.
+     *
+     * @param suppliers functions returning the value to be used to complete the returned CompletableFuture
+     * @param <T>       the function's return type
+     * @return the new CompletableFuture
+     * @see #allResultsOf(CompletionStage[])
+     * @see CompletableFuture#supplyAsync(Supplier)
+     */
+    public static <T> CompletableFuture<List<T>> mSupplyAsync(Supplier<? extends T>... suppliers) {
+        return mSupplyAsync(AsyncPoolHolder.ASYNC_POOL, suppliers);
+    }
+
+    /**
+     * Returns a new CompletableFuture that is asynchronously completed
+     * by tasks running in the supplied Executor with the values obtained by calling the given Suppliers.
+     *
+     * @param executor  the executor to use for asynchronous execution
+     * @param suppliers functions returning the value to be used to complete the returned CompletableFuture
+     * @param <T>       the function's return type
+     * @return the new CompletableFuture
+     * @see #allResultsOf(CompletionStage[])
+     * @see CompletableFuture#supplyAsync(Supplier, Executor)
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> CompletableFuture<List<T>> mSupplyAsync(Executor executor, Supplier<? extends T>... suppliers) {
+        CompletableFuture<T>[] cfs = new CompletableFuture[suppliers.length];
+        for (int i = 0; i < suppliers.length; i++) {
+            cfs[i] = CompletableFuture.supplyAsync((Supplier<T>) suppliers[i], executor);
+        }
+        return allResultsOf(cfs);
+    }
+
+    /**
+     * Returns a new CompletableFuture that is asynchronously completed
+     * by tasks running in the CompletableFuture's default asynchronous execution facility
+     * with the values obtained by calling the given Suppliers.
+     * <p>
+     * This method is the same as {@link #mSupplyAsync(Supplier[])} except for the fast-fail behavior.
+     *
+     * @param suppliers functions returning the value to be used to complete the returned CompletableFuture
+     * @param <T>       the function's return type
+     * @return the new CompletableFuture
+     * @see #allResultsOfFastFail(CompletionStage[])
+     * @see CompletableFuture#supplyAsync(Supplier)
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> CompletableFuture<List<T>> mSupplyFastFailAsync(Supplier<? extends T>... suppliers) {
+        return mSupplyFastFailAsync(AsyncPoolHolder.ASYNC_POOL, suppliers);
+    }
+
+    /**
+     * Returns a new CompletableFuture that is asynchronously completed
+     * by tasks running in the supplied Executor with the values obtained by calling the given Suppliers.
+     * <p>
+     * This method is the same as {@link #mSupplyAsync(Executor, Supplier[])} except for the fast-fail behavior.
+     *
+     * @param executor  the executor to use for asynchronous execution
+     * @param suppliers functions returning the value to be used to complete the returned CompletableFuture
+     * @param <T>       the function's return type
+     * @return the new CompletableFuture
+     * @see #allResultsOfFastFail(CompletionStage[])
+     * @see CompletableFuture#supplyAsync(Supplier, Executor)
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> CompletableFuture<List<T>> mSupplyFastFailAsync(Executor executor, Supplier<? extends T>... suppliers) {
+        CompletableFuture<T>[] cfs = new CompletableFuture[suppliers.length];
+        for (int i = 0; i < suppliers.length; i++) {
+            cfs[i] = CompletableFuture.supplyAsync((Supplier<T>) suppliers[i], executor);
+        }
+        return allResultsOfFastFail(cfs);
+    }
+
+    /**
+     * Returns a new CompletableFuture with the most results in the <strong>same order</strong> of tasks
+     * running in the CompletableFuture's default asynchronous execution facility
+     * in the given time({@code timeout}), aka as many results as possible in the given time.
+     * <p>
+     * If the given supplier is successful in time, its result is the completed value;
+     * Otherwise the given valueIfNotSuccess.
+     *
+     * @param suppliers functions returning the value to be used to complete the returned CompletableFuture
+     * @param <T>       the function's return type
+     * @return the new CompletableFuture
+     * @see #mostResultsOfSuccess(long, TimeUnit, Object, CompletionStage[])
+     * @see CompletableFuture#supplyAsync(Supplier)
+     */
+    public static <T> CompletableFuture<List<T>> mSupplyMostSuccess(
+            long timeout, TimeUnit unit, @Nullable T valueIfNotSuccess, Supplier<? extends T>... suppliers) {
+        return mSupplyMostSuccess(AsyncPoolHolder.ASYNC_POOL, timeout, unit, valueIfNotSuccess, suppliers);
+    }
+
+    /**
+     * Returns a new CompletableFuture with the most results in the <strong>same order</strong> of tasks
+     * running in the supplied Executor in the given time({@code timeout}),
+     * aka as many results as possible in the given time.
+     * <p>
+     * If the given supplier is successful in time, its result is the completed value;
+     * Otherwise the given valueIfNotSuccess.
+     *
+     * @param suppliers functions returning the value to be used to complete the returned CompletableFuture
+     * @param <T>       the function's return type
+     * @return the new CompletableFuture
+     * @see #mostResultsOfSuccess(Executor, long, TimeUnit, Object, CompletionStage[])
+     * @see CompletableFuture#supplyAsync(Supplier, Executor)
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> CompletableFuture<List<T>> mSupplyMostSuccess(
+            Executor executor, long timeout, TimeUnit unit,
+            @Nullable T valueIfNotSuccess, Supplier<? extends T>... suppliers) {
+        CompletableFuture<T>[] cfs = new CompletableFuture[suppliers.length];
+        for (int i = 0; i < suppliers.length; i++) {
+            cfs[i] = CompletableFuture.supplyAsync((Supplier<T>) suppliers[i], executor);
+        }
+        return mostResultsOfSuccess(executor, timeout, unit, valueIfNotSuccess, cfs);
+    }
+
     ////////////////////////////////////////////////////////////////////////////////
     //# allOf*/mostResultsOfSuccess methods
     ////////////////////////////////////////////////////////////////////////////////
@@ -836,6 +1027,120 @@ private static <T> CompletableFuture<T> mostTupleOfSuccess0(
     //    - thenCombineFastFail*(BiFunction):    (T1, T2) -> U
     ////////////////////////////////////////////////////////////////////////////////
 
+    public static CompletableFuture<Void> thenMRunAsync(CompletionStage<?> cf, Runnable... actions) {
+        return thenMRunAsync(cf, AsyncPoolHolder.ASYNC_POOL, actions);
+    }
+
+    public static CompletableFuture<Void> thenMRunAsync(CompletionStage<?> cf, Executor executor, Runnable... actions) {
+        return toNonMinCf(cf).thenCompose(unused -> mRunAsync(executor, actions));
+    }
+
+    public static CompletableFuture<Void> thenMRunFastFailAsync(CompletionStage<?> cf, Runnable... actions) {
+        return thenMRunFastFailAsync(cf, AsyncPoolHolder.ASYNC_POOL, actions);
+    }
+
+    public static CompletableFuture<Void> thenMRunFastFailAsync(CompletionStage<?> cf, Executor executor, Runnable... actions) {
+        return toNonMinCf(cf).thenCompose(unused -> mRunFastFailAsync(executor, actions));
+    }
+
+    public static <T> CompletableFuture<Void> thenMAcceptAsync(
+            CompletionStage<? extends T> cf, Consumer<? super T>... actions) {
+        return thenMAcceptAsync(cf, AsyncPoolHolder.ASYNC_POOL, actions);
+    }
+
+    public static <T> CompletableFuture<Void> thenMAcceptAsync(
+            CompletionStage<? extends T> cf, Executor executor, Consumer<? super T>... actions) {
+        return toNonMinCf(cf).thenCompose(v -> {
+            Runnable[] rs = new Runnable[actions.length];
+            for (int i = 0; i < actions.length; i++) {
+                final int idx = i;
+                rs[idx] = () -> actions[idx].accept(v);
+            }
+            return mRunAsync(executor, rs);
+        });
+    }
+
+    public static <T> CompletableFuture<Void> thenMAcceptFastFailAsync(
+            CompletionStage<? extends T> cf, Consumer<? super T>... actions) {
+        return thenMAcceptFastFailAsync(cf, AsyncPoolHolder.ASYNC_POOL, actions);
+    }
+
+    public static <T> CompletableFuture<Void> thenMAcceptFastFailAsync(
+            CompletionStage<? extends T> cf, Executor executor, Consumer<? super T>... actions) {
+        return toNonMinCf(cf).thenCompose(v -> {
+            Runnable[] rs = new Runnable[actions.length];
+            for (int i = 0; i < actions.length; i++) {
+                final int idx = i;
+                rs[idx] = () -> actions[idx].accept(v);
+            }
+            return mRunFastFailAsync(executor, rs);
+        });
+    }
+
+    public <T, U> CompletableFuture<List<U>> thenMApplyAsync(
+            CompletionStage<? extends T> cf, Function<? super T, ? extends U>... fns) {
+        return thenMApplyAsync(cf, AsyncPoolHolder.ASYNC_POOL, fns);
+    }
+
+    public <T, U> CompletableFuture<List<U>> thenMApplyAsync(
+            CompletionStage<? extends T> cf, Executor executor, Function<? super T, ? extends U>... fns) {
+        return toNonMinCf(cf).thenCompose(v -> {
+            @SuppressWarnings("unchecked")
+            Supplier<U>[] suppliers = new Supplier[fns.length];
+            for (int i = 0; i < fns.length; i++) {
+                final int idx = i;
+                suppliers[idx] = () -> fns[idx].apply(v);
+            }
+            return mSupplyAsync(executor, suppliers);
+        });
+    }
+
+    public <T, U> CompletableFuture<List<U>> thenMApplyFastFailAsync(
+            CompletionStage<? extends T> cf, Function<? super T, ? extends U>... fns) {
+        return thenMApplyFastFailAsync(cf, AsyncPoolHolder.ASYNC_POOL, fns);
+    }
+
+    public <T, U> CompletableFuture<List<U>> thenMApplyFastFailAsync(
+            CompletionStage<? extends T> cf, Executor executor, Function<? super T, ? extends U>... fns) {
+        return toNonMinCf(cf).thenCompose(v -> {
+            @SuppressWarnings("unchecked")
+            Supplier<U>[] suppliers = new Supplier[fns.length];
+            for (int i = 0; i < fns.length; i++) {
+                final int idx = i;
+                suppliers[idx] = () -> fns[idx].apply(v);
+            }
+            return mSupplyFastFailAsync(executor, suppliers);
+        });
+    }
+
+    public <T, U> CompletableFuture<List<U>> thenMApplyMostSuccessAsync(
+            CompletionStage<? extends T> cf, long timeout, TimeUnit unit,
+            @Nullable U valueIfNotSuccess, Function<? super T, ? extends U>... fns) {
+        return thenMApplyMostSuccessAsync(cf, AsyncPoolHolder.ASYNC_POOL, timeout, unit, valueIfNotSuccess, fns);
+    }
+
+    public <T, U> CompletableFuture<List<U>> thenMApplyMostSuccessAsync(
+            CompletionStage<? extends T> cf, Executor executor, long timeout, TimeUnit unit,
+            @Nullable U valueIfNotSuccess, Function<? super T, ? extends U>... fns) {
+        return toNonMinCf(cf).thenCompose(v -> {
+            @SuppressWarnings("unchecked")
+            Supplier<U>[] suppliers = new Supplier[fns.length];
+            for (int i = 0; i < fns.length; i++) {
+                final int idx = i;
+                suppliers[idx] = () -> fns[idx].apply(v);
+            }
+            return mSupplyMostSuccess(executor, timeout, unit, valueIfNotSuccess, suppliers);
+        });
+    }
+
+    ////////////////////////////////////////////////////////////////////////////////
+    //# `then both(binary input)` methods with fast-fail support:
+    //
+    //    - runAfterBothFastFail*(Runnable):     Void, Void -> Void
+    //    - thenAcceptBothFastFail*(BiConsumer): (T1, T2) -> Void
+    //    - thenCombineFastFail*(BiFunction):    (T1, T2) -> U
+    ////////////////////////////////////////////////////////////////////////////////
+
     /**
      * Returns a new CompletableFuture that, when two given stages both complete normally, executes the given action.
      * if any of the given stage complete exceptionally, then the returned CompletableFuture also does so