--- a/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java Thu Sep 27 17:13:07 2018 +0200
+++ b/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java Fri Sep 28 08:45:46 2018 -0700
@@ -957,17 +957,17 @@
@SuppressWarnings("serial")
static final class UniExceptionally<T> extends UniCompletion<T,T> {
Function<? super Throwable, ? extends T> fn;
- UniExceptionally(CompletableFuture<T> dep, CompletableFuture<T> src,
+ UniExceptionally(Executor executor,
+ CompletableFuture<T> dep, CompletableFuture<T> src,
Function<? super Throwable, ? extends T> fn) {
- super(null, dep, src); this.fn = fn;
+ super(executor, dep, src); this.fn = fn;
}
- final CompletableFuture<T> tryFire(int mode) { // never ASYNC
- // assert mode != ASYNC;
+ final CompletableFuture<T> tryFire(int mode) {
CompletableFuture<T> d; CompletableFuture<T> a;
Object r; Function<? super Throwable, ? extends T> f;
if ((d = dep) == null || (f = fn) == null
|| (a = src) == null || (r = a.result) == null
- || !d.uniExceptionally(r, f, this))
+ || !d.uniExceptionally(r, f, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
@@ -980,11 +980,11 @@
Throwable x;
if (result == null) {
try {
- if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) {
- if (c != null && !c.claim())
- return false;
+ if (c != null && !c.claim())
+ return false;
+ if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
completeValue(f.apply(x));
- } else
+ else
internalComplete(r);
} catch (Throwable ex) {
completeThrowable(ex);
@@ -994,14 +994,88 @@
}
private CompletableFuture<T> uniExceptionallyStage(
- Function<Throwable, ? extends T> f) {
+ Executor e, Function<Throwable, ? extends T> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<T> d = newIncompleteFuture();
Object r;
if ((r = result) == null)
- unipush(new UniExceptionally<T>(d, this, f));
+ unipush(new UniExceptionally<T>(e, d, this, f));
+ else if (e == null)
+ d.uniExceptionally(r, f, null);
+ else {
+ try {
+ e.execute(new UniExceptionally<T>(null, d, this, f));
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
+ }
+ return d;
+ }
+
+ @SuppressWarnings("serial")
+ static final class UniComposeExceptionally<T> extends UniCompletion<T,T> {
+ Function<Throwable, ? extends CompletionStage<T>> fn;
+ UniComposeExceptionally(Executor executor, CompletableFuture<T> dep,
+ CompletableFuture<T> src,
+ Function<Throwable, ? extends CompletionStage<T>> fn) {
+ super(executor, dep, src); this.fn = fn;
+ }
+ final CompletableFuture<T> tryFire(int mode) {
+ CompletableFuture<T> d; CompletableFuture<T> a;
+ Function<Throwable, ? extends CompletionStage<T>> f;
+ Object r; Throwable x;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null)
+ return null;
+ if (d.result == null) {
+ if ((r instanceof AltResult) &&
+ (x = ((AltResult)r).ex) != null) {
+ try {
+ if (mode <= 0 && !claim())
+ return null;
+ CompletableFuture<T> g = f.apply(x).toCompletableFuture();
+ if ((r = g.result) != null)
+ d.completeRelay(r);
+ else {
+ g.unipush(new UniRelay<T,T>(d, g));
+ if (d.result == null)
+ return null;
+ }
+ } catch (Throwable ex) {
+ d.completeThrowable(ex);
+ }
+ }
+ else
+ d.internalComplete(r);
+ }
+ dep = null; src = null; fn = null;
+ return d.postFire(a, mode);
+ }
+ }
+
+ private CompletableFuture<T> uniComposeExceptionallyStage(
+ Executor e, Function<Throwable, ? extends CompletionStage<T>> f) {
+ if (f == null) throw new NullPointerException();
+ CompletableFuture<T> d = newIncompleteFuture();
+ Object r, s; Throwable x;
+ if ((r = result) == null)
+ unipush(new UniComposeExceptionally<T>(e, d, this, f));
+ else if (!(r instanceof AltResult) || (x = ((AltResult)r).ex) == null)
+ d.internalComplete(r);
else
- d.uniExceptionally(r, f, null);
+ try {
+ if (e != null)
+ e.execute(new UniComposeExceptionally<T>(null, d, this, f));
+ else {
+ CompletableFuture<T> g = f.apply(x).toCompletableFuture();
+ if ((s = g.result) != null)
+ d.result = encodeRelay(s);
+ else
+ g.unipush(new UniRelay<T,T>(d, g));
+ }
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
return d;
}
@@ -1093,7 +1167,7 @@
Object r, s; Throwable x;
if ((r = result) == null)
unipush(new UniCompose<T,V>(e, d, this, f));
- else if (e == null) {
+ else {
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
d.result = encodeThrowable(x, r);
@@ -1102,23 +1176,20 @@
r = null;
}
try {
- @SuppressWarnings("unchecked") T t = (T) r;
- CompletableFuture<V> g = f.apply(t).toCompletableFuture();
- if ((s = g.result) != null)
- d.result = encodeRelay(s);
+ if (e != null)
+ e.execute(new UniCompose<T,V>(null, d, this, f));
else {
- g.unipush(new UniRelay<V,V>(d, g));
+ @SuppressWarnings("unchecked") T t = (T) r;
+ CompletableFuture<V> g = f.apply(t).toCompletableFuture();
+ if ((s = g.result) != null)
+ d.result = encodeRelay(s);
+ else
+ g.unipush(new UniRelay<V,V>(d, g));
}
} catch (Throwable ex) {
d.result = encodeThrowable(ex);
}
}
- else
- try {
- e.execute(new UniCompose<T,V>(null, d, this, f));
- } catch (Throwable ex) {
- d.result = encodeThrowable(ex);
- }
return d;
}
@@ -1898,7 +1969,7 @@
* Creates a new complete CompletableFuture with given encoded result.
*/
CompletableFuture(Object r) {
- this.result = r;
+ RESULT.setRelease(this, r);
}
/**
@@ -2285,28 +2356,36 @@
return this;
}
- // not in interface CompletionStage
-
- /**
- * Returns a new CompletableFuture that is completed when this
- * CompletableFuture completes, with the result of the given
- * function of the exception triggering this CompletableFuture's
- * completion when it completes exceptionally; otherwise, if this
- * CompletableFuture completes normally, then the returned
- * CompletableFuture also completes normally with the same value.
- * Note: More flexible versions of this functionality are
- * available using methods {@code whenComplete} and {@code handle}.
- *
- * @param fn the function to use to compute the value of the
- * returned CompletableFuture if this CompletableFuture completed
- * exceptionally
- * @return the new CompletableFuture
- */
public CompletableFuture<T> exceptionally(
Function<Throwable, ? extends T> fn) {
- return uniExceptionallyStage(fn);
+ return uniExceptionallyStage(null, fn);
+ }
+
+ public CompletableFuture<T> exceptionallyAsync(
+ Function<Throwable, ? extends T> fn) {
+ return uniExceptionallyStage(defaultExecutor(), fn);
+ }
+
+ public CompletableFuture<T> exceptionallyAsync(
+ Function<Throwable, ? extends T> fn, Executor executor) {
+ return uniExceptionallyStage(screenExecutor(executor), fn);
}
+ public CompletableFuture<T> exceptionallyCompose(
+ Function<Throwable, ? extends CompletionStage<T>> fn) {
+ return uniComposeExceptionallyStage(null, fn);
+ }
+
+ public CompletableFuture<T> exceptionallyComposeAsync(
+ Function<Throwable, ? extends CompletionStage<T>> fn) {
+ return uniComposeExceptionallyStage(defaultExecutor(), fn);
+ }
+
+ public CompletableFuture<T> exceptionallyComposeAsync(
+ Function<Throwable, ? extends CompletionStage<T>> fn,
+ Executor executor) {
+ return uniComposeExceptionallyStage(screenExecutor(executor), fn);
+ }
/* ------------- Arbitrary-arity constructions -------------- */
--- a/src/java.base/share/classes/java/util/concurrent/CompletionStage.java Thu Sep 27 17:13:07 2018 +0200
+++ b/src/java.base/share/classes/java/util/concurrent/CompletionStage.java Fri Sep 28 08:45:46 2018 -0700
@@ -851,6 +851,130 @@
(Function<Throwable, ? extends T> fn);
/**
+ * Returns a new CompletionStage that, when this stage completes
+ * exceptionally, is executed with this stage's exception as the
+ * argument to the supplied function, using this stage's default
+ * asynchronous execution facility. Otherwise, if this stage
+ * completes normally, then the returned stage also completes
+ * normally with the same value.
+ *
+ * @implSpec The default implementation invokes {@link #handle},
+ * relaying to {@link #handleAsync} on exception, then {@link
+ * #thenCompose} for result.
+ *
+ * @param fn the function to use to compute the value of the
+ * returned CompletionStage if this CompletionStage completed
+ * exceptionally
+ * @return the new CompletionStage
+ * @since 12
+ */
+ public default CompletionStage<T> exceptionallyAsync
+ (Function<Throwable, ? extends T> fn) {
+ return handle((r, ex) -> (ex == null)
+ ? this
+ : this.<T>handleAsync((r1, ex1) -> fn.apply(ex1)))
+ .thenCompose(Function.identity());
+ }
+
+ /**
+ * Returns a new CompletionStage that, when this stage completes
+ * exceptionally, is executed with this stage's exception as the
+ * argument to the supplied function, using the supplied Executor.
+ * Otherwise, if this stage completes normally, then the returned
+ * stage also completes normally with the same value.
+ *
+ * @implSpec The default implementation invokes {@link #handle},
+ * relaying to {@link #handleAsync} on exception, then {@link
+ * #thenCompose} for result.
+ *
+ * @param fn the function to use to compute the value of the
+ * returned CompletionStage if this CompletionStage completed
+ * exceptionally
+ * @param executor the executor to use for asynchronous execution
+ * @return the new CompletionStage
+ * @since 12
+ */
+ public default CompletionStage<T> exceptionallyAsync
+ (Function<Throwable, ? extends T> fn, Executor executor) {
+ return handle((r, ex) -> (ex == null)
+ ? this
+ : this.<T>handleAsync((r1, ex1) -> fn.apply(ex1), executor))
+ .thenCompose(Function.identity());
+ }
+
+ /**
+ * Returns a new CompletionStage that, when this stage completes
+ * exceptionally, is composed using the results of the supplied
+ * function applied to this stage's exception.
+ *
+ * @implSpec The default implementation invokes {@link #handle},
+ * invoking the given function on exception, then {@link
+ * #thenCompose} for result.
+ *
+ * @param fn the function to use to compute the returned
+ * CompletionStage if this CompletionStage completed exceptionally
+ * @return the new CompletionStage
+ * @since 12
+ */
+ public default CompletionStage<T> exceptionallyCompose
+ (Function<Throwable, ? extends CompletionStage<T>> fn) {
+ return handle((r, ex) -> (ex == null)
+ ? this
+ : fn.apply(ex))
+ .thenCompose(Function.identity());
+ }
+
+ /**
+ * Returns a new CompletionStage that, when this stage completes
+ * exceptionally, is composed using the results of the supplied
+ * function applied to this stage's exception, using this stage's
+ * default asynchronous execution facility.
+ *
+ * @implSpec The default implementation invokes {@link #handle},
+ * relaying to {@link #handleAsync} on exception, then {@link
+ * #thenCompose} for result.
+ *
+ * @param fn the function to use to compute the returned
+ * CompletionStage if this CompletionStage completed exceptionally
+ * @return the new CompletionStage
+ * @since 12
+ */
+ public default CompletionStage<T> exceptionallyComposeAsync
+ (Function<Throwable, ? extends CompletionStage<T>> fn) {
+ return handle((r, ex) -> (ex == null)
+ ? this
+ : this.handleAsync((r1, ex1) -> fn.apply(ex1))
+ .thenCompose(Function.identity()))
+ .thenCompose(Function.identity());
+ }
+
+ /**
+ * Returns a new CompletionStage that, when this stage completes
+ * exceptionally, is composed using the results of the supplied
+ * function applied to this stage's exception, using the
+ * supplied Executor.
+ *
+ * @implSpec The default implementation invokes {@link #handle},
+ * relaying to {@link #handleAsync} on exception, then {@link
+ * #thenCompose} for result.
+ *
+ * @param fn the function to use to compute the returned
+ * CompletionStage if this CompletionStage completed exceptionally
+ * @param executor the executor to use for asynchronous execution
+ * @return the new CompletionStage
+ * @since 12
+ */
+ public default CompletionStage<T> exceptionallyComposeAsync
+ (Function<Throwable, ? extends CompletionStage<T>> fn,
+ Executor executor) {
+ return handle((r, ex) -> (ex == null)
+ ? this
+ : this.handleAsync((r1, ex1) -> fn.apply(ex1), executor)
+ .thenCompose(Function.identity()))
+ .thenCompose(Function.identity());
+ }
+
+ /**
* Returns a {@link CompletableFuture} maintaining the same
* completion properties as this stage. If this stage is already a
* CompletableFuture, this method may return this stage itself.
--- a/test/jdk/java/util/concurrent/tck/CompletableFutureTest.java Thu Sep 27 17:13:07 2018 +0200
+++ b/test/jdk/java/util/concurrent/tck/CompletableFutureTest.java Fri Sep 28 08:45:46 2018 -0700
@@ -86,9 +86,13 @@
assertFalse(f.isDone());
assertFalse(f.isCancelled());
assertTrue(f.toString().matches(".*\\[.*Not completed.*\\]"));
+
+ Object result = null;
try {
- assertNull(f.getNow(null));
+ result = f.getNow(null);
} catch (Throwable fail) { threadUnexpectedException(fail); }
+ assertNull(result);
+
try {
f.get(randomExpiredTimeout(), randomTimeUnit());
shouldThrow();
@@ -97,14 +101,18 @@
catch (Throwable fail) { threadUnexpectedException(fail); }
}
- <T> void checkCompletedNormally(CompletableFuture<T> f, T value) {
- checkTimedGet(f, value);
-
+ <T> void checkCompletedNormally(CompletableFuture<T> f, T expectedValue) {
+ checkTimedGet(f, expectedValue);
+
+ assertEquals(expectedValue, f.join());
+ assertEquals(expectedValue, f.getNow(null));
+
+ T result = null;
try {
- assertEquals(value, f.join());
- assertEquals(value, f.getNow(null));
- assertEquals(value, f.get());
+ result = f.get();
} catch (Throwable fail) { threadUnexpectedException(fail); }
+ assertEquals(expectedValue, result);
+
assertTrue(f.isDone());
assertFalse(f.isCancelled());
assertFalse(f.isCompletedExceptionally());
@@ -570,9 +578,28 @@
public CompletableFuture<Integer> apply(Integer x) {
invoked();
value = x;
- CompletableFuture<Integer> f = new CompletableFuture<>();
- assertTrue(f.complete(inc(x)));
- return f;
+ return CompletableFuture.completedFuture(inc(x));
+ }
+ }
+
+ static class FailingExceptionalCompletableFutureFunction extends CheckedAction
+ implements Function<Throwable, CompletableFuture<Integer>>
+ {
+ final CFException ex;
+ FailingExceptionalCompletableFutureFunction(ExecutionMode m) { super(m); ex = new CFException(); }
+ public CompletableFuture<Integer> apply(Throwable x) {
+ invoked();
+ throw ex;
+ }
+ }
+
+ static class ExceptionalCompletableFutureFunction extends CheckedAction
+ implements Function<Throwable, CompletionStage<Integer>> {
+ final Integer value = 3;
+ ExceptionalCompletableFutureFunction(ExecutionMode m) { super(m); }
+ public CompletionStage<Integer> apply(Throwable x) {
+ invoked();
+ return CompletableFuture.completedFuture(value);
}
}
@@ -691,8 +718,16 @@
Function<? super T,U> a) {
return f.applyToEither(g, a);
}
+ public <T> CompletableFuture<T> exceptionally
+ (CompletableFuture<T> f,
+ Function<Throwable, ? extends T> fn) {
+ return f.exceptionally(fn);
+ }
+ public <T> CompletableFuture<T> exceptionallyCompose
+ (CompletableFuture<T> f, Function<Throwable, ? extends CompletionStage<T>> fn) {
+ return f.exceptionallyCompose(fn);
+ }
},
-
ASYNC {
public void checkExecutionMode() {
assertEquals(defaultExecutorIsCommonPool,
@@ -765,6 +800,17 @@
Function<? super T,U> a) {
return f.applyToEitherAsync(g, a);
}
+ public <T> CompletableFuture<T> exceptionally
+ (CompletableFuture<T> f,
+ Function<Throwable, ? extends T> fn) {
+ return f.exceptionallyAsync(fn);
+ }
+
+ public <T> CompletableFuture<T> exceptionallyCompose
+ (CompletableFuture<T> f, Function<Throwable, ? extends CompletionStage<T>> fn) {
+ return f.exceptionallyComposeAsync(fn);
+ }
+
},
EXECUTOR {
@@ -838,6 +884,16 @@
Function<? super T,U> a) {
return f.applyToEitherAsync(g, a, new ThreadExecutor());
}
+ public <T> CompletableFuture<T> exceptionally
+ (CompletableFuture<T> f,
+ Function<Throwable, ? extends T> fn) {
+ return f.exceptionallyAsync(fn, new ThreadExecutor());
+ }
+ public <T> CompletableFuture<T> exceptionallyCompose
+ (CompletableFuture<T> f, Function<Throwable, ? extends CompletionStage<T>> fn) {
+ return f.exceptionallyComposeAsync(fn, new ThreadExecutor());
+ }
+
};
public abstract void checkExecutionMode();
@@ -880,6 +936,12 @@
(CompletableFuture<T> f,
CompletionStage<? extends T> g,
Function<? super T,U> a);
+ public abstract <T> CompletableFuture<T> exceptionally
+ (CompletableFuture<T> f,
+ Function<Throwable, ? extends T> fn);
+ public abstract <T> CompletableFuture<T> exceptionallyCompose
+ (CompletableFuture<T> f,
+ Function<Throwable, ? extends CompletionStage<T>> fn);
}
/**
@@ -887,15 +949,14 @@
* normally, and source result is propagated
*/
public void testExceptionally_normalCompletion() {
+ for (ExecutionMode m : ExecutionMode.values())
for (boolean createIncomplete : new boolean[] { true, false })
for (Integer v1 : new Integer[] { 1, null })
{
- final AtomicInteger a = new AtomicInteger(0);
final CompletableFuture<Integer> f = new CompletableFuture<>();
if (!createIncomplete) assertTrue(f.complete(v1));
- final CompletableFuture<Integer> g = f.exceptionally
- ((Throwable t) -> {
- a.getAndIncrement();
+ final CompletableFuture<Integer> g = m.exceptionally
+ (f, (Throwable t) -> {
threadFail("should not be called");
return null; // unreached
});
@@ -903,7 +964,6 @@
checkCompletedNormally(g, v1);
checkCompletedNormally(f, v1);
- assertEquals(0, a.get());
}}
/**
@@ -911,6 +971,7 @@
* exception
*/
public void testExceptionally_exceptionalCompletion() {
+ for (ExecutionMode m : ExecutionMode.values())
for (boolean createIncomplete : new boolean[] { true, false })
for (Integer v1 : new Integer[] { 1, null })
{
@@ -918,9 +979,9 @@
final CFException ex = new CFException();
final CompletableFuture<Integer> f = new CompletableFuture<>();
if (!createIncomplete) f.completeExceptionally(ex);
- final CompletableFuture<Integer> g = f.exceptionally
- ((Throwable t) -> {
- ExecutionMode.SYNC.checkExecutionMode();
+ final CompletableFuture<Integer> g = m.exceptionally
+ (f, (Throwable t) -> {
+ m.checkExecutionMode();
threadAssertSame(t, ex);
a.getAndIncrement();
return v1;
@@ -936,6 +997,7 @@
* exceptionally with that exception
*/
public void testExceptionally_exceptionalCompletionActionFailed() {
+ for (ExecutionMode m : ExecutionMode.values())
for (boolean createIncomplete : new boolean[] { true, false })
{
final AtomicInteger a = new AtomicInteger(0);
@@ -943,9 +1005,9 @@
final CFException ex2 = new CFException();
final CompletableFuture<Integer> f = new CompletableFuture<>();
if (!createIncomplete) f.completeExceptionally(ex1);
- final CompletableFuture<Integer> g = f.exceptionally
- ((Throwable t) -> {
- ExecutionMode.SYNC.checkExecutionMode();
+ final CompletableFuture<Integer> g = m.exceptionally
+ (f, (Throwable t) -> {
+ m.checkExecutionMode();
threadAssertSame(t, ex1);
a.getAndIncrement();
throw ex2;
@@ -3116,6 +3178,121 @@
checkCompletedNormally(f, v1);
}}
+ /**
+ * exceptionallyCompose result completes normally after normal
+ * completion of source
+ */
+ public void testExceptionallyCompose_normalCompletion() {
+ for (ExecutionMode m : ExecutionMode.values())
+ for (boolean createIncomplete : new boolean[] { true, false })
+ for (Integer v1 : new Integer[] { 1, null })
+ {
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final ExceptionalCompletableFutureFunction r =
+ new ExceptionalCompletableFutureFunction(m);
+ if (!createIncomplete) assertTrue(f.complete(v1));
+ final CompletableFuture<Integer> g = m.exceptionallyCompose(f, r);
+ if (createIncomplete) assertTrue(f.complete(v1));
+
+ checkCompletedNormally(f, v1);
+ checkCompletedNormally(g, v1);
+ r.assertNotInvoked();
+ }}
+
+ /**
+ * exceptionallyCompose result completes normally after exceptional
+ * completion of source
+ */
+ public void testExceptionallyCompose_exceptionalCompletion() {
+ for (ExecutionMode m : ExecutionMode.values())
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final CFException ex = new CFException();
+ final ExceptionalCompletableFutureFunction r =
+ new ExceptionalCompletableFutureFunction(m);
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletableFuture<Integer> g = m.exceptionallyCompose(f, r);
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedExceptionally(f, ex);
+ checkCompletedNormally(g, r.value);
+ r.assertInvoked();
+ }}
+
+ /**
+ * exceptionallyCompose completes exceptionally on exception if action does
+ */
+ public void testExceptionallyCompose_actionFailed() {
+ for (ExecutionMode m : ExecutionMode.values())
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final CFException ex = new CFException();
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final FailingExceptionalCompletableFutureFunction r
+ = new FailingExceptionalCompletableFutureFunction(m);
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletableFuture<Integer> g = m.exceptionallyCompose(f, r);
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedExceptionally(f, ex);
+ checkCompletedWithWrappedException(g, r.ex);
+ r.assertInvoked();
+ }}
+
+ /**
+ * exceptionallyCompose result completes exceptionally if the
+ * result of the action does
+ */
+ public void testExceptionallyCompose_actionReturnsFailingFuture() {
+ for (ExecutionMode m : ExecutionMode.values())
+ for (int order = 0; order < 6; order++)
+ {
+ final CFException ex0 = new CFException();
+ final CFException ex = new CFException();
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final CompletableFuture<Integer> g = new CompletableFuture<>();
+ final CompletableFuture<Integer> h;
+ // Test all permutations of orders
+ switch (order) {
+ case 0:
+ assertTrue(f.completeExceptionally(ex0));
+ assertTrue(g.completeExceptionally(ex));
+ h = m.exceptionallyCompose(f, (x -> g));
+ break;
+ case 1:
+ assertTrue(f.completeExceptionally(ex0));
+ h = m.exceptionallyCompose(f, (x -> g));
+ assertTrue(g.completeExceptionally(ex));
+ break;
+ case 2:
+ assertTrue(g.completeExceptionally(ex));
+ assertTrue(f.completeExceptionally(ex0));
+ h = m.exceptionallyCompose(f, (x -> g));
+ break;
+ case 3:
+ assertTrue(g.completeExceptionally(ex));
+ h = m.exceptionallyCompose(f, (x -> g));
+ assertTrue(f.completeExceptionally(ex0));
+ break;
+ case 4:
+ h = m.exceptionallyCompose(f, (x -> g));
+ assertTrue(f.completeExceptionally(ex0));
+ assertTrue(g.completeExceptionally(ex));
+ break;
+ case 5:
+ h = m.exceptionallyCompose(f, (x -> g));
+ assertTrue(f.completeExceptionally(ex0));
+ assertTrue(g.completeExceptionally(ex));
+ break;
+ default: throw new AssertionError();
+ }
+
+ checkCompletedExceptionally(g, ex);
+ checkCompletedWithWrappedException(h, ex);
+ checkCompletedExceptionally(f, ex0);
+ }}
+
// other static methods
/**
@@ -4527,4 +4704,429 @@
// return stage.toCompletableFuture().copy().isDone();
// }
+ // For testing default implementations
+ // Only non-default interface methods defined.
+ static final class DelegatedCompletionStage<T> implements CompletionStage<T> {
+ final CompletableFuture<T> cf;
+ DelegatedCompletionStage(CompletableFuture<T> cf) { this.cf = cf; }
+ public CompletableFuture<T> toCompletableFuture() {
+ return cf; }
+ public CompletionStage<Void> thenRun
+ (Runnable action) {
+ return cf.thenRun(action); }
+ public CompletionStage<Void> thenRunAsync
+ (Runnable action) {
+ return cf.thenRunAsync(action); }
+ public CompletionStage<Void> thenRunAsync
+ (Runnable action,
+ Executor executor) {
+ return cf.thenRunAsync(action, executor); }
+ public CompletionStage<Void> thenAccept
+ (Consumer<? super T> action) {
+ return cf.thenAccept(action); }
+ public CompletionStage<Void> thenAcceptAsync
+ (Consumer<? super T> action) {
+ return cf.thenAcceptAsync(action); }
+ public CompletionStage<Void> thenAcceptAsync
+ (Consumer<? super T> action,
+ Executor executor) {
+ return cf.thenAcceptAsync(action, executor); }
+ public <U> CompletionStage<U> thenApply
+ (Function<? super T,? extends U> a) {
+ return cf.thenApply(a); }
+ public <U> CompletionStage<U> thenApplyAsync
+ (Function<? super T,? extends U> fn) {
+ return cf.thenApplyAsync(fn); }
+ public <U> CompletionStage<U> thenApplyAsync
+ (Function<? super T,? extends U> fn,
+ Executor executor) {
+ return cf.thenApplyAsync(fn, executor); }
+ public <U,V> CompletionStage<V> thenCombine
+ (CompletionStage<? extends U> other,
+ BiFunction<? super T,? super U,? extends V> fn) {
+ return cf.thenCombine(other, fn); }
+ public <U,V> CompletionStage<V> thenCombineAsync
+ (CompletionStage<? extends U> other,
+ BiFunction<? super T,? super U,? extends V> fn) {
+ return cf.thenCombineAsync(other, fn); }
+ public <U,V> CompletionStage<V> thenCombineAsync
+ (CompletionStage<? extends U> other,
+ BiFunction<? super T,? super U,? extends V> fn,
+ Executor executor) {
+ return cf.thenCombineAsync(other, fn, executor); }
+ public <U> CompletionStage<Void> thenAcceptBoth
+ (CompletionStage<? extends U> other,
+ BiConsumer<? super T, ? super U> action) {
+ return cf.thenAcceptBoth(other, action); }
+ public <U> CompletionStage<Void> thenAcceptBothAsync
+ (CompletionStage<? extends U> other,
+ BiConsumer<? super T, ? super U> action) {
+ return cf.thenAcceptBothAsync(other, action); }
+ public <U> CompletionStage<Void> thenAcceptBothAsync
+ (CompletionStage<? extends U> other,
+ BiConsumer<? super T, ? super U> action,
+ Executor executor) {
+ return cf.thenAcceptBothAsync(other, action, executor); }
+ public CompletionStage<Void> runAfterBoth
+ (CompletionStage<?> other,
+ Runnable action) {
+ return cf.runAfterBoth(other, action); }
+ public CompletionStage<Void> runAfterBothAsync
+ (CompletionStage<?> other,
+ Runnable action) {
+ return cf.runAfterBothAsync(other, action); }
+ public CompletionStage<Void> runAfterBothAsync
+ (CompletionStage<?> other,
+ Runnable action,
+ Executor executor) {
+ return cf.runAfterBothAsync(other, action, executor); }
+ public <U> CompletionStage<U> applyToEither
+ (CompletionStage<? extends T> other,
+ Function<? super T, U> fn) {
+ return cf.applyToEither(other, fn); }
+ public <U> CompletionStage<U> applyToEitherAsync
+ (CompletionStage<? extends T> other,
+ Function<? super T, U> fn) {
+ return cf.applyToEitherAsync(other, fn); }
+ public <U> CompletionStage<U> applyToEitherAsync
+ (CompletionStage<? extends T> other,
+ Function<? super T, U> fn,
+ Executor executor) {
+ return cf.applyToEitherAsync(other, fn, executor); }
+ public CompletionStage<Void> acceptEither
+ (CompletionStage<? extends T> other,
+ Consumer<? super T> action) {
+ return cf.acceptEither(other, action); }
+ public CompletionStage<Void> acceptEitherAsync
+ (CompletionStage<? extends T> other,
+ Consumer<? super T> action) {
+ return cf.acceptEitherAsync(other, action); }
+ public CompletionStage<Void> acceptEitherAsync
+ (CompletionStage<? extends T> other,
+ Consumer<? super T> action,
+ Executor executor) {
+ return cf.acceptEitherAsync(other, action, executor); }
+ public CompletionStage<Void> runAfterEither
+ (CompletionStage<?> other,
+ Runnable action) {
+ return cf.runAfterEither(other, action); }
+ public CompletionStage<Void> runAfterEitherAsync
+ (CompletionStage<?> other,
+ Runnable action) {
+ return cf.runAfterEitherAsync(other, action); }
+ public CompletionStage<Void> runAfterEitherAsync
+ (CompletionStage<?> other,
+ Runnable action,
+ Executor executor) {
+ return cf.runAfterEitherAsync(other, action, executor); }
+ public <U> CompletionStage<U> thenCompose
+ (Function<? super T, ? extends CompletionStage<U>> fn) {
+ return cf.thenCompose(fn); }
+ public <U> CompletionStage<U> thenComposeAsync
+ (Function<? super T, ? extends CompletionStage<U>> fn) {
+ return cf.thenComposeAsync(fn); }
+ public <U> CompletionStage<U> thenComposeAsync
+ (Function<? super T, ? extends CompletionStage<U>> fn,
+ Executor executor) {
+ return cf.thenComposeAsync(fn, executor); }
+ public <U> CompletionStage<U> handle
+ (BiFunction<? super T, Throwable, ? extends U> fn) {
+ return cf.handle(fn); }
+ public <U> CompletionStage<U> handleAsync
+ (BiFunction<? super T, Throwable, ? extends U> fn) {
+ return cf.handleAsync(fn); }
+ public <U> CompletionStage<U> handleAsync
+ (BiFunction<? super T, Throwable, ? extends U> fn,
+ Executor executor) {
+ return cf.handleAsync(fn, executor); }
+ public CompletionStage<T> whenComplete
+ (BiConsumer<? super T, ? super Throwable> action) {
+ return cf.whenComplete(action); }
+ public CompletionStage<T> whenCompleteAsync
+ (BiConsumer<? super T, ? super Throwable> action) {
+ return cf.whenCompleteAsync(action); }
+ public CompletionStage<T> whenCompleteAsync
+ (BiConsumer<? super T, ? super Throwable> action,
+ Executor executor) {
+ return cf.whenCompleteAsync(action, executor); }
+ public CompletionStage<T> exceptionally
+ (Function<Throwable, ? extends T> fn) {
+ return cf.exceptionally(fn); }
+ }
+
+ /**
+ * default-implemented exceptionallyAsync action is not invoked when
+ * source completes normally, and source result is propagated
+ */
+ public void testDefaultExceptionallyAsync_normalCompletion() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ for (Integer v1 : new Integer[] { 1, null })
+ {
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) assertTrue(f.complete(v1));
+ final CompletionStage<Integer> g = d.exceptionallyAsync
+ ((Throwable t) -> {
+ threadFail("should not be called");
+ return null; // unreached
+ });
+ if (createIncomplete) assertTrue(f.complete(v1));
+
+ checkCompletedNormally(g.toCompletableFuture(), v1);
+ }}
+
+ /**
+ * default-implemented exceptionallyAsync action completes with
+ * function value on source exception
+ */
+ public void testDefaultExceptionallyAsync_exceptionalCompletion() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ for (Integer v1 : new Integer[] { 1, null })
+ {
+ final AtomicInteger a = new AtomicInteger(0);
+ final CFException ex = new CFException();
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletionStage<Integer> g = d.exceptionallyAsync
+ ((Throwable t) -> {
+ threadAssertSame(t, ex);
+ a.getAndIncrement();
+ return v1;
+ });
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedNormally(g.toCompletableFuture(), v1);
+ assertEquals(1, a.get());
+ }}
+
+ /**
+ * Under default implementation, if an "exceptionally action"
+ * throws an exception, it completes exceptionally with that
+ * exception
+ */
+ public void testDefaultExceptionallyAsync_exceptionalCompletionActionFailed() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final AtomicInteger a = new AtomicInteger(0);
+ final CFException ex1 = new CFException();
+ final CFException ex2 = new CFException();
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) f.completeExceptionally(ex1);
+ final CompletionStage<Integer> g = d.exceptionallyAsync
+ ((Throwable t) -> {
+ threadAssertSame(t, ex1);
+ a.getAndIncrement();
+ throw ex2;
+ });
+ if (createIncomplete) f.completeExceptionally(ex1);
+
+ checkCompletedWithWrappedException(g.toCompletableFuture(), ex2);
+ checkCompletedExceptionally(f, ex1);
+ checkCompletedExceptionally(d.toCompletableFuture(), ex1);
+ assertEquals(1, a.get());
+ }}
+
+ /**
+ * default-implemented exceptionallyCompose result completes
+ * normally after normal completion of source
+ */
+ public void testDefaultExceptionallyCompose_normalCompletion() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ for (Integer v1 : new Integer[] { 1, null })
+ {
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final ExceptionalCompletableFutureFunction r =
+ new ExceptionalCompletableFutureFunction(ExecutionMode.SYNC);
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) assertTrue(f.complete(v1));
+ final CompletionStage<Integer> g = d.exceptionallyCompose(r);
+ if (createIncomplete) assertTrue(f.complete(v1));
+
+ checkCompletedNormally(f, v1);
+ checkCompletedNormally(g.toCompletableFuture(), v1);
+ r.assertNotInvoked();
+ }}
+
+ /**
+ * default-implemented exceptionallyCompose result completes
+ * normally after exceptional completion of source
+ */
+ public void testDefaultExceptionallyCompose_exceptionalCompletion() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final CFException ex = new CFException();
+ final ExceptionalCompletableFutureFunction r =
+ new ExceptionalCompletableFutureFunction(ExecutionMode.SYNC);
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletionStage<Integer> g = d.exceptionallyCompose(r);
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedExceptionally(f, ex);
+ checkCompletedNormally(g.toCompletableFuture(), r.value);
+ r.assertInvoked();
+ }}
+
+ /**
+ * default-implemented exceptionallyCompose completes
+ * exceptionally on exception if action does
+ */
+ public void testDefaultExceptionallyCompose_actionFailed() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final CFException ex = new CFException();
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final FailingExceptionalCompletableFutureFunction r
+ = new FailingExceptionalCompletableFutureFunction(ExecutionMode.SYNC);
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletionStage<Integer> g = d.exceptionallyCompose(r);
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedExceptionally(f, ex);
+ checkCompletedWithWrappedException(g.toCompletableFuture(), r.ex);
+ r.assertInvoked();
+ }}
+
+ /**
+ * default-implemented exceptionallyComposeAsync result completes
+ * normally after normal completion of source
+ */
+ public void testDefaultExceptionallyComposeAsync_normalCompletion() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ for (Integer v1 : new Integer[] { 1, null })
+ {
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final ExceptionalCompletableFutureFunction r =
+ new ExceptionalCompletableFutureFunction(ExecutionMode.ASYNC);
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) assertTrue(f.complete(v1));
+ final CompletionStage<Integer> g = d.exceptionallyComposeAsync(r);
+ if (createIncomplete) assertTrue(f.complete(v1));
+
+ checkCompletedNormally(f, v1);
+ checkCompletedNormally(g.toCompletableFuture(), v1);
+ r.assertNotInvoked();
+ }}
+
+ /**
+ * default-implemented exceptionallyComposeAsync result completes
+ * normally after exceptional completion of source
+ */
+ public void testDefaultExceptionallyComposeAsync_exceptionalCompletion() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final CFException ex = new CFException();
+ final ExceptionalCompletableFutureFunction r =
+ new ExceptionalCompletableFutureFunction(ExecutionMode.ASYNC);
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletionStage<Integer> g = d.exceptionallyComposeAsync(r);
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedExceptionally(f, ex);
+ checkCompletedNormally(g.toCompletableFuture(), r.value);
+ r.assertInvoked();
+ }}
+
+ /**
+ * default-implemented exceptionallyComposeAsync completes
+ * exceptionally on exception if action does
+ */
+ public void testDefaultExceptionallyComposeAsync_actionFailed() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final CFException ex = new CFException();
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final FailingExceptionalCompletableFutureFunction r
+ = new FailingExceptionalCompletableFutureFunction(ExecutionMode.ASYNC);
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletionStage<Integer> g = d.exceptionallyComposeAsync(r);
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedExceptionally(f, ex);
+ checkCompletedWithWrappedException(g.toCompletableFuture(), r.ex);
+ r.assertInvoked();
+ }}
+
+ /**
+ * default-implemented exceptionallyComposeAsync result completes
+ * normally after normal completion of source
+ */
+ public void testDefaultExceptionallyComposeAsyncExecutor_normalCompletion() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ for (Integer v1 : new Integer[] { 1, null })
+ {
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final ExceptionalCompletableFutureFunction r =
+ new ExceptionalCompletableFutureFunction(ExecutionMode.EXECUTOR);
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) assertTrue(f.complete(v1));
+ final CompletionStage<Integer> g = d.exceptionallyComposeAsync(r, new ThreadExecutor());
+ if (createIncomplete) assertTrue(f.complete(v1));
+
+ checkCompletedNormally(f, v1);
+ checkCompletedNormally(g.toCompletableFuture(), v1);
+ r.assertNotInvoked();
+ }}
+
+ /**
+ * default-implemented exceptionallyComposeAsync result completes
+ * normally after exceptional completion of source
+ */
+ public void testDefaultExceptionallyComposeAsyncExecutor_exceptionalCompletion() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final CFException ex = new CFException();
+ final ExceptionalCompletableFutureFunction r =
+ new ExceptionalCompletableFutureFunction(ExecutionMode.EXECUTOR);
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletionStage<Integer> g = d.exceptionallyComposeAsync(r, new ThreadExecutor());
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedExceptionally(f, ex);
+ checkCompletedNormally(g.toCompletableFuture(), r.value);
+ r.assertInvoked();
+ }}
+
+ /**
+ * default-implemented exceptionallyComposeAsync completes
+ * exceptionally on exception if action does
+ */
+ public void testDefaultExceptionallyComposeAsyncExecutor_actionFailed() {
+ for (boolean createIncomplete : new boolean[] { true, false })
+ {
+ final CFException ex = new CFException();
+ final CompletableFuture<Integer> f = new CompletableFuture<>();
+ final FailingExceptionalCompletableFutureFunction r
+ = new FailingExceptionalCompletableFutureFunction(ExecutionMode.EXECUTOR);
+ final DelegatedCompletionStage<Integer> d =
+ new DelegatedCompletionStage<Integer>(f);
+ if (!createIncomplete) f.completeExceptionally(ex);
+ final CompletionStage<Integer> g = d.exceptionallyComposeAsync(r, new ThreadExecutor());
+ if (createIncomplete) f.completeExceptionally(ex);
+
+ checkCompletedExceptionally(f, ex);
+ checkCompletedWithWrappedException(g.toCompletableFuture(), r.ex);
+ r.assertInvoked();
+ }}
+
}