ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/test/tck/CompletableFutureTest.java
(Generate patch)

Comparing jsr166/src/test/tck/CompletableFutureTest.java (file contents):
Revision 1.122 by jsr166, Sun Sep 6 22:21:07 2015 UTC vs.
Revision 1.148 by jsr166, Sun Jun 26 15:46:22 2016 UTC

# Line 7 | Line 7
7  
8   import static java.util.concurrent.TimeUnit.MILLISECONDS;
9   import static java.util.concurrent.TimeUnit.SECONDS;
10 + import static java.util.concurrent.CompletableFuture.completedFuture;
11 + import static java.util.concurrent.CompletableFuture.failedFuture;
12 +
13 + import java.lang.reflect.Method;
14 + import java.lang.reflect.Modifier;
15 +
16 + import java.util.stream.Collectors;
17 + import java.util.stream.Stream;
18  
19   import java.util.ArrayList;
20 + import java.util.Arrays;
21   import java.util.List;
22   import java.util.Objects;
23 + import java.util.Set;
24   import java.util.concurrent.Callable;
25   import java.util.concurrent.CancellationException;
26   import java.util.concurrent.CompletableFuture;
# Line 28 | Line 38 | import java.util.function.BiConsumer;
38   import java.util.function.BiFunction;
39   import java.util.function.Consumer;
40   import java.util.function.Function;
41 + import java.util.function.Predicate;
42   import java.util.function.Supplier;
43  
44 + import junit.framework.AssertionFailedError;
45   import junit.framework.Test;
46   import junit.framework.TestSuite;
47  
# Line 828 | Line 840 | public class CompletableFutureTest exten
840          if (!createIncomplete) assertTrue(f.complete(v1));
841          final CompletableFuture<Integer> g = f.exceptionally
842              ((Throwable t) -> {
831                // Should not be called
843                  a.getAndIncrement();
844 <                throw new AssertionError();
844 >                threadFail("should not be called");
845 >                return null;            // unreached
846              });
847          if (createIncomplete) assertTrue(f.complete(v1));
848  
# Line 864 | Line 876 | public class CompletableFutureTest exten
876          assertEquals(1, a.get());
877      }}
878  
879 +    /**
880 +     * If an "exceptionally action" throws an exception, it completes
881 +     * exceptionally with that exception
882 +     */
883      public void testExceptionally_exceptionalCompletionActionFailed() {
884          for (boolean createIncomplete : new boolean[] { true, false })
885      {
# Line 882 | Line 898 | public class CompletableFutureTest exten
898          if (createIncomplete) f.completeExceptionally(ex1);
899  
900          checkCompletedWithWrappedException(g, ex2);
901 +        checkCompletedExceptionally(f, ex1);
902          assertEquals(1, a.get());
903      }}
904  
# Line 889 | Line 906 | public class CompletableFutureTest exten
906       * whenComplete action executes on normal completion, propagating
907       * source result.
908       */
909 <    public void testWhenComplete_normalCompletion1() {
909 >    public void testWhenComplete_normalCompletion() {
910          for (ExecutionMode m : ExecutionMode.values())
911          for (boolean createIncomplete : new boolean[] { true, false })
912          for (Integer v1 : new Integer[] { 1, null })
# Line 899 | Line 916 | public class CompletableFutureTest exten
916          if (!createIncomplete) assertTrue(f.complete(v1));
917          final CompletableFuture<Integer> g = m.whenComplete
918              (f,
919 <             (Integer x, Throwable t) -> {
919 >             (Integer result, Throwable t) -> {
920                  m.checkExecutionMode();
921 <                threadAssertSame(x, v1);
921 >                threadAssertSame(result, v1);
922                  threadAssertNull(t);
923                  a.getAndIncrement();
924              });
# Line 926 | Line 943 | public class CompletableFutureTest exten
943          if (!createIncomplete) f.completeExceptionally(ex);
944          final CompletableFuture<Integer> g = m.whenComplete
945              (f,
946 <             (Integer x, Throwable t) -> {
946 >             (Integer result, Throwable t) -> {
947                  m.checkExecutionMode();
948 <                threadAssertNull(x);
948 >                threadAssertNull(result);
949                  threadAssertSame(t, ex);
950                  a.getAndIncrement();
951              });
# Line 953 | Line 970 | public class CompletableFutureTest exten
970          if (!createIncomplete) assertTrue(f.cancel(mayInterruptIfRunning));
971          final CompletableFuture<Integer> g = m.whenComplete
972              (f,
973 <             (Integer x, Throwable t) -> {
973 >             (Integer result, Throwable t) -> {
974                  m.checkExecutionMode();
975 <                threadAssertNull(x);
975 >                threadAssertNull(result);
976                  threadAssertTrue(t instanceof CancellationException);
977                  a.getAndIncrement();
978              });
# Line 970 | Line 987 | public class CompletableFutureTest exten
987       * If a whenComplete action throws an exception when triggered by
988       * a normal completion, it completes exceptionally
989       */
990 <    public void testWhenComplete_actionFailed() {
990 >    public void testWhenComplete_sourceCompletedNormallyActionFailed() {
991          for (boolean createIncomplete : new boolean[] { true, false })
992          for (ExecutionMode m : ExecutionMode.values())
993          for (Integer v1 : new Integer[] { 1, null })
# Line 981 | Line 998 | public class CompletableFutureTest exten
998          if (!createIncomplete) assertTrue(f.complete(v1));
999          final CompletableFuture<Integer> g = m.whenComplete
1000              (f,
1001 <             (Integer x, Throwable t) -> {
1001 >             (Integer result, Throwable t) -> {
1002                  m.checkExecutionMode();
1003 <                threadAssertSame(x, v1);
1003 >                threadAssertSame(result, v1);
1004                  threadAssertNull(t);
1005                  a.getAndIncrement();
1006                  throw ex;
# Line 998 | Line 1015 | public class CompletableFutureTest exten
1015      /**
1016       * If a whenComplete action throws an exception when triggered by
1017       * a source completion that also throws an exception, the source
1018 <     * exception takes precedence.
1018 >     * exception takes precedence (unlike handle)
1019       */
1020 <    public void testWhenComplete_actionFailedSourceFailed() {
1020 >    public void testWhenComplete_sourceFailedActionFailed() {
1021          for (boolean createIncomplete : new boolean[] { true, false })
1022          for (ExecutionMode m : ExecutionMode.values())
1023      {
# Line 1012 | Line 1029 | public class CompletableFutureTest exten
1029          if (!createIncomplete) f.completeExceptionally(ex1);
1030          final CompletableFuture<Integer> g = m.whenComplete
1031              (f,
1032 <             (Integer x, Throwable t) -> {
1032 >             (Integer result, Throwable t) -> {
1033                  m.checkExecutionMode();
1034                  threadAssertSame(t, ex1);
1035 <                threadAssertNull(x);
1035 >                threadAssertNull(result);
1036                  a.getAndIncrement();
1037                  throw ex2;
1038              });
# Line 1023 | Line 1040 | public class CompletableFutureTest exten
1040  
1041          checkCompletedWithWrappedException(g, ex1);
1042          checkCompletedExceptionally(f, ex1);
1043 +        if (testImplementationDetails) {
1044 +            assertEquals(1, ex1.getSuppressed().length);
1045 +            assertSame(ex2, ex1.getSuppressed()[0]);
1046 +        }
1047          assertEquals(1, a.get());
1048      }}
1049  
# Line 1040 | Line 1061 | public class CompletableFutureTest exten
1061          if (!createIncomplete) assertTrue(f.complete(v1));
1062          final CompletableFuture<Integer> g = m.handle
1063              (f,
1064 <             (Integer x, Throwable t) -> {
1064 >             (Integer result, Throwable t) -> {
1065                  m.checkExecutionMode();
1066 <                threadAssertSame(x, v1);
1066 >                threadAssertSame(result, v1);
1067                  threadAssertNull(t);
1068                  a.getAndIncrement();
1069                  return inc(v1);
# Line 1069 | Line 1090 | public class CompletableFutureTest exten
1090          if (!createIncomplete) f.completeExceptionally(ex);
1091          final CompletableFuture<Integer> g = m.handle
1092              (f,
1093 <             (Integer x, Throwable t) -> {
1093 >             (Integer result, Throwable t) -> {
1094                  m.checkExecutionMode();
1095 <                threadAssertNull(x);
1095 >                threadAssertNull(result);
1096                  threadAssertSame(t, ex);
1097                  a.getAndIncrement();
1098                  return v1;
# Line 1098 | Line 1119 | public class CompletableFutureTest exten
1119          if (!createIncomplete) assertTrue(f.cancel(mayInterruptIfRunning));
1120          final CompletableFuture<Integer> g = m.handle
1121              (f,
1122 <             (Integer x, Throwable t) -> {
1122 >             (Integer result, Throwable t) -> {
1123                  m.checkExecutionMode();
1124 <                threadAssertNull(x);
1124 >                threadAssertNull(result);
1125                  threadAssertTrue(t instanceof CancellationException);
1126                  a.getAndIncrement();
1127                  return v1;
# Line 1113 | Line 1134 | public class CompletableFutureTest exten
1134      }}
1135  
1136      /**
1137 <     * handle result completes exceptionally if action does
1137 >     * If a "handle action" throws an exception when triggered by
1138 >     * a normal completion, it completes exceptionally
1139       */
1140 <    public void testHandle_sourceFailedActionFailed() {
1140 >    public void testHandle_sourceCompletedNormallyActionFailed() {
1141          for (ExecutionMode m : ExecutionMode.values())
1142          for (boolean createIncomplete : new boolean[] { true, false })
1143 +        for (Integer v1 : new Integer[] { 1, null })
1144      {
1145          final CompletableFuture<Integer> f = new CompletableFuture<>();
1146          final AtomicInteger a = new AtomicInteger(0);
1147 <        final CFException ex1 = new CFException();
1148 <        final CFException ex2 = new CFException();
1126 <        if (!createIncomplete) f.completeExceptionally(ex1);
1147 >        final CFException ex = new CFException();
1148 >        if (!createIncomplete) assertTrue(f.complete(v1));
1149          final CompletableFuture<Integer> g = m.handle
1150              (f,
1151 <             (Integer x, Throwable t) -> {
1151 >             (Integer result, Throwable t) -> {
1152                  m.checkExecutionMode();
1153 <                threadAssertNull(x);
1154 <                threadAssertSame(ex1, t);
1153 >                threadAssertSame(result, v1);
1154 >                threadAssertNull(t);
1155                  a.getAndIncrement();
1156 <                throw ex2;
1156 >                throw ex;
1157              });
1158 <        if (createIncomplete) f.completeExceptionally(ex1);
1158 >        if (createIncomplete) assertTrue(f.complete(v1));
1159  
1160 <        checkCompletedWithWrappedException(g, ex2);
1161 <        checkCompletedExceptionally(f, ex1);
1160 >        checkCompletedWithWrappedException(g, ex);
1161 >        checkCompletedNormally(f, v1);
1162          assertEquals(1, a.get());
1163      }}
1164  
1165 <    public void testHandle_sourceCompletedNormallyActionFailed() {
1166 <        for (ExecutionMode m : ExecutionMode.values())
1165 >    /**
1166 >     * If a "handle action" throws an exception when triggered by
1167 >     * a source completion that also throws an exception, the action
1168 >     * exception takes precedence (unlike whenComplete)
1169 >     */
1170 >    public void testHandle_sourceFailedActionFailed() {
1171          for (boolean createIncomplete : new boolean[] { true, false })
1172 <        for (Integer v1 : new Integer[] { 1, null })
1172 >        for (ExecutionMode m : ExecutionMode.values())
1173      {
1148        final CompletableFuture<Integer> f = new CompletableFuture<>();
1174          final AtomicInteger a = new AtomicInteger(0);
1175 <        final CFException ex = new CFException();
1176 <        if (!createIncomplete) assertTrue(f.complete(v1));
1175 >        final CFException ex1 = new CFException();
1176 >        final CFException ex2 = new CFException();
1177 >        final CompletableFuture<Integer> f = new CompletableFuture<>();
1178 >
1179 >        if (!createIncomplete) f.completeExceptionally(ex1);
1180          final CompletableFuture<Integer> g = m.handle
1181              (f,
1182 <             (Integer x, Throwable t) -> {
1182 >             (Integer result, Throwable t) -> {
1183                  m.checkExecutionMode();
1184 <                threadAssertSame(x, v1);
1185 <                threadAssertNull(t);
1184 >                threadAssertNull(result);
1185 >                threadAssertSame(ex1, t);
1186                  a.getAndIncrement();
1187 <                throw ex;
1187 >                throw ex2;
1188              });
1189 <        if (createIncomplete) assertTrue(f.complete(v1));
1189 >        if (createIncomplete) f.completeExceptionally(ex1);
1190  
1191 <        checkCompletedWithWrappedException(g, ex);
1192 <        checkCompletedNormally(f, v1);
1191 >        checkCompletedWithWrappedException(g, ex2);
1192 >        checkCompletedExceptionally(f, ex1);
1193          assertEquals(1, a.get());
1194      }}
1195  
# Line 3267 | Line 3295 | public class CompletableFutureTest exten
3295              () -> f.obtrudeException(null),
3296  
3297              () -> CompletableFuture.delayedExecutor(1L, SECONDS, null),
3298 <            () -> CompletableFuture.delayedExecutor(1L, null, new ThreadExecutor()),
3298 >            () -> CompletableFuture.delayedExecutor(1L, null, exec),
3299              () -> CompletableFuture.delayedExecutor(1L, null),
3300  
3301              () -> f.orTimeout(1L, null),
# Line 3497 | Line 3525 | public class CompletableFutureTest exten
3525          long timeoutMillis = timeoutMillis();
3526          CompletableFuture<Integer> f = new CompletableFuture<>();
3527          long startTime = System.nanoTime();
3528 <        f.orTimeout(timeoutMillis, MILLISECONDS);
3528 >        assertSame(f, f.orTimeout(timeoutMillis, MILLISECONDS));
3529          checkCompletedWithTimeoutException(f);
3530          assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
3531      }
# Line 3512 | Line 3540 | public class CompletableFutureTest exten
3540          CompletableFuture<Integer> g = new CompletableFuture<>();
3541          long startTime = System.nanoTime();
3542          f.complete(v1);
3543 <        f.orTimeout(LONG_DELAY_MS, MILLISECONDS);
3544 <        g.orTimeout(LONG_DELAY_MS, MILLISECONDS);
3543 >        assertSame(f, f.orTimeout(LONG_DELAY_MS, MILLISECONDS));
3544 >        assertSame(g, g.orTimeout(LONG_DELAY_MS, MILLISECONDS));
3545          g.complete(v1);
3546          checkCompletedNormally(f, v1);
3547          checkCompletedNormally(g, v1);
# Line 3528 | Line 3556 | public class CompletableFutureTest exten
3556                         () -> testCompleteOnTimeout_timesOut(null));
3557      }
3558  
3559 +    /**
3560 +     * completeOnTimeout completes with given value if not complete
3561 +     */
3562      public void testCompleteOnTimeout_timesOut(Integer v) {
3563          long timeoutMillis = timeoutMillis();
3564          CompletableFuture<Integer> f = new CompletableFuture<>();
3565          long startTime = System.nanoTime();
3566 <        f.completeOnTimeout(v, timeoutMillis, MILLISECONDS);
3566 >        assertSame(f, f.completeOnTimeout(v, timeoutMillis, MILLISECONDS));
3567          assertSame(v, f.join());
3568          assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
3569          f.complete(99);         // should have no effect
# Line 3549 | Line 3580 | public class CompletableFutureTest exten
3580          CompletableFuture<Integer> g = new CompletableFuture<>();
3581          long startTime = System.nanoTime();
3582          f.complete(v1);
3583 <        f.completeOnTimeout(-1, LONG_DELAY_MS, MILLISECONDS);
3584 <        g.completeOnTimeout(-1, LONG_DELAY_MS, MILLISECONDS);
3583 >        assertSame(f, f.completeOnTimeout(-1, LONG_DELAY_MS, MILLISECONDS));
3584 >        assertSame(g, g.completeOnTimeout(-1, LONG_DELAY_MS, MILLISECONDS));
3585          g.complete(v1);
3586          checkCompletedNormally(f, v1);
3587          checkCompletedNormally(g, v1);
# Line 3601 | Line 3632 | public class CompletableFutureTest exten
3632      //--- tests of implementation details; not part of official tck ---
3633  
3634      Object resultOf(CompletableFuture<?> f) {
3635 +        SecurityManager sm = System.getSecurityManager();
3636 +        if (sm != null) {
3637 +            try {
3638 +                System.setSecurityManager(null);
3639 +            } catch (SecurityException giveUp) {
3640 +                return "Reflection not available";
3641 +            }
3642 +        }
3643 +
3644          try {
3645              java.lang.reflect.Field resultField
3646                  = CompletableFuture.class.getDeclaredField("result");
3647              resultField.setAccessible(true);
3648              return resultField.get(f);
3649 <        } catch (Throwable t) { throw new AssertionError(t); }
3649 >        } catch (Throwable t) {
3650 >            throw new AssertionError(t);
3651 >        } finally {
3652 >            if (sm != null) System.setSecurityManager(sm);
3653 >        }
3654      }
3655  
3656      public void testExceptionPropagationReusesResultObject() {
# Line 3629 | Line 3673 | public class CompletableFutureTest exten
3673          funs.add((y) -> m.applyToEither(y, incomplete, new IncFunction(m)));
3674  
3675          funs.add((y) -> m.runAfterBoth(y, v42, new Noop(m)));
3676 +        funs.add((y) -> m.runAfterBoth(v42, y, new Noop(m)));
3677          funs.add((y) -> m.thenAcceptBoth(y, v42, new SubtractAction(m)));
3678 +        funs.add((y) -> m.thenAcceptBoth(v42, y, new SubtractAction(m)));
3679          funs.add((y) -> m.thenCombine(y, v42, new SubtractFunction(m)));
3680 +        funs.add((y) -> m.thenCombine(v42, y, new SubtractFunction(m)));
3681  
3682 <        funs.add((y) -> m.whenComplete(y, (Integer x, Throwable t) -> {}));
3682 >        funs.add((y) -> m.whenComplete(y, (Integer r, Throwable t) -> {}));
3683  
3684          funs.add((y) -> m.thenCompose(y, new CompletableFutureInc(m)));
3685  
3686          funs.add((y) -> CompletableFuture.allOf(new CompletableFuture<?>[] {y, v42}));
3687 +        funs.add((y) -> CompletableFuture.allOf(new CompletableFuture<?>[] {v42, y}));
3688          funs.add((y) -> CompletableFuture.anyOf(new CompletableFuture<?>[] {y, incomplete}));
3689 +        funs.add((y) -> CompletableFuture.anyOf(new CompletableFuture<?>[] {incomplete, y}));
3690  
3691          for (Function<CompletableFuture<Integer>, CompletableFuture<?>>
3692                   fun : funs) {
# Line 3688 | Line 3737 | public class CompletableFutureTest exten
3737          }
3738      }}
3739  
3740 +    /**
3741 +     * Minimal completion stages throw UOE for all non-CompletionStage methods
3742 +     */
3743 +    public void testMinimalCompletionStage_minimality() {
3744 +        if (!testImplementationDetails) return;
3745 +        Function<Method, String> toSignature =
3746 +            (method) -> method.getName() + Arrays.toString(method.getParameterTypes());
3747 +        Predicate<Method> isNotStatic =
3748 +            (method) -> (method.getModifiers() & Modifier.STATIC) == 0;
3749 +        List<Method> minimalMethods =
3750 +            Stream.of(Object.class, CompletionStage.class)
3751 +            .flatMap((klazz) -> Stream.of(klazz.getMethods()))
3752 +            .filter(isNotStatic)
3753 +            .collect(Collectors.toList());
3754 +        // Methods from CompletableFuture permitted NOT to throw UOE
3755 +        String[] signatureWhitelist = {
3756 +            "newIncompleteFuture[]",
3757 +            "defaultExecutor[]",
3758 +            "minimalCompletionStage[]",
3759 +            "copy[]",
3760 +        };
3761 +        Set<String> permittedMethodSignatures =
3762 +            Stream.concat(minimalMethods.stream().map(toSignature),
3763 +                          Stream.of(signatureWhitelist))
3764 +            .collect(Collectors.toSet());
3765 +        List<Method> allMethods = Stream.of(CompletableFuture.class.getMethods())
3766 +            .filter(isNotStatic)
3767 +            .filter((method) -> !permittedMethodSignatures.contains(toSignature.apply(method)))
3768 +            .collect(Collectors.toList());
3769 +
3770 +        CompletionStage<Integer> minimalStage =
3771 +            new CompletableFuture<Integer>().minimalCompletionStage();
3772 +
3773 +        List<Method> bugs = new ArrayList<>();
3774 +        for (Method method : allMethods) {
3775 +            Class<?>[] parameterTypes = method.getParameterTypes();
3776 +            Object[] args = new Object[parameterTypes.length];
3777 +            // Manufacture boxed primitives for primitive params
3778 +            for (int i = 0; i < args.length; i++) {
3779 +                Class<?> type = parameterTypes[i];
3780 +                if (parameterTypes[i] == boolean.class)
3781 +                    args[i] = false;
3782 +                else if (parameterTypes[i] == int.class)
3783 +                    args[i] = 0;
3784 +                else if (parameterTypes[i] == long.class)
3785 +                    args[i] = 0L;
3786 +            }
3787 +            try {
3788 +                method.invoke(minimalStage, args);
3789 +                bugs.add(method);
3790 +            }
3791 +            catch (java.lang.reflect.InvocationTargetException expected) {
3792 +                if (! (expected.getCause() instanceof UnsupportedOperationException)) {
3793 +                    bugs.add(method);
3794 +                    // expected.getCause().printStackTrace();
3795 +                }
3796 +            }
3797 +            catch (ReflectiveOperationException bad) { throw new Error(bad); }
3798 +        }
3799 +        if (!bugs.isEmpty())
3800 +            throw new Error("Methods did not throw UOE: " + bugs.toString());
3801 +    }
3802 +
3803 +    static class Monad {
3804 +        static class ZeroException extends RuntimeException {
3805 +            public ZeroException() { super("monadic zero"); }
3806 +        }
3807 +        // "return", "unit"
3808 +        static <T> CompletableFuture<T> unit(T value) {
3809 +            return completedFuture(value);
3810 +        }
3811 +        // monadic zero ?
3812 +        static <T> CompletableFuture<T> zero() {
3813 +            return failedFuture(new ZeroException());
3814 +        }
3815 +        // >=>
3816 +        static <T,U,V> Function<T, CompletableFuture<V>> compose
3817 +            (Function<T, CompletableFuture<U>> f,
3818 +             Function<U, CompletableFuture<V>> g) {
3819 +            return (x) -> f.apply(x).thenCompose(g);
3820 +        }
3821 +
3822 +        static void assertZero(CompletableFuture<?> f) {
3823 +            try {
3824 +                f.getNow(null);
3825 +                throw new AssertionFailedError("should throw");
3826 +            } catch (CompletionException success) {
3827 +                assertTrue(success.getCause() instanceof ZeroException);
3828 +            }
3829 +        }
3830 +
3831 +        static <T> void assertFutureEquals(CompletableFuture<T> f,
3832 +                                           CompletableFuture<T> g) {
3833 +            T fval = null, gval = null;
3834 +            Throwable fex = null, gex = null;
3835 +
3836 +            try { fval = f.get(); }
3837 +            catch (ExecutionException ex) { fex = ex.getCause(); }
3838 +            catch (Throwable ex) { fex = ex; }
3839 +
3840 +            try { gval = g.get(); }
3841 +            catch (ExecutionException ex) { gex = ex.getCause(); }
3842 +            catch (Throwable ex) { gex = ex; }
3843 +
3844 +            if (fex != null || gex != null)
3845 +                assertSame(fex.getClass(), gex.getClass());
3846 +            else
3847 +                assertEquals(fval, gval);
3848 +        }
3849 +
3850 +        static class PlusFuture<T> extends CompletableFuture<T> {
3851 +            AtomicReference<Throwable> firstFailure = new AtomicReference<>(null);
3852 +        }
3853 +
3854 +        /** Implements "monadic plus". */
3855 +        static <T> CompletableFuture<T> plus(CompletableFuture<? extends T> f,
3856 +                                             CompletableFuture<? extends T> g) {
3857 +            PlusFuture<T> plus = new PlusFuture<T>();
3858 +            BiConsumer<T, Throwable> action = (T result, Throwable ex) -> {
3859 +                try {
3860 +                    if (ex == null) {
3861 +                        if (plus.complete(result))
3862 +                            if (plus.firstFailure.get() != null)
3863 +                                plus.firstFailure.set(null);
3864 +                    }
3865 +                    else if (plus.firstFailure.compareAndSet(null, ex)) {
3866 +                        if (plus.isDone())
3867 +                            plus.firstFailure.set(null);
3868 +                    }
3869 +                    else {
3870 +                        // first failure has precedence
3871 +                        Throwable first = plus.firstFailure.getAndSet(null);
3872 +
3873 +                        // may fail with "Self-suppression not permitted"
3874 +                        try { first.addSuppressed(ex); }
3875 +                        catch (Exception ignored) {}
3876 +
3877 +                        plus.completeExceptionally(first);
3878 +                    }
3879 +                } catch (Throwable unexpected) {
3880 +                    plus.completeExceptionally(unexpected);
3881 +                }
3882 +            };
3883 +            f.whenComplete(action);
3884 +            g.whenComplete(action);
3885 +            return plus;
3886 +        }
3887 +    }
3888 +
3889 +    /**
3890 +     * CompletableFuture is an additive monad - sort of.
3891 +     * https://en.wikipedia.org/wiki/Monad_(functional_programming)#Additive_monads
3892 +     */
3893 +    public void testAdditiveMonad() throws Throwable {
3894 +        Function<Long, CompletableFuture<Long>> unit = Monad::unit;
3895 +        CompletableFuture<Long> zero = Monad.zero();
3896 +
3897 +        // Some mutually non-commutative functions
3898 +        Function<Long, CompletableFuture<Long>> triple
3899 +            = (x) -> Monad.unit(3 * x);
3900 +        Function<Long, CompletableFuture<Long>> inc
3901 +            = (x) -> Monad.unit(x + 1);
3902 +
3903 +        // unit is a right identity: m >>= unit === m
3904 +        Monad.assertFutureEquals(inc.apply(5L).thenCompose(unit),
3905 +                                 inc.apply(5L));
3906 +        // unit is a left identity: (unit x) >>= f === f x
3907 +        Monad.assertFutureEquals(unit.apply(5L).thenCompose(inc),
3908 +                                 inc.apply(5L));
3909 +
3910 +        // associativity: (m >>= f) >>= g === m >>= ( \x -> (f x >>= g) )
3911 +        Monad.assertFutureEquals(
3912 +            unit.apply(5L).thenCompose(inc).thenCompose(triple),
3913 +            unit.apply(5L).thenCompose((x) -> inc.apply(x).thenCompose(triple)));
3914 +
3915 +        // The case for CompletableFuture as an additive monad is weaker...
3916 +
3917 +        // zero is a monadic zero
3918 +        Monad.assertZero(zero);
3919 +
3920 +        // left zero: zero >>= f === zero
3921 +        Monad.assertZero(zero.thenCompose(inc));
3922 +        // right zero: f >>= (\x -> zero) === zero
3923 +        Monad.assertZero(inc.apply(5L).thenCompose((x) -> zero));
3924 +
3925 +        // f plus zero === f
3926 +        Monad.assertFutureEquals(Monad.unit(5L),
3927 +                                 Monad.plus(Monad.unit(5L), zero));
3928 +        // zero plus f === f
3929 +        Monad.assertFutureEquals(Monad.unit(5L),
3930 +                                 Monad.plus(zero, Monad.unit(5L)));
3931 +        // zero plus zero === zero
3932 +        Monad.assertZero(Monad.plus(zero, zero));
3933 +        {
3934 +            CompletableFuture<Long> f = Monad.plus(Monad.unit(5L),
3935 +                                                   Monad.unit(8L));
3936 +            // non-determinism
3937 +            assertTrue(f.get() == 5L || f.get() == 8L);
3938 +        }
3939 +
3940 +        CompletableFuture<Long> godot = new CompletableFuture<>();
3941 +        // f plus godot === f (doesn't wait for godot)
3942 +        Monad.assertFutureEquals(Monad.unit(5L),
3943 +                                 Monad.plus(Monad.unit(5L), godot));
3944 +        // godot plus f === f (doesn't wait for godot)
3945 +        Monad.assertFutureEquals(Monad.unit(5L),
3946 +                                 Monad.plus(godot, Monad.unit(5L)));
3947 +    }
3948 +
3949 +    /**
3950 +     * A single CompletableFuture with many dependents.
3951 +     * A demo of scalability - runtime is O(n).
3952 +     */
3953 +    public void testManyDependents() throws Throwable {
3954 +        final int n = 1_000;
3955 +        final CompletableFuture<Void> head = new CompletableFuture<>();
3956 +        final CompletableFuture<Void> complete = CompletableFuture.completedFuture((Void)null);
3957 +        final AtomicInteger count = new AtomicInteger(0);
3958 +        for (int i = 0; i < n; i++) {
3959 +            head.thenRun(() -> count.getAndIncrement());
3960 +            head.thenAccept((x) -> count.getAndIncrement());
3961 +            head.thenApply((x) -> count.getAndIncrement());
3962 +
3963 +            head.runAfterBoth(complete, () -> count.getAndIncrement());
3964 +            head.thenAcceptBoth(complete, (x, y) -> count.getAndIncrement());
3965 +            head.thenCombine(complete, (x, y) -> count.getAndIncrement());
3966 +            complete.runAfterBoth(head, () -> count.getAndIncrement());
3967 +            complete.thenAcceptBoth(head, (x, y) -> count.getAndIncrement());
3968 +            complete.thenCombine(head, (x, y) -> count.getAndIncrement());
3969 +
3970 +            head.runAfterEither(new CompletableFuture<Void>(), () -> count.getAndIncrement());
3971 +            head.acceptEither(new CompletableFuture<Void>(), (x) -> count.getAndIncrement());
3972 +            head.applyToEither(new CompletableFuture<Void>(), (x) -> count.getAndIncrement());
3973 +            new CompletableFuture<Void>().runAfterEither(head, () -> count.getAndIncrement());
3974 +            new CompletableFuture<Void>().acceptEither(head, (x) -> count.getAndIncrement());
3975 +            new CompletableFuture<Void>().applyToEither(head, (x) -> count.getAndIncrement());
3976 +        }
3977 +        head.complete(null);
3978 +        assertEquals(5 * 3 * n, count.get());
3979 +    }
3980 +
3981 + //     static <U> U join(CompletionStage<U> stage) {
3982 + //         CompletableFuture<U> f = new CompletableFuture<>();
3983 + //         stage.whenComplete((v, ex) -> {
3984 + //             if (ex != null) f.completeExceptionally(ex); else f.complete(v);
3985 + //         });
3986 + //         return f.join();
3987 + //     }
3988 +
3989 + //     static <U> boolean isDone(CompletionStage<U> stage) {
3990 + //         CompletableFuture<U> f = new CompletableFuture<>();
3991 + //         stage.whenComplete((v, ex) -> {
3992 + //             if (ex != null) f.completeExceptionally(ex); else f.complete(v);
3993 + //         });
3994 + //         return f.isDone();
3995 + //     }
3996 +
3997 + //     static <U> U join2(CompletionStage<U> stage) {
3998 + //         return stage.toCompletableFuture().copy().join();
3999 + //     }
4000 +
4001 + //     static <U> boolean isDone2(CompletionStage<U> stage) {
4002 + //         return stage.toCompletableFuture().copy().isDone();
4003 + //     }
4004 +
4005   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines