stan/jdk-24
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

8210971: Add exception handling methods to CompletionStage and CompletableFuture

Browse Source 
Reviewed-by: martin, chegar
This commit is contained in:
Doug Lea 2018-09-28 08:45:46 -07:00
parent 5d3b3156e8
commit 0b431957a5
3 changed files with 869 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

165
src/java.base/share/classes/java/util/concurrent/CompletableFuture.java
View File

@ -957,17 +957,17 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
@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 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
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 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
}
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));
else
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
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 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
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 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
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 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* Creates a new complete CompletableFuture with given encoded result.
*/
CompletableFuture(Object r) {
this.result = r;
RESULT.setRelease(this, r);
}
/**
@ -2285,28 +2356,36 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
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 -------------- */

124
src/java.base/share/classes/java/util/concurrent/CompletionStage.java
View File

@ -850,6 +850,130 @@ public interface CompletionStage<T> {
public CompletionStage<T> exceptionally
(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

644
test/jdk/java/util/concurrent/tck/CompletableFutureTest.java
View File

@ -86,9 +86,13 @@ public class CompletableFutureTest extends JSR166TestCase {
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 @@ public class CompletableFutureTest extends JSR166TestCase {
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 class CompletableFutureTest extends JSR166TestCase {
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 @@ public class CompletableFutureTest extends JSR166TestCase {
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 @@ public class CompletableFutureTest extends JSR166TestCase {
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 @@ public class CompletableFutureTest extends JSR166TestCase {
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 @@ public class CompletableFutureTest extends JSR166TestCase {
(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 @@ public class CompletableFutureTest extends JSR166TestCase {
* 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 @@ public class CompletableFutureTest extends JSR166TestCase {
checkCompletedNormally(g, v1);
checkCompletedNormally(f, v1);
assertEquals(0, a.get());
}}
/**
@ -911,6 +971,7 @@ public class CompletableFutureTest extends JSR166TestCase {
* 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 @@ public class CompletableFutureTest extends JSR166TestCase {
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 @@ public class CompletableFutureTest extends JSR166TestCase {
* 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 @@ public class CompletableFutureTest extends JSR166TestCase {
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 @@ public class CompletableFutureTest extends JSR166TestCase {
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 @@ public class CompletableFutureTest extends JSR166TestCase {
// 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();
}}
}