From 31abe5ad0873666d96a73323bff293e332c07b59 Mon Sep 17 00:00:00 2001 From: Jerry Lee Date: Tue, 12 Dec 2023 01:31:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20implement=20bothFastFail/eitherSuccess?= =?UTF-8?q?=20methods=20=E2=98=98=EF=B8=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/java/io/foldright/cffu/Cffu.java | 4 +- .../cffu/CompletableFutureUtils.java | 385 +++++++++++++++++- 2 files changed, 382 insertions(+), 7 deletions(-) diff --git a/cffu-core/src/main/java/io/foldright/cffu/Cffu.java b/cffu-core/src/main/java/io/foldright/cffu/Cffu.java index b5cb611e..e0038dac 100644 --- a/cffu-core/src/main/java/io/foldright/cffu/Cffu.java +++ b/cffu-core/src/main/java/io/foldright/cffu/Cffu.java @@ -719,8 +719,8 @@ public Cffu> cffuCombineFastFail( //# `then either(binary input)` methods of CompletionStage: // // - runAfterEither*(Runnable): Void, Void -> Void - // - acceptEither*(BiConsumer): (T1, T2) -> Void - // - applyToEither*(BiFunction): (T1, T2) -> U + // - acceptEither*(Consumer): (T, T) -> Void + // - applyToEither*(Function): (T, T) -> U //////////////////////////////////////////////////////////////////////////////// /** 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 f0087d3b..87d6ec57 100644 --- a/cffu-core/src/main/java/io/foldright/cffu/CompletableFutureUtils.java +++ b/cffu-core/src/main/java/io/foldright/cffu/CompletableFutureUtils.java @@ -14,8 +14,7 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.*; -import java.util.function.Function; -import java.util.function.Supplier; +import java.util.function.*; import static java.util.Objects.requireNonNull; @@ -141,7 +140,7 @@ public static CompletableFuture> allOfFastFailWithResult(Completable return (CompletableFuture) CompletableFuture.anyOf(failedOrBeIncomplete); } - private static void requireCfsAndEleNonNull(CompletableFuture... cfs) { + private static void requireCfsAndEleNonNull(CompletionStage... cfs) { requireNonNull(cfs, "cfs is null"); for (int i = 0; i < cfs.length; i++) { requireNonNull(cfs[i], "cf" + (i + 1) + " is null"); @@ -189,7 +188,7 @@ private static void fill(CompletableFuture[] cfs, @Contract(pure = true) @SafeVarargs @SuppressWarnings("unchecked") - public static CompletableFuture anyOfWithType(CompletableFuture... cfs) { + public static CompletableFuture anyOfWithType(CompletableFuture... cfs) { return (CompletableFuture) CompletableFuture.anyOf(cfs); } @@ -247,10 +246,257 @@ public static CompletableFuture anyOfSuccess(CompletableFuture... cfs @Contract(pure = true) @SafeVarargs @SuppressWarnings("unchecked") - public static CompletableFuture anyOfSuccessWithType(CompletableFuture... cfs) { + public static CompletableFuture anyOfSuccessWithType(CompletableFuture... cfs) { return (CompletableFuture) anyOfSuccess(cfs); } + //////////////////////////////////////////////////////////////////////////////// + //# `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 *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#runAfterBoth(CompletionStage, Runnable)} + * except for the fast-fail behavior. + * + * @param action the action to perform before completing the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#runAfterBoth(CompletionStage, Runnable) + */ + public static CompletableFuture runAfterBothFastFail( + CompletionStage cf1, CompletionStage cf2, Runnable action) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + + return allOfFastFail(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenRun(action); + } + + /** + * Returns a new CompletableFuture that, when two given stages both complete normally, + * executes the given action using CompletableFuture's default asynchronous execution facility. + * if any of the given stage complete exceptionally, then the returned CompletableFuture + * also does so *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#runAfterBothAsync(CompletionStage, Runnable)} + * except for the fast-fail behavior. + * + * @param action the action to perform before completing the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#runAfterBothAsync(CompletionStage, Runnable) + */ + public static CompletableFuture runAfterBothFastFailAsync( + CompletionStage cf1, CompletionStage cf2, Runnable action) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + + return allOfFastFail(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenRunAsync(action); + } + + /** + * Returns a new CompletableFuture that, when two given stages both complete normally, + * executes the given action using the supplied executor. + * if any of the given stage complete exceptionally, then the returned CompletableFuture + * also does so *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#runAfterBothAsync(CompletionStage, Runnable, Executor)} + * except for the fast-fail behavior. + * + * @param action the action to perform before completing the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#runAfterBothAsync(CompletionStage, Runnable, Executor) + */ + public static CompletableFuture runAfterBothFastFailAsync( + CompletionStage cf1, CompletionStage cf2, Runnable action, Executor executor) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + requireNonNull(executor, "executor is null"); + + return allOfFastFail(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenRunAsync(action, executor); + } + + /** + * Returns a new CompletableFuture that, when tow given stage both complete normally, + * is executed with the two results as arguments to the supplied action. + * if any of the given stage complete exceptionally, then the returned CompletableFuture + * also does so *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#thenAcceptBoth(CompletionStage, BiConsumer)} + * except for the fast-fail behavior. + * + * @param action the action to perform before completing the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#thenAcceptBoth(CompletionStage, BiConsumer) + */ + @SuppressWarnings("unchecked") + public static CompletableFuture thenAcceptBothFastFail( + CompletionStage cf1, CompletionStage cf2, BiConsumer action) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + + final Object[] result = new Object[2]; + return allOfFastFail( + cf1.toCompletableFuture().thenAccept(t1 -> result[0] = t1), + cf2.toCompletableFuture().thenAccept(t2 -> result[1] = t2) + ).thenAccept(unused -> action.accept((T) result[0], (U) result[1])); + } + + /** + * Returns a new CompletableFuture that, when tow given stage both complete normally, + * is executed using CompletableFuture's default asynchronous execution facility, + * with the two results as arguments to the supplied action. + * if any of the given stage complete exceptionally, then the returned CompletableFuture + * also does so *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#thenAcceptBothAsync(CompletionStage, BiConsumer)} + * except for the fast-fail behavior. + * + * @param action the action to perform before completing the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#thenAcceptBothAsync(CompletionStage, BiConsumer) + */ + @SuppressWarnings("unchecked") + public static CompletableFuture thenAcceptBothFastFailAsync( + CompletionStage cf1, CompletionStage cf2, BiConsumer action) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + + final Object[] result = new Object[2]; + return allOfFastFail( + cf1.toCompletableFuture().thenAccept(t1 -> result[0] = t1), + cf2.toCompletableFuture().thenAccept(t2 -> result[1] = t2) + ).thenAcceptAsync(unused -> action.accept((T) result[0], (U) result[1])); + } + + /** + * Returns a new CompletableFuture that, when tow given stage both complete normally, + * is executed using the supplied executor, + * with the two results as arguments to the supplied action. + * if any of the given stage complete exceptionally, then the returned CompletableFuture + * also does so *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#thenAcceptBothAsync(CompletionStage, BiConsumer, Executor)} + * except for the fast-fail behavior. + * + * @param action the action to perform before completing the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#thenAcceptBothAsync(CompletionStage, BiConsumer, Executor) + */ + @SuppressWarnings("unchecked") + public static CompletableFuture thenAcceptBothFastFailAsync( + CompletionStage cf1, CompletionStage cf2, + BiConsumer action, Executor executor) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + requireNonNull(executor, "executor is null"); + + final Object[] result = new Object[2]; + return allOfFastFail( + cf1.toCompletableFuture().thenAccept(t1 -> result[0] = t1), + cf2.toCompletableFuture().thenAccept(t2 -> result[1] = t2) + ).thenAcceptAsync(unused -> action.accept((T) result[0], (U) result[1]), executor); + } + + /** + * Returns a new CompletableFuture that, when tow given stage both complete normally, + * is executed with the two results as arguments to the supplied function. + * if any of the given stage complete exceptionally, then the returned CompletableFuture + * also does so *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#thenCombine(CompletionStage, BiFunction)} + * except for the fast-fail behavior. + * + * @param fn the function to use to compute the value of the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#thenCombine(CompletionStage, BiFunction) + */ + @SuppressWarnings("unchecked") + public static CompletableFuture thenCombineFastFail( + CompletionStage cf1, CompletionStage cf2, BiFunction fn + ) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(fn, "fn is null"); + final Object[] result = new Object[2]; + return allOfFastFail( + cf1.toCompletableFuture().thenAccept(t1 -> result[0] = t1), + cf2.toCompletableFuture().thenAccept(t2 -> result[1] = t2) + ).thenApply(unused -> fn.apply((T) result[0], (U) result[1])); + } + + /** + * Returns a new CompletableFuture that, when tow given stage both complete normally, + * is executed using CompletableFuture's default asynchronous execution facility, + * with the two results as arguments to the supplied function. + * if any of the given stage complete exceptionally, then the returned CompletableFuture + * also does so *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#thenCombineAsync(CompletionStage, BiFunction)} + * except for the fast-fail behavior. + * + * @param fn the function to use to compute the value of the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#thenCombineAsync(CompletionStage, BiFunction) + */ + @SuppressWarnings("unchecked") + public static CompletableFuture thenCombineFastFailAsync( + CompletionStage cf1, CompletionStage cf2, BiFunction fn + ) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(fn, "fn is null"); + + final Object[] result = new Object[2]; + return allOfFastFail( + cf1.toCompletableFuture().thenAccept(t1 -> result[0] = t1), + cf2.toCompletableFuture().thenAccept(t2 -> result[1] = t2) + ).thenApplyAsync(unused -> fn.apply((T) result[0], (U) result[1])); + } + + /** + * Returns a new CompletableFuture that, when tow given stage both complete normally, + * is executed using the supplied executor, + * with the two results as arguments to the supplied function. + * if any of the given stage complete exceptionally, then the returned CompletableFuture + * also does so *without* waiting other incomplete given CompletableFutures, + * with a CompletionException holding this exception as its cause. + *

+ * this method is same as {@link CompletableFuture#thenCombineAsync(CompletionStage, BiFunction, Executor)} + * except for the fast-fail behavior. + * + * @param fn the function to use to compute the value of the returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#thenCombineAsync(CompletionStage, BiFunction, Executor) + */ + @SuppressWarnings("unchecked") + public static CompletableFuture thenCombineFastFailAsync( + CompletionStage cf1, CompletionStage cf2, + BiFunction fn, Executor executor + ) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(fn, "fn is null"); + requireNonNull(executor, "executor is null"); + + final Object[] result = new Object[2]; + return allOfFastFail( + cf1.toCompletableFuture().thenAccept(t1 -> result[0] = t1), + cf2.toCompletableFuture().thenAccept(t2 -> result[1] = t2) + ).thenApplyAsync(unused -> fn.apply((T) result[0], (U) result[1]), executor); + } + //////////////////////////////////////////////////////////////////////////////// //# combine/combineFastFail methods //////////////////////////////////////////////////////////////////////////////// @@ -475,6 +721,135 @@ public static CompletableFuture> ); } + //////////////////////////////////////////////////////////////////////////////// + //# `then either(binary input)` methods with either(any)-success support: + // + // - runAfterEitherSuccess*(Runnable): Void, Void -> Void + // - acceptEitherSuccess*(Consumer): (T, T) -> Void + // - applyToEitherSuccess*(Function): (T, T) -> U + //////////////////////////////////////////////////////////////////////////////// + + /** + * Returns a new CompletableFuture that, when either given stage success, executes the given action. + * Otherwise, all two given CompletableFutures complete exceptionally, + * the returned CompletableFuture also does so, with a CompletionException holding + * an exception from any of the given CompletableFutures as its cause. + *

+ * this method is same as {@link CompletableFuture#runAfterEither(CompletionStage, Runnable)} + * except for the either-success behavior. + * + * @param action the action to perform before completing the + * returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#runAfterEither(CompletionStage, Runnable) + */ + public static CompletableFuture runAfterEitherSuccess( + CompletionStage cf1, CompletionStage cf2, Runnable action) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + + return anyOfSuccess(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenRun(action); + } + + /** + * Returns a new CompletableFuture that, when either given stage success, executes the given action + * using CompletableFuture's default asynchronous execution facility. + * Otherwise, all two given CompletableFutures complete exceptionally, + * the returned CompletableFuture also does so, with a CompletionException holding + * an exception from any of the given CompletableFutures as its cause. + *

+ * this method is same as {@link CompletableFuture#runAfterEitherAsync(CompletionStage, Runnable)} + * except for the either-success behavior. + * + * @param action the action to perform before completing the + * returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#runAfterEitherAsync(CompletionStage, Runnable) + */ + public static CompletableFuture runAfterEitherSuccessAsync( + CompletionStage cf1, CompletionStage cf2, Runnable action) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + + return anyOfSuccess(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenRunAsync(action); + } + + /** + * Returns a new CompletableFuture that, when either given stage success, executes the given action + * using the supplied executor. + * Otherwise, all two given CompletableFutures complete exceptionally, + * the returned CompletableFuture also does so, with a CompletionException holding + * an exception from any of the given CompletableFutures as its cause. + *

+ * this method is same as {@link CompletableFuture#runAfterEitherAsync(CompletionStage, Runnable, Executor)} + * except for the either-success behavior. + * + * @param action the action to perform before completing the + * returned CompletableFuture + * @return the new CompletableFuture + * @see CompletableFuture#runAfterEitherAsync(CompletionStage, Runnable, Executor) + */ + + public static CompletableFuture runAfterEitherSuccessAsync( + CompletionStage cf1, CompletionStage cf2, Runnable action, Executor executor) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + requireNonNull(executor, "executor is null"); + + return anyOfSuccess(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenRunAsync(action, executor); + } + + public static CompletableFuture acceptEitherSuccess( + CompletionStage cf1, CompletionStage cf2, Consumer action) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + + return anyOfWithType(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenAccept(action); + } + + public static CompletableFuture acceptEitherSuccessAsync( + CompletionStage cf1, CompletionStage cf2, Consumer action) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + + return anyOfWithType(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenAcceptAsync(action); + } + + public static CompletableFuture acceptEitherSuccessAsync( + CompletionStage cf1, CompletionStage cf2, + Consumer action, Executor executor) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(action, "action is null"); + requireNonNull(executor, "executor is null"); + + return anyOfWithType(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenAcceptAsync(action, executor); + } + + public static CompletableFuture applyToEitherSuccess( + CompletionStage cf1, CompletionStage cf2, Function fn) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(fn, "fn is null"); + + return anyOfWithType(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenApply(fn); + } + + public static CompletableFuture applyToEitherSuccessAsync( + CompletionStage cf1, CompletionStage cf2, Function fn) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(fn, "fn is null"); + + return anyOfWithType(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenApplyAsync(fn); + } + + public static CompletableFuture applyToEitherSuccessAsync( + CompletionStage cf1, CompletionStage cf2, Function fn, Executor executor) { + requireCfsAndEleNonNull(cf1, cf2); + requireNonNull(fn, "fn is null"); + requireNonNull(executor, "executor is null"); + + return anyOfWithType(cf1.toCompletableFuture(), cf2.toCompletableFuture()).thenApplyAsync(fn, executor); + } + //////////////////////////////////////////////////////////////////////////////// //# Backport CF methods // compatibility for low Java version