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.92 by jsr166, Tue Jun 17 20:50:01 2014 UTC vs.
Revision 1.134 by jsr166, Sun Nov 15 19:55:38 2015 UTC

# Line 5 | Line 5
5   * http://creativecommons.org/publicdomain/zero/1.0/
6   */
7  
8 < import junit.framework.*;
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;
10 import java.util.concurrent.Executor;
11 import java.util.concurrent.ExecutorService;
12 import java.util.concurrent.Executors;
25   import java.util.concurrent.CancellationException;
14 import java.util.concurrent.CountDownLatch;
15 import java.util.concurrent.ExecutionException;
16 import java.util.concurrent.Future;
26   import java.util.concurrent.CompletableFuture;
27   import java.util.concurrent.CompletionException;
28   import java.util.concurrent.CompletionStage;
29 + import java.util.concurrent.ExecutionException;
30 + import java.util.concurrent.Executor;
31   import java.util.concurrent.ForkJoinPool;
32   import java.util.concurrent.ForkJoinTask;
33   import java.util.concurrent.TimeoutException;
34 + import java.util.concurrent.TimeUnit;
35   import java.util.concurrent.atomic.AtomicInteger;
36 < import static java.util.concurrent.TimeUnit.MILLISECONDS;
25 < import static java.util.concurrent.TimeUnit.SECONDS;
26 < import java.util.*;
27 < import java.util.function.Supplier;
28 < import java.util.function.Consumer;
36 > import java.util.concurrent.atomic.AtomicReference;
37   import java.util.function.BiConsumer;
30 import java.util.function.Function;
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  
48   public class CompletableFutureTest extends JSR166TestCase {
49  
50      public static void main(String[] args) {
51 <        junit.textui.TestRunner.run(suite());
51 >        main(suite(), args);
52      }
53      public static Test suite() {
54          return new TestSuite(CompletableFutureTest.class);
# Line 44 | Line 59 | public class CompletableFutureTest exten
59      void checkIncomplete(CompletableFuture<?> f) {
60          assertFalse(f.isDone());
61          assertFalse(f.isCancelled());
62 <        assertTrue(f.toString().contains("[Not completed]"));
62 >        assertTrue(f.toString().contains("Not completed"));
63          try {
64              assertNull(f.getNow(null));
65          } catch (Throwable fail) { threadUnexpectedException(fail); }
# Line 57 | Line 72 | public class CompletableFutureTest exten
72      }
73  
74      <T> void checkCompletedNormally(CompletableFuture<T> f, T value) {
75 <        try {
76 <            assertEquals(value, f.get(LONG_DELAY_MS, MILLISECONDS));
62 <        } catch (Throwable fail) { threadUnexpectedException(fail); }
75 >        checkTimedGet(f, value);
76 >
77          try {
78              assertEquals(value, f.join());
79          } catch (Throwable fail) { threadUnexpectedException(fail); }
# Line 75 | Line 89 | public class CompletableFutureTest exten
89          assertTrue(f.toString().contains("[Completed normally]"));
90      }
91  
92 <    void checkCompletedWithWrappedCFException(CompletableFuture<?> f) {
93 <        try {
94 <            f.get(LONG_DELAY_MS, MILLISECONDS);
95 <            shouldThrow();
96 <        } catch (ExecutionException success) {
97 <            assertTrue(success.getCause() instanceof CFException);
98 <        } catch (Throwable fail) { threadUnexpectedException(fail); }
99 <        try {
100 <            f.join();
101 <            shouldThrow();
102 <        } catch (CompletionException success) {
103 <            assertTrue(success.getCause() instanceof CFException);
104 <        }
105 <        try {
106 <            f.getNow(null);
107 <            shouldThrow();
108 <        } catch (CompletionException success) {
95 <            assertTrue(success.getCause() instanceof CFException);
92 >    /**
93 >     * Returns the "raw" internal exceptional completion of f,
94 >     * without any additional wrapping with CompletionException.
95 >     */
96 >    <U> Throwable exceptionalCompletion(CompletableFuture<U> f) {
97 >        // handle (and whenComplete) can distinguish between "direct"
98 >        // and "wrapped" exceptional completion
99 >        return f.handle((U u, Throwable t) -> t).join();
100 >    }
101 >
102 >    void checkCompletedExceptionally(CompletableFuture<?> f,
103 >                                     boolean wrapped,
104 >                                     Consumer<Throwable> checker) {
105 >        Throwable cause = exceptionalCompletion(f);
106 >        if (wrapped) {
107 >            assertTrue(cause instanceof CompletionException);
108 >            cause = cause.getCause();
109          }
110 <        try {
98 <            f.get();
99 <            shouldThrow();
100 <        } catch (ExecutionException success) {
101 <            assertTrue(success.getCause() instanceof CFException);
102 <        } catch (Throwable fail) { threadUnexpectedException(fail); }
103 <        assertTrue(f.isDone());
104 <        assertFalse(f.isCancelled());
105 <        assertTrue(f.toString().contains("[Completed exceptionally]"));
106 <    }
110 >        checker.accept(cause);
111  
112 <    <U> void checkCompletedExceptionallyWithRootCause(CompletableFuture<U> f,
109 <                                                      Throwable ex) {
112 >        long startTime = System.nanoTime();
113          try {
114              f.get(LONG_DELAY_MS, MILLISECONDS);
115              shouldThrow();
116          } catch (ExecutionException success) {
117 <            assertSame(ex, success.getCause());
117 >            assertSame(cause, success.getCause());
118          } catch (Throwable fail) { threadUnexpectedException(fail); }
119 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
120 +
121          try {
122              f.join();
123              shouldThrow();
124          } catch (CompletionException success) {
125 <            assertSame(ex, success.getCause());
126 <        }
125 >            assertSame(cause, success.getCause());
126 >        } catch (Throwable fail) { threadUnexpectedException(fail); }
127 >
128          try {
129              f.getNow(null);
130              shouldThrow();
131          } catch (CompletionException success) {
132 <            assertSame(ex, success.getCause());
133 <        }
132 >            assertSame(cause, success.getCause());
133 >        } catch (Throwable fail) { threadUnexpectedException(fail); }
134 >
135          try {
136              f.get();
137              shouldThrow();
138          } catch (ExecutionException success) {
139 <            assertSame(ex, success.getCause());
139 >            assertSame(cause, success.getCause());
140          } catch (Throwable fail) { threadUnexpectedException(fail); }
141  
135        assertTrue(f.isDone());
142          assertFalse(f.isCancelled());
143 +        assertTrue(f.isDone());
144 +        assertTrue(f.isCompletedExceptionally());
145          assertTrue(f.toString().contains("[Completed exceptionally]"));
146      }
147  
148 <    <U> void checkCompletedWithWrappedException(CompletableFuture<U> f,
149 <                                                Throwable ex) {
150 <        checkCompletedExceptionallyWithRootCause(f, ex);
143 <        try {
144 <            CompletableFuture<Throwable> spy = f.handle
145 <                ((U u, Throwable t) -> t);
146 <            assertTrue(spy.join() instanceof CompletionException);
147 <            assertSame(ex, spy.join().getCause());
148 <        } catch (Throwable fail) { threadUnexpectedException(fail); }
148 >    void checkCompletedWithWrappedCFException(CompletableFuture<?> f) {
149 >        checkCompletedExceptionally(f, true,
150 >            (t) -> assertTrue(t instanceof CFException));
151      }
152  
153 <    <U> void checkCompletedExceptionally(CompletableFuture<U> f, Throwable ex) {
154 <        checkCompletedExceptionallyWithRootCause(f, ex);
155 <        try {
156 <            CompletableFuture<Throwable> spy = f.handle
157 <                ((U u, Throwable t) -> t);
158 <            assertSame(ex, spy.join());
159 <        } catch (Throwable fail) { threadUnexpectedException(fail); }
153 >    void checkCompletedWithWrappedCancellationException(CompletableFuture<?> f) {
154 >        checkCompletedExceptionally(f, true,
155 >            (t) -> assertTrue(t instanceof CancellationException));
156 >    }
157 >
158 >    void checkCompletedWithTimeoutException(CompletableFuture<?> f) {
159 >        checkCompletedExceptionally(f, false,
160 >            (t) -> assertTrue(t instanceof TimeoutException));
161 >    }
162 >
163 >    void checkCompletedWithWrappedException(CompletableFuture<?> f,
164 >                                            Throwable ex) {
165 >        checkCompletedExceptionally(f, true, (t) -> assertSame(t, ex));
166 >    }
167 >
168 >    void checkCompletedExceptionally(CompletableFuture<?> f, Throwable ex) {
169 >        checkCompletedExceptionally(f, false, (t) -> assertSame(t, ex));
170      }
171  
172      void checkCancelled(CompletableFuture<?> f) {
173 +        long startTime = System.nanoTime();
174          try {
175              f.get(LONG_DELAY_MS, MILLISECONDS);
176              shouldThrow();
177          } catch (CancellationException success) {
178          } catch (Throwable fail) { threadUnexpectedException(fail); }
179 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
180 +
181          try {
182              f.join();
183              shouldThrow();
# Line 176 | Line 191 | public class CompletableFutureTest exten
191              shouldThrow();
192          } catch (CancellationException success) {
193          } catch (Throwable fail) { threadUnexpectedException(fail); }
179        assertTrue(f.isDone());
180        assertTrue(f.isCompletedExceptionally());
181        assertTrue(f.isCancelled());
182        assertTrue(f.toString().contains("[Completed exceptionally]"));
183    }
194  
195 <    void checkCompletedWithWrappedCancellationException(CompletableFuture<?> f) {
196 <        try {
187 <            f.get(LONG_DELAY_MS, MILLISECONDS);
188 <            shouldThrow();
189 <        } catch (ExecutionException success) {
190 <            assertTrue(success.getCause() instanceof CancellationException);
191 <        } catch (Throwable fail) { threadUnexpectedException(fail); }
192 <        try {
193 <            f.join();
194 <            shouldThrow();
195 <        } catch (CompletionException success) {
196 <            assertTrue(success.getCause() instanceof CancellationException);
197 <        }
198 <        try {
199 <            f.getNow(null);
200 <            shouldThrow();
201 <        } catch (CompletionException success) {
202 <            assertTrue(success.getCause() instanceof CancellationException);
203 <        }
204 <        try {
205 <            f.get();
206 <            shouldThrow();
207 <        } catch (ExecutionException success) {
208 <            assertTrue(success.getCause() instanceof CancellationException);
209 <        } catch (Throwable fail) { threadUnexpectedException(fail); }
195 >        assertTrue(exceptionalCompletion(f) instanceof CancellationException);
196 >
197          assertTrue(f.isDone());
211        assertFalse(f.isCancelled());
198          assertTrue(f.isCompletedExceptionally());
199 +        assertTrue(f.isCancelled());
200          assertTrue(f.toString().contains("[Completed exceptionally]"));
201      }
202  
# Line 257 | Line 244 | public class CompletableFutureTest exten
244      {
245          CompletableFuture<Integer> f = new CompletableFuture<>();
246          checkIncomplete(f);
247 <        assertTrue(f.cancel(true));
248 <        assertTrue(f.cancel(true));
247 >        assertTrue(f.cancel(mayInterruptIfRunning));
248 >        assertTrue(f.cancel(mayInterruptIfRunning));
249 >        assertTrue(f.cancel(!mayInterruptIfRunning));
250          checkCancelled(f);
251      }}
252  
# Line 530 | Line 518 | public class CompletableFutureTest exten
518          }
519      }
520  
533
521      class CompletableFutureInc extends CheckedIntegerAction
522          implements Function<Integer, CompletableFuture<Integer>>
523      {
# Line 569 | Line 556 | public class CompletableFutureTest exten
556          }
557      }
558  
559 +    static final boolean defaultExecutorIsCommonPool
560 +        = ForkJoinPool.getCommonPoolParallelism() > 1;
561 +
562      /**
563       * Permits the testing of parallel code for the 3 different
564       * execution modes without copy/pasting all the test methods.
565       */
566      enum ExecutionMode {
567 <        DEFAULT {
567 >        SYNC {
568              public void checkExecutionMode() {
569                  assertFalse(ThreadExecutor.startedCurrentThread());
570                  assertNull(ForkJoinTask.getPool());
# Line 650 | Line 640 | public class CompletableFutureTest exten
640  
641          ASYNC {
642              public void checkExecutionMode() {
643 <                assertSame(ForkJoinPool.commonPool(),
644 <                           ForkJoinTask.getPool());
643 >                assertEquals(defaultExecutorIsCommonPool,
644 >                             (ForkJoinPool.commonPool() == ForkJoinTask.getPool()));
645              }
646              public CompletableFuture<Void> runAsync(Runnable a) {
647                  return CompletableFuture.runAsync(a);
# Line 850 | Line 840 | public class CompletableFutureTest exten
840          if (!createIncomplete) assertTrue(f.complete(v1));
841          final CompletableFuture<Integer> g = f.exceptionally
842              ((Throwable t) -> {
853                // 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 875 | Line 865 | public class CompletableFutureTest exten
865          if (!createIncomplete) f.completeExceptionally(ex);
866          final CompletableFuture<Integer> g = f.exceptionally
867              ((Throwable t) -> {
868 <                ExecutionMode.DEFAULT.checkExecutionMode();
868 >                ExecutionMode.SYNC.checkExecutionMode();
869                  threadAssertSame(t, ex);
870                  a.getAndIncrement();
871                  return v1;
# Line 888 | Line 878 | public class CompletableFutureTest exten
878  
879      public void testExceptionally_exceptionalCompletionActionFailed() {
880          for (boolean createIncomplete : new boolean[] { true, false })
891        for (Integer v1 : new Integer[] { 1, null })
881      {
882          final AtomicInteger a = new AtomicInteger(0);
883          final CFException ex1 = new CFException();
# Line 897 | Line 886 | public class CompletableFutureTest exten
886          if (!createIncomplete) f.completeExceptionally(ex1);
887          final CompletableFuture<Integer> g = f.exceptionally
888              ((Throwable t) -> {
889 <                ExecutionMode.DEFAULT.checkExecutionMode();
889 >                ExecutionMode.SYNC.checkExecutionMode();
890                  threadAssertSame(t, ex1);
891                  a.getAndIncrement();
892                  throw ex2;
# Line 912 | Line 901 | public class CompletableFutureTest exten
901       * whenComplete action executes on normal completion, propagating
902       * source result.
903       */
904 <    public void testWhenComplete_normalCompletion1() {
904 >    public void testWhenComplete_normalCompletion() {
905          for (ExecutionMode m : ExecutionMode.values())
906          for (boolean createIncomplete : new boolean[] { true, false })
907          for (Integer v1 : new Integer[] { 1, null })
# Line 942 | Line 931 | public class CompletableFutureTest exten
931      public void testWhenComplete_exceptionalCompletion() {
932          for (ExecutionMode m : ExecutionMode.values())
933          for (boolean createIncomplete : new boolean[] { true, false })
945        for (Integer v1 : new Integer[] { 1, null })
934      {
935          final AtomicInteger a = new AtomicInteger(0);
936          final CFException ex = new CFException();
# Line 994 | Line 982 | public class CompletableFutureTest exten
982       * If a whenComplete action throws an exception when triggered by
983       * a normal completion, it completes exceptionally
984       */
985 <    public void testWhenComplete_actionFailed() {
985 >    public void testWhenComplete_sourceCompletedNormallyActionFailed() {
986          for (boolean createIncomplete : new boolean[] { true, false })
987          for (ExecutionMode m : ExecutionMode.values())
988          for (Integer v1 : new Integer[] { 1, null })
# Line 1022 | Line 1010 | public class CompletableFutureTest exten
1010      /**
1011       * If a whenComplete action throws an exception when triggered by
1012       * a source completion that also throws an exception, the source
1013 <     * exception takes precedence.
1013 >     * exception takes precedence (unlike handle)
1014       */
1015 <    public void testWhenComplete_actionFailedSourceFailed() {
1015 >    public void testWhenComplete_sourceFailedActionFailed() {
1016          for (boolean createIncomplete : new boolean[] { true, false })
1017          for (ExecutionMode m : ExecutionMode.values())
1030        for (Integer v1 : new Integer[] { 1, null })
1018      {
1019          final AtomicInteger a = new AtomicInteger(0);
1020          final CFException ex1 = new CFException();
# Line 1138 | Line 1125 | public class CompletableFutureTest exten
1125      }}
1126  
1127      /**
1128 <     * handle result completes exceptionally if action does
1128 >     * If a "handle action" throws an exception when triggered by
1129 >     * a normal completion, it completes exceptionally
1130       */
1131 <    public void testHandle_sourceFailedActionFailed() {
1131 >    public void testHandle_sourceCompletedNormallyActionFailed() {
1132          for (ExecutionMode m : ExecutionMode.values())
1133          for (boolean createIncomplete : new boolean[] { true, false })
1134 +        for (Integer v1 : new Integer[] { 1, null })
1135      {
1136          final CompletableFuture<Integer> f = new CompletableFuture<>();
1137          final AtomicInteger a = new AtomicInteger(0);
1138 <        final CFException ex1 = new CFException();
1139 <        final CFException ex2 = new CFException();
1151 <        if (!createIncomplete) f.completeExceptionally(ex1);
1138 >        final CFException ex = new CFException();
1139 >        if (!createIncomplete) assertTrue(f.complete(v1));
1140          final CompletableFuture<Integer> g = m.handle
1141              (f,
1142               (Integer x, Throwable t) -> {
1143                  m.checkExecutionMode();
1144 <                threadAssertNull(x);
1145 <                threadAssertSame(ex1, t);
1144 >                threadAssertSame(x, v1);
1145 >                threadAssertNull(t);
1146                  a.getAndIncrement();
1147 <                throw ex2;
1147 >                throw ex;
1148              });
1149 <        if (createIncomplete) f.completeExceptionally(ex1);
1149 >        if (createIncomplete) assertTrue(f.complete(v1));
1150  
1151 <        checkCompletedWithWrappedException(g, ex2);
1152 <        checkCompletedExceptionally(f, ex1);
1151 >        checkCompletedWithWrappedException(g, ex);
1152 >        checkCompletedNormally(f, v1);
1153          assertEquals(1, a.get());
1154      }}
1155  
1156 <    public void testHandle_sourceCompletedNormallyActionFailed() {
1157 <        for (ExecutionMode m : ExecutionMode.values())
1156 >    /**
1157 >     * If a "handle action" throws an exception when triggered by
1158 >     * a source completion that also throws an exception, the action
1159 >     * exception takes precedence (unlike whenComplete)
1160 >     */
1161 >    public void testHandle_sourceFailedActionFailed() {
1162          for (boolean createIncomplete : new boolean[] { true, false })
1163 <        for (Integer v1 : new Integer[] { 1, null })
1163 >        for (ExecutionMode m : ExecutionMode.values())
1164      {
1173        final CompletableFuture<Integer> f = new CompletableFuture<>();
1165          final AtomicInteger a = new AtomicInteger(0);
1166 <        final CFException ex = new CFException();
1167 <        if (!createIncomplete) assertTrue(f.complete(v1));
1166 >        final CFException ex1 = new CFException();
1167 >        final CFException ex2 = new CFException();
1168 >        final CompletableFuture<Integer> f = new CompletableFuture<>();
1169 >
1170 >        if (!createIncomplete) f.completeExceptionally(ex1);
1171          final CompletableFuture<Integer> g = m.handle
1172              (f,
1173               (Integer x, Throwable t) -> {
1174                  m.checkExecutionMode();
1175 <                threadAssertSame(x, v1);
1176 <                threadAssertNull(t);
1175 >                threadAssertNull(x);
1176 >                threadAssertSame(ex1, t);
1177                  a.getAndIncrement();
1178 <                throw ex;
1178 >                throw ex2;
1179              });
1180 <        if (createIncomplete) assertTrue(f.complete(v1));
1180 >        if (createIncomplete) f.completeExceptionally(ex1);
1181  
1182 <        checkCompletedWithWrappedException(g, ex);
1183 <        checkCompletedNormally(f, v1);
1182 >        checkCompletedWithWrappedException(g, ex2);
1183 >        checkCompletedExceptionally(f, ex1);
1184          assertEquals(1, a.get());
1185      }}
1186  
# Line 1600 | Line 1594 | public class CompletableFutureTest exten
1594      {
1595          final CompletableFuture<Integer> f = new CompletableFuture<>();
1596          final CompletableFuture<Integer> g = new CompletableFuture<>();
1597 <        final SubtractFunction r1 = new SubtractFunction(m);
1598 <        final SubtractFunction r2 = new SubtractFunction(m);
1605 <        final SubtractFunction r3 = new SubtractFunction(m);
1597 >        final SubtractFunction[] rs = new SubtractFunction[6];
1598 >        for (int i = 0; i < rs.length; i++) rs[i] = new SubtractFunction(m);
1599  
1600          final CompletableFuture<Integer> fst =  fFirst ? f : g;
1601          final CompletableFuture<Integer> snd = !fFirst ? f : g;
1602          final Integer w1 =  fFirst ? v1 : v2;
1603          final Integer w2 = !fFirst ? v1 : v2;
1604  
1605 <        final CompletableFuture<Integer> h1 = m.thenCombine(f, g, r1);
1605 >        final CompletableFuture<Integer> h0 = m.thenCombine(f, g, rs[0]);
1606 >        final CompletableFuture<Integer> h1 = m.thenCombine(fst, fst, rs[1]);
1607          assertTrue(fst.complete(w1));
1608 <        final CompletableFuture<Integer> h2 = m.thenCombine(f, g, r2);
1609 <        checkIncomplete(h1);
1610 <        checkIncomplete(h2);
1611 <        r1.assertNotInvoked();
1612 <        r2.assertNotInvoked();
1608 >        final CompletableFuture<Integer> h2 = m.thenCombine(f, g, rs[2]);
1609 >        final CompletableFuture<Integer> h3 = m.thenCombine(fst, fst, rs[3]);
1610 >        checkIncomplete(h0); rs[0].assertNotInvoked();
1611 >        checkIncomplete(h2); rs[2].assertNotInvoked();
1612 >        checkCompletedNormally(h1, subtract(w1, w1));
1613 >        checkCompletedNormally(h3, subtract(w1, w1));
1614 >        rs[1].assertValue(subtract(w1, w1));
1615 >        rs[3].assertValue(subtract(w1, w1));
1616          assertTrue(snd.complete(w2));
1617 <        final CompletableFuture<Integer> h3 = m.thenCombine(f, g, r3);
1617 >        final CompletableFuture<Integer> h4 = m.thenCombine(f, g, rs[4]);
1618  
1619 <        checkCompletedNormally(h1, subtract(v1, v2));
1619 >        checkCompletedNormally(h0, subtract(v1, v2));
1620          checkCompletedNormally(h2, subtract(v1, v2));
1621 <        checkCompletedNormally(h3, subtract(v1, v2));
1622 <        r1.assertValue(subtract(v1, v2));
1623 <        r2.assertValue(subtract(v1, v2));
1624 <        r3.assertValue(subtract(v1, v2));
1621 >        checkCompletedNormally(h4, subtract(v1, v2));
1622 >        rs[0].assertValue(subtract(v1, v2));
1623 >        rs[2].assertValue(subtract(v1, v2));
1624 >        rs[4].assertValue(subtract(v1, v2));
1625 >
1626          checkCompletedNormally(f, v1);
1627          checkCompletedNormally(g, v2);
1628      }}
# Line 2971 | Line 2969 | public class CompletableFutureTest exten
2969          checkCancelled(f);
2970      }}
2971  
2972 +    /**
2973 +     * thenCompose result completes exceptionally if the result of the action does
2974 +     */
2975 +    public void testThenCompose_actionReturnsFailingFuture() {
2976 +        for (ExecutionMode m : ExecutionMode.values())
2977 +        for (int order = 0; order < 6; order++)
2978 +        for (Integer v1 : new Integer[] { 1, null })
2979 +    {
2980 +        final CFException ex = new CFException();
2981 +        final CompletableFuture<Integer> f = new CompletableFuture<>();
2982 +        final CompletableFuture<Integer> g = new CompletableFuture<>();
2983 +        final CompletableFuture<Integer> h;
2984 +        // Test all permutations of orders
2985 +        switch (order) {
2986 +        case 0:
2987 +            assertTrue(f.complete(v1));
2988 +            assertTrue(g.completeExceptionally(ex));
2989 +            h = m.thenCompose(f, (x -> g));
2990 +            break;
2991 +        case 1:
2992 +            assertTrue(f.complete(v1));
2993 +            h = m.thenCompose(f, (x -> g));
2994 +            assertTrue(g.completeExceptionally(ex));
2995 +            break;
2996 +        case 2:
2997 +            assertTrue(g.completeExceptionally(ex));
2998 +            assertTrue(f.complete(v1));
2999 +            h = m.thenCompose(f, (x -> g));
3000 +            break;
3001 +        case 3:
3002 +            assertTrue(g.completeExceptionally(ex));
3003 +            h = m.thenCompose(f, (x -> g));
3004 +            assertTrue(f.complete(v1));
3005 +            break;
3006 +        case 4:
3007 +            h = m.thenCompose(f, (x -> g));
3008 +            assertTrue(f.complete(v1));
3009 +            assertTrue(g.completeExceptionally(ex));
3010 +            break;
3011 +        case 5:
3012 +            h = m.thenCompose(f, (x -> g));
3013 +            assertTrue(f.complete(v1));
3014 +            assertTrue(g.completeExceptionally(ex));
3015 +            break;
3016 +        default: throw new AssertionError();
3017 +        }
3018 +
3019 +        checkCompletedExceptionally(g, ex);
3020 +        checkCompletedWithWrappedException(h, ex);
3021 +        checkCompletedNormally(f, v1);
3022 +    }}
3023 +
3024      // other static methods
3025  
3026      /**
# Line 3031 | Line 3081 | public class CompletableFutureTest exten
3081              for (int i = 0; i < k; i++) {
3082                  checkIncomplete(f);
3083                  checkIncomplete(CompletableFuture.allOf(fs));
3084 <                if (i != k/2) {
3084 >                if (i != k / 2) {
3085                      fs[i].complete(i);
3086                      checkCompletedNormally(fs[i], i);
3087                  } else {
# Line 3138 | Line 3188 | public class CompletableFutureTest exten
3188          CompletableFuture<Integer> f = new CompletableFuture<>();
3189          CompletableFuture<Integer> g = new CompletableFuture<>();
3190          CompletableFuture<Integer> nullFuture = (CompletableFuture<Integer>)null;
3141        CompletableFuture<?> h;
3191          ThreadExecutor exec = new ThreadExecutor();
3192  
3193          Runnable[] throwingActions = {
3194              () -> CompletableFuture.supplyAsync(null),
3195              () -> CompletableFuture.supplyAsync(null, exec),
3196 <            () -> CompletableFuture.supplyAsync(new IntegerSupplier(ExecutionMode.DEFAULT, 42), null),
3196 >            () -> CompletableFuture.supplyAsync(new IntegerSupplier(ExecutionMode.SYNC, 42), null),
3197  
3198              () -> CompletableFuture.runAsync(null),
3199              () -> CompletableFuture.runAsync(null, exec),
# Line 3235 | Line 3284 | public class CompletableFutureTest exten
3284              () -> CompletableFuture.anyOf(null, f),
3285  
3286              () -> f.obtrudeException(null),
3287 +
3288 +            () -> CompletableFuture.delayedExecutor(1L, SECONDS, null),
3289 +            () -> CompletableFuture.delayedExecutor(1L, null, new ThreadExecutor()),
3290 +            () -> CompletableFuture.delayedExecutor(1L, null),
3291 +
3292 +            () -> f.orTimeout(1L, null),
3293 +            () -> f.completeOnTimeout(42, 1L, null),
3294 +
3295 +            () -> CompletableFuture.failedFuture(null),
3296 +            () -> CompletableFuture.failedStage(null),
3297          };
3298  
3299          assertThrows(NullPointerException.class, throwingActions);
# Line 3249 | Line 3308 | public class CompletableFutureTest exten
3308          assertSame(f, f.toCompletableFuture());
3309      }
3310  
3311 +    // jdk9
3312 +
3313 +    /**
3314 +     * newIncompleteFuture returns an incomplete CompletableFuture
3315 +     */
3316 +    public void testNewIncompleteFuture() {
3317 +        for (Integer v1 : new Integer[] { 1, null })
3318 +    {
3319 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3320 +        CompletableFuture<Integer> g = f.newIncompleteFuture();
3321 +        checkIncomplete(f);
3322 +        checkIncomplete(g);
3323 +        f.complete(v1);
3324 +        checkCompletedNormally(f, v1);
3325 +        checkIncomplete(g);
3326 +        g.complete(v1);
3327 +        checkCompletedNormally(g, v1);
3328 +        assertSame(g.getClass(), CompletableFuture.class);
3329 +    }}
3330 +
3331 +    /**
3332 +     * completedStage returns a completed CompletionStage
3333 +     */
3334 +    public void testCompletedStage() {
3335 +        AtomicInteger x = new AtomicInteger(0);
3336 +        AtomicReference<Throwable> r = new AtomicReference<Throwable>();
3337 +        CompletionStage<Integer> f = CompletableFuture.completedStage(1);
3338 +        f.whenComplete((v, e) -> {if (e != null) r.set(e); else x.set(v);});
3339 +        assertEquals(x.get(), 1);
3340 +        assertNull(r.get());
3341 +    }
3342 +
3343 +    /**
3344 +     * defaultExecutor by default returns the commonPool if
3345 +     * it supports more than one thread.
3346 +     */
3347 +    public void testDefaultExecutor() {
3348 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3349 +        Executor e = f.defaultExecutor();
3350 +        Executor c = ForkJoinPool.commonPool();
3351 +        if (ForkJoinPool.getCommonPoolParallelism() > 1)
3352 +            assertSame(e, c);
3353 +        else
3354 +            assertNotSame(e, c);
3355 +    }
3356 +
3357 +    /**
3358 +     * failedFuture returns a CompletableFuture completed
3359 +     * exceptionally with the given Exception
3360 +     */
3361 +    public void testFailedFuture() {
3362 +        CFException ex = new CFException();
3363 +        CompletableFuture<Integer> f = CompletableFuture.failedFuture(ex);
3364 +        checkCompletedExceptionally(f, ex);
3365 +    }
3366 +
3367 +    /**
3368 +     * failedFuture(null) throws NPE
3369 +     */
3370 +    public void testFailedFuture_null() {
3371 +        try {
3372 +            CompletableFuture<Integer> f = CompletableFuture.failedFuture(null);
3373 +            shouldThrow();
3374 +        } catch (NullPointerException success) {}
3375 +    }
3376 +
3377 +    /**
3378 +     * copy returns a CompletableFuture that is completed normally,
3379 +     * with the same value, when source is.
3380 +     */
3381 +    public void testCopy() {
3382 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3383 +        CompletableFuture<Integer> g = f.copy();
3384 +        checkIncomplete(f);
3385 +        checkIncomplete(g);
3386 +        f.complete(1);
3387 +        checkCompletedNormally(f, 1);
3388 +        checkCompletedNormally(g, 1);
3389 +    }
3390 +
3391 +    /**
3392 +     * copy returns a CompletableFuture that is completed exceptionally
3393 +     * when source is.
3394 +     */
3395 +    public void testCopy2() {
3396 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3397 +        CompletableFuture<Integer> g = f.copy();
3398 +        checkIncomplete(f);
3399 +        checkIncomplete(g);
3400 +        CFException ex = new CFException();
3401 +        f.completeExceptionally(ex);
3402 +        checkCompletedExceptionally(f, ex);
3403 +        checkCompletedWithWrappedException(g, ex);
3404 +    }
3405 +
3406 +    /**
3407 +     * minimalCompletionStage returns a CompletableFuture that is
3408 +     * completed normally, with the same value, when source is.
3409 +     */
3410 +    public void testMinimalCompletionStage() {
3411 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3412 +        CompletionStage<Integer> g = f.minimalCompletionStage();
3413 +        AtomicInteger x = new AtomicInteger(0);
3414 +        AtomicReference<Throwable> r = new AtomicReference<Throwable>();
3415 +        checkIncomplete(f);
3416 +        g.whenComplete((v, e) -> {if (e != null) r.set(e); else x.set(v);});
3417 +        f.complete(1);
3418 +        checkCompletedNormally(f, 1);
3419 +        assertEquals(x.get(), 1);
3420 +        assertNull(r.get());
3421 +    }
3422 +
3423 +    /**
3424 +     * minimalCompletionStage returns a CompletableFuture that is
3425 +     * completed exceptionally when source is.
3426 +     */
3427 +    public void testMinimalCompletionStage2() {
3428 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3429 +        CompletionStage<Integer> g = f.minimalCompletionStage();
3430 +        AtomicInteger x = new AtomicInteger(0);
3431 +        AtomicReference<Throwable> r = new AtomicReference<Throwable>();
3432 +        g.whenComplete((v, e) -> {if (e != null) r.set(e); else x.set(v);});
3433 +        checkIncomplete(f);
3434 +        CFException ex = new CFException();
3435 +        f.completeExceptionally(ex);
3436 +        checkCompletedExceptionally(f, ex);
3437 +        assertEquals(x.get(), 0);
3438 +        assertEquals(r.get().getCause(), ex);
3439 +    }
3440 +
3441 +    /**
3442 +     * failedStage returns a CompletionStage completed
3443 +     * exceptionally with the given Exception
3444 +     */
3445 +    public void testFailedStage() {
3446 +        CFException ex = new CFException();
3447 +        CompletionStage<Integer> f = CompletableFuture.failedStage(ex);
3448 +        AtomicInteger x = new AtomicInteger(0);
3449 +        AtomicReference<Throwable> r = new AtomicReference<Throwable>();
3450 +        f.whenComplete((v, e) -> {if (e != null) r.set(e); else x.set(v);});
3451 +        assertEquals(x.get(), 0);
3452 +        assertEquals(r.get(), ex);
3453 +    }
3454 +
3455 +    /**
3456 +     * completeAsync completes with value of given supplier
3457 +     */
3458 +    public void testCompleteAsync() {
3459 +        for (Integer v1 : new Integer[] { 1, null })
3460 +    {
3461 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3462 +        f.completeAsync(() -> v1);
3463 +        f.join();
3464 +        checkCompletedNormally(f, v1);
3465 +    }}
3466 +
3467 +    /**
3468 +     * completeAsync completes exceptionally if given supplier throws
3469 +     */
3470 +    public void testCompleteAsync2() {
3471 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3472 +        CFException ex = new CFException();
3473 +        f.completeAsync(() -> {if (true) throw ex; return 1;});
3474 +        try {
3475 +            f.join();
3476 +            shouldThrow();
3477 +        } catch (CompletionException success) {}
3478 +        checkCompletedWithWrappedException(f, ex);
3479 +    }
3480 +
3481 +    /**
3482 +     * completeAsync with given executor completes with value of given supplier
3483 +     */
3484 +    public void testCompleteAsync3() {
3485 +        for (Integer v1 : new Integer[] { 1, null })
3486 +    {
3487 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3488 +        ThreadExecutor executor = new ThreadExecutor();
3489 +        f.completeAsync(() -> v1, executor);
3490 +        assertSame(v1, f.join());
3491 +        checkCompletedNormally(f, v1);
3492 +        assertEquals(1, executor.count.get());
3493 +    }}
3494 +
3495 +    /**
3496 +     * completeAsync with given executor completes exceptionally if
3497 +     * given supplier throws
3498 +     */
3499 +    public void testCompleteAsync4() {
3500 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3501 +        CFException ex = new CFException();
3502 +        ThreadExecutor executor = new ThreadExecutor();
3503 +        f.completeAsync(() -> {if (true) throw ex; return 1;}, executor);
3504 +        try {
3505 +            f.join();
3506 +            shouldThrow();
3507 +        } catch (CompletionException success) {}
3508 +        checkCompletedWithWrappedException(f, ex);
3509 +        assertEquals(1, executor.count.get());
3510 +    }
3511 +
3512 +    /**
3513 +     * orTimeout completes with TimeoutException if not complete
3514 +     */
3515 +    public void testOrTimeout_timesOut() {
3516 +        long timeoutMillis = timeoutMillis();
3517 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3518 +        long startTime = System.nanoTime();
3519 +        f.orTimeout(timeoutMillis, MILLISECONDS);
3520 +        checkCompletedWithTimeoutException(f);
3521 +        assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
3522 +    }
3523 +
3524 +    /**
3525 +     * orTimeout completes normally if completed before timeout
3526 +     */
3527 +    public void testOrTimeout_completed() {
3528 +        for (Integer v1 : new Integer[] { 1, null })
3529 +    {
3530 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3531 +        CompletableFuture<Integer> g = new CompletableFuture<>();
3532 +        long startTime = System.nanoTime();
3533 +        f.complete(v1);
3534 +        f.orTimeout(LONG_DELAY_MS, MILLISECONDS);
3535 +        g.orTimeout(LONG_DELAY_MS, MILLISECONDS);
3536 +        g.complete(v1);
3537 +        checkCompletedNormally(f, v1);
3538 +        checkCompletedNormally(g, v1);
3539 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
3540 +    }}
3541 +
3542 +    /**
3543 +     * completeOnTimeout completes with given value if not complete
3544 +     */
3545 +    public void testCompleteOnTimeout_timesOut() {
3546 +        testInParallel(() -> testCompleteOnTimeout_timesOut(42),
3547 +                       () -> testCompleteOnTimeout_timesOut(null));
3548 +    }
3549 +
3550 +    public void testCompleteOnTimeout_timesOut(Integer v) {
3551 +        long timeoutMillis = timeoutMillis();
3552 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3553 +        long startTime = System.nanoTime();
3554 +        f.completeOnTimeout(v, timeoutMillis, MILLISECONDS);
3555 +        assertSame(v, f.join());
3556 +        assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
3557 +        f.complete(99);         // should have no effect
3558 +        checkCompletedNormally(f, v);
3559 +    }
3560 +
3561 +    /**
3562 +     * completeOnTimeout has no effect if completed within timeout
3563 +     */
3564 +    public void testCompleteOnTimeout_completed() {
3565 +        for (Integer v1 : new Integer[] { 1, null })
3566 +    {
3567 +        CompletableFuture<Integer> f = new CompletableFuture<>();
3568 +        CompletableFuture<Integer> g = new CompletableFuture<>();
3569 +        long startTime = System.nanoTime();
3570 +        f.complete(v1);
3571 +        f.completeOnTimeout(-1, LONG_DELAY_MS, MILLISECONDS);
3572 +        g.completeOnTimeout(-1, LONG_DELAY_MS, MILLISECONDS);
3573 +        g.complete(v1);
3574 +        checkCompletedNormally(f, v1);
3575 +        checkCompletedNormally(g, v1);
3576 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
3577 +    }}
3578 +
3579 +    /**
3580 +     * delayedExecutor returns an executor that delays submission
3581 +     */
3582 +    public void testDelayedExecutor() {
3583 +        testInParallel(() -> testDelayedExecutor(null, null),
3584 +                       () -> testDelayedExecutor(null, 1),
3585 +                       () -> testDelayedExecutor(new ThreadExecutor(), 1),
3586 +                       () -> testDelayedExecutor(new ThreadExecutor(), 1));
3587 +    }
3588 +
3589 +    public void testDelayedExecutor(Executor executor, Integer v) throws Exception {
3590 +        long timeoutMillis = timeoutMillis();
3591 +        // Use an "unreasonably long" long timeout to catch lingering threads
3592 +        long longTimeoutMillis = 1000 * 60 * 60 * 24;
3593 +        final Executor delayer, longDelayer;
3594 +        if (executor == null) {
3595 +            delayer = CompletableFuture.delayedExecutor(timeoutMillis, MILLISECONDS);
3596 +            longDelayer = CompletableFuture.delayedExecutor(longTimeoutMillis, MILLISECONDS);
3597 +        } else {
3598 +            delayer = CompletableFuture.delayedExecutor(timeoutMillis, MILLISECONDS, executor);
3599 +            longDelayer = CompletableFuture.delayedExecutor(longTimeoutMillis, MILLISECONDS, executor);
3600 +        }
3601 +        long startTime = System.nanoTime();
3602 +        CompletableFuture<Integer> f =
3603 +            CompletableFuture.supplyAsync(() -> v, delayer);
3604 +        CompletableFuture<Integer> g =
3605 +            CompletableFuture.supplyAsync(() -> v, longDelayer);
3606 +
3607 +        assertNull(g.getNow(null));
3608 +
3609 +        assertSame(v, f.get(LONG_DELAY_MS, MILLISECONDS));
3610 +        long millisElapsed = millisElapsedSince(startTime);
3611 +        assertTrue(millisElapsed >= timeoutMillis);
3612 +        assertTrue(millisElapsed < LONG_DELAY_MS / 2);
3613 +
3614 +        checkCompletedNormally(f, v);
3615 +
3616 +        checkIncomplete(g);
3617 +        assertTrue(g.cancel(true));
3618 +    }
3619 +
3620      //--- tests of implementation details; not part of official tck ---
3621  
3622      Object resultOf(CompletableFuture<?> f) {
# Line 3339 | Line 3707 | public class CompletableFutureTest exten
3707          }
3708      }}
3709  
3710 +    /**
3711 +     * Minimal completion stages throw UOE for all non-CompletionStage methods
3712 +     */
3713 +    public void testMinimalCompletionStage_minimality() {
3714 +        if (!testImplementationDetails) return;
3715 +        Function<Method, String> toSignature =
3716 +            (method) -> method.getName() + Arrays.toString(method.getParameterTypes());
3717 +        Predicate<Method> isNotStatic =
3718 +            (method) -> (method.getModifiers() & Modifier.STATIC) == 0;
3719 +        List<Method> minimalMethods =
3720 +            Stream.of(Object.class, CompletionStage.class)
3721 +            .flatMap((klazz) -> Stream.of(klazz.getMethods()))
3722 +            .filter(isNotStatic)
3723 +            .collect(Collectors.toList());
3724 +        // Methods from CompletableFuture permitted NOT to throw UOE
3725 +        String[] signatureWhitelist = {
3726 +            "newIncompleteFuture[]",
3727 +            "defaultExecutor[]",
3728 +            "minimalCompletionStage[]",
3729 +            "copy[]",
3730 +        };
3731 +        Set<String> permittedMethodSignatures =
3732 +            Stream.concat(minimalMethods.stream().map(toSignature),
3733 +                          Stream.of(signatureWhitelist))
3734 +            .collect(Collectors.toSet());
3735 +        List<Method> allMethods = Stream.of(CompletableFuture.class.getMethods())
3736 +            .filter(isNotStatic)
3737 +            .filter((method) -> !permittedMethodSignatures.contains(toSignature.apply(method)))
3738 +            .collect(Collectors.toList());
3739 +
3740 +        CompletionStage<Integer> minimalStage =
3741 +            new CompletableFuture<Integer>().minimalCompletionStage();
3742 +
3743 +        List<Method> bugs = new ArrayList<>();
3744 +        for (Method method : allMethods) {
3745 +            Class<?>[] parameterTypes = method.getParameterTypes();
3746 +            Object[] args = new Object[parameterTypes.length];
3747 +            // Manufacture boxed primitives for primitive params
3748 +            for (int i = 0; i < args.length; i++) {
3749 +                Class<?> type = parameterTypes[i];
3750 +                if (parameterTypes[i] == boolean.class)
3751 +                    args[i] = false;
3752 +                else if (parameterTypes[i] == int.class)
3753 +                    args[i] = 0;
3754 +                else if (parameterTypes[i] == long.class)
3755 +                    args[i] = 0L;
3756 +            }
3757 +            try {
3758 +                method.invoke(minimalStage, args);
3759 +                bugs.add(method);
3760 +            }
3761 +            catch (java.lang.reflect.InvocationTargetException expected) {
3762 +                if (! (expected.getCause() instanceof UnsupportedOperationException)) {
3763 +                    bugs.add(method);
3764 +                    // expected.getCause().printStackTrace();
3765 +                }
3766 +            }
3767 +            catch (ReflectiveOperationException bad) { throw new Error(bad); }
3768 +        }
3769 +        if (!bugs.isEmpty())
3770 +            throw new Error("Methods did not throw UOE: " + bugs.toString());
3771 +    }
3772 +
3773 +    static class Monad {
3774 +        static class ZeroException extends RuntimeException {
3775 +            public ZeroException() { super("monadic zero"); }
3776 +        }
3777 +        // "return", "unit"
3778 +        static <T> CompletableFuture<T> unit(T value) {
3779 +            return completedFuture(value);
3780 +        }
3781 +        // monadic zero ?
3782 +        static <T> CompletableFuture<T> zero() {
3783 +            return failedFuture(new ZeroException());
3784 +        }
3785 +        // >=>
3786 +        static <T,U,V> Function<T, CompletableFuture<V>> compose
3787 +            (Function<T, CompletableFuture<U>> f,
3788 +             Function<U, CompletableFuture<V>> g) {
3789 +            return (x) -> f.apply(x).thenCompose(g);
3790 +        }
3791 +
3792 +        static void assertZero(CompletableFuture<?> f) {
3793 +            try {
3794 +                f.getNow(null);
3795 +                throw new AssertionFailedError("should throw");
3796 +            } catch (CompletionException success) {
3797 +                assertTrue(success.getCause() instanceof ZeroException);
3798 +            }
3799 +        }
3800 +
3801 +        static <T> void assertFutureEquals(CompletableFuture<T> f,
3802 +                                           CompletableFuture<T> g) {
3803 +            T fval = null, gval = null;
3804 +            Throwable fex = null, gex = null;
3805 +
3806 +            try { fval = f.get(); }
3807 +            catch (ExecutionException ex) { fex = ex.getCause(); }
3808 +            catch (Throwable ex) { fex = ex; }
3809 +
3810 +            try { gval = g.get(); }
3811 +            catch (ExecutionException ex) { gex = ex.getCause(); }
3812 +            catch (Throwable ex) { gex = ex; }
3813 +
3814 +            if (fex != null || gex != null)
3815 +                assertSame(fex.getClass(), gex.getClass());
3816 +            else
3817 +                assertEquals(fval, gval);
3818 +        }
3819 +
3820 +        static class PlusFuture<T> extends CompletableFuture<T> {
3821 +            AtomicReference<Throwable> firstFailure = new AtomicReference<>(null);
3822 +        }
3823 +
3824 +        // Monadic "plus"
3825 +        static <T> CompletableFuture<T> plus(CompletableFuture<? extends T> f,
3826 +                                             CompletableFuture<? extends T> g) {
3827 +            PlusFuture<T> plus = new PlusFuture<T>();
3828 +            BiConsumer<T, Throwable> action = (T result, Throwable ex) -> {
3829 +                if (ex == null) {
3830 +                    if (plus.complete(result))
3831 +                        if (plus.firstFailure.get() != null)
3832 +                            plus.firstFailure.set(null);
3833 +                }
3834 +                else if (plus.firstFailure.compareAndSet(null, ex)) {
3835 +                    if (plus.isDone())
3836 +                        plus.firstFailure.set(null);
3837 +                }
3838 +                else {
3839 +                    // first failure has precedence
3840 +                    Throwable first = plus.firstFailure.getAndSet(null);
3841 +
3842 +                    // may fail with "Self-suppression not permitted"
3843 +                    try { first.addSuppressed(ex); }
3844 +                    catch (Exception ignored) {}
3845 +
3846 +                    plus.completeExceptionally(first);
3847 +                }
3848 +            };
3849 +            f.whenComplete(action);
3850 +            g.whenComplete(action);
3851 +            return plus;
3852 +        }
3853 +    }
3854 +
3855 +    /**
3856 +     * CompletableFuture is an additive monad - sort of.
3857 +     * https://en.wikipedia.org/wiki/Monad_(functional_programming)#Additive_monads
3858 +     */
3859 +    public void testAdditiveMonad() throws Throwable {
3860 +        Function<Long, CompletableFuture<Long>> unit = Monad::unit;
3861 +        CompletableFuture<Long> zero = Monad.zero();
3862 +
3863 +        // Some mutually non-commutative functions
3864 +        Function<Long, CompletableFuture<Long>> triple
3865 +            = (x) -> Monad.unit(3 * x);
3866 +        Function<Long, CompletableFuture<Long>> inc
3867 +            = (x) -> Monad.unit(x + 1);
3868 +
3869 +        // unit is a right identity: m >>= unit === m
3870 +        Monad.assertFutureEquals(inc.apply(5L).thenCompose(unit),
3871 +                                 inc.apply(5L));
3872 +        // unit is a left identity: (unit x) >>= f === f x
3873 +        Monad.assertFutureEquals(unit.apply(5L).thenCompose(inc),
3874 +                                 inc.apply(5L));
3875 +
3876 +        // associativity: (m >>= f) >>= g === m >>= ( \x -> (f x >>= g) )
3877 +        Monad.assertFutureEquals(
3878 +            unit.apply(5L).thenCompose(inc).thenCompose(triple),
3879 +            unit.apply(5L).thenCompose((x) -> inc.apply(x).thenCompose(triple)));
3880 +
3881 +        // The case for CompletableFuture as an additive monad is weaker...
3882 +
3883 +        // zero is a monadic zero
3884 +        Monad.assertZero(zero);
3885 +
3886 +        // left zero: zero >>= f === zero
3887 +        Monad.assertZero(zero.thenCompose(inc));
3888 +        // right zero: f >>= (\x -> zero) === zero
3889 +        Monad.assertZero(inc.apply(5L).thenCompose((x) -> zero));
3890 +
3891 +        // f plus zero === f
3892 +        Monad.assertFutureEquals(Monad.unit(5L),
3893 +                                 Monad.plus(Monad.unit(5L), zero));
3894 +        // zero plus f === f
3895 +        Monad.assertFutureEquals(Monad.unit(5L),
3896 +                                 Monad.plus(zero, Monad.unit(5L)));
3897 +        // zero plus zero === zero
3898 +        Monad.assertZero(Monad.plus(zero, zero));
3899 +        {
3900 +            CompletableFuture<Long> f = Monad.plus(Monad.unit(5L),
3901 +                                                   Monad.unit(8L));
3902 +            // non-determinism
3903 +            assertTrue(f.get() == 5L || f.get() == 8L);
3904 +        }
3905 +
3906 +        CompletableFuture<Long> godot = new CompletableFuture<>();
3907 +        // f plus godot === f (doesn't wait for godot)
3908 +        Monad.assertFutureEquals(Monad.unit(5L),
3909 +                                 Monad.plus(Monad.unit(5L), godot));
3910 +        // godot plus f === f (doesn't wait for godot)
3911 +        Monad.assertFutureEquals(Monad.unit(5L),
3912 +                                 Monad.plus(godot, Monad.unit(5L)));
3913 +    }
3914 +
3915 + //     static <U> U join(CompletionStage<U> stage) {
3916 + //         CompletableFuture<U> f = new CompletableFuture<>();
3917 + //         stage.whenComplete((v, ex) -> {
3918 + //             if (ex != null) f.completeExceptionally(ex); else f.complete(v);
3919 + //         });
3920 + //         return f.join();
3921 + //     }
3922 +
3923 + //     static <U> boolean isDone(CompletionStage<U> stage) {
3924 + //         CompletableFuture<U> f = new CompletableFuture<>();
3925 + //         stage.whenComplete((v, ex) -> {
3926 + //             if (ex != null) f.completeExceptionally(ex); else f.complete(v);
3927 + //         });
3928 + //         return f.isDone();
3929 + //     }
3930 +
3931 + //     static <U> U join2(CompletionStage<U> stage) {
3932 + //         return stage.toCompletableFuture().copy().join();
3933 + //     }
3934 +
3935 + //     static <U> boolean isDone2(CompletionStage<U> stage) {
3936 + //         return stage.toCompletableFuture().copy().isDone();
3937 + //     }
3938 +
3939   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines