Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

SettableListenableFuture consistently tracks cancellation state

Browse Source 
Issue: SPR-15202
This commit is contained in:
Juergen Hoeller
2017-01-31 15:57:52 +01:00
parent 9b26fcd7d2
commit 9666fcc41d
2 changed files with 103 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
spring-core/src/main/java/org/springframework/util/concurrent/SettableListenableFuture.java
View File

@@ -27,13 +27,14 @@ import org.springframework.util.ReflectionUtils;
/** /**
* A {@link org.springframework.util.concurrent.ListenableFuture ListenableFuture} * A {@link org.springframework.util.concurrent.ListenableFuture ListenableFuture}
* whose value can be set via {@link #set(Object)} or * whose value can be set via {@link #set(T)} or {@link #setException(Throwable)}.
* {@link #setException(Throwable)}. It may also be cancelled. * It may also be cancelled.
* *
* <p>Inspired by {@code com.google.common.util.concurrent.SettableFuture}. * <p>Inspired by {@code com.google.common.util.concurrent.SettableFuture}.
* *
* @author Mattias Severson * @author Mattias Severson
* @author Rossen Stoyanchev * @author Rossen Stoyanchev
* @author Juergen Hoeller
* @since 4.1 * @since 4.1
*/ */
public class SettableListenableFuture<T> implements ListenableFuture<T> { public class SettableListenableFuture<T> implements ListenableFuture<T> {
@@ -92,8 +93,8 @@ public class SettableListenableFuture<T> implements ListenableFuture<T> {
@Override @Override
public boolean cancel(boolean mayInterruptIfRunning) { public boolean cancel(boolean mayInterruptIfRunning) {
this.settableTask.setCancelled(); boolean cancelled = this.settableTask.setCancelled();
boolean cancelled = this.listenableFuture.cancel(mayInterruptIfRunning); this.listenableFuture.cancel(mayInterruptIfRunning);
if (cancelled && mayInterruptIfRunning) { if (cancelled && mayInterruptIfRunning) {
interruptTask(); interruptTask();
} }
@@ -102,12 +103,12 @@ public class SettableListenableFuture<T> implements ListenableFuture<T> {
@Override @Override
public boolean isCancelled() { public boolean isCancelled() {
return this.listenableFuture.isCancelled(); return this.settableTask.isCancelled();
} }
@Override @Override
public boolean isDone() { public boolean isDone() {
return this.listenableFuture.isDone(); return this.settableTask.isDone();
} }
/** /**
@@ -152,26 +153,28 @@ public class SettableListenableFuture<T> implements ListenableFuture<T> {
private static final Object NO_VALUE = new Object(); private static final Object NO_VALUE = new Object();
private static final Object CANCELLED = new Object();
private final AtomicReference<Object> value = new AtomicReference<>(NO_VALUE); private final AtomicReference<Object> value = new AtomicReference<>(NO_VALUE);
private volatile boolean cancelled = false;
public boolean setValue(T value) { public boolean setValue(T value) {
if (this.cancelled) {
return false;
}
return this.value.compareAndSet(NO_VALUE, value); return this.value.compareAndSet(NO_VALUE, value);
} }
public boolean setException(Throwable exception) { public boolean setException(Throwable exception) {
if (this.cancelled) {
return false;
}
return this.value.compareAndSet(NO_VALUE, exception); return this.value.compareAndSet(NO_VALUE, exception);
} }
public void setCancelled() { public boolean setCancelled() {
this.cancelled = true; return this.value.compareAndSet(NO_VALUE, CANCELLED);
}
public boolean isCancelled() {
return (this.value.get() == CANCELLED);
}
public boolean isDone() {
return (this.value.get() != NO_VALUE);
} }
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")

114
spring-core/src/test/java/org/springframework/util/concurrent/SettableListenableFutureTests.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2016 the original author or authors. * Copyright 2002-2017 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -30,6 +30,7 @@ import static org.mockito.Mockito.*;
/** /**
* @author Mattias Severson * @author Mattias Severson
* @author Juergen Hoeller
*/ */
@SuppressWarnings({ "rawtypes", "unchecked" }) @SuppressWarnings({ "rawtypes", "unchecked" })
public class SettableListenableFutureTests { public class SettableListenableFutureTests {
@@ -39,29 +40,31 @@ public class SettableListenableFutureTests {
@Test @Test
public void validateInitialValues() { public void validateInitialValues() {
assertFalse(settableListenableFuture.isDone());
assertFalse(settableListenableFuture.isCancelled()); assertFalse(settableListenableFuture.isCancelled());
assertFalse(settableListenableFuture.isDone());
} }
@Test @Test
public void returnsSetValue() throws ExecutionException, InterruptedException { public void returnsSetValue() throws ExecutionException, InterruptedException {
String string = "hello"; String string = "hello";
boolean wasSet = settableListenableFuture.set(string); assertTrue(settableListenableFuture.set(string));
assertTrue(wasSet);
assertThat(settableListenableFuture.get(), equalTo(string)); assertThat(settableListenableFuture.get(), equalTo(string));
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void setValueUpdatesDoneStatus() { public void setValueUpdatesDoneStatus() {
settableListenableFuture.set("hello"); settableListenableFuture.set("hello");
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone()); assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void throwsSetExceptionWrappedInExecutionException() throws ExecutionException, InterruptedException { public void throwsSetExceptionWrappedInExecutionException() throws ExecutionException, InterruptedException {
Throwable exception = new RuntimeException(); Throwable exception = new RuntimeException();
boolean wasSet = settableListenableFuture.setException(exception); assertTrue(settableListenableFuture.setException(exception));
assertTrue(wasSet);
try { try {
settableListenableFuture.get(); settableListenableFuture.get();
fail("Expected ExecutionException"); fail("Expected ExecutionException");
@@ -69,13 +72,16 @@ public class SettableListenableFutureTests {
catch (ExecutionException ex) { catch (ExecutionException ex) {
assertThat(ex.getCause(), equalTo(exception)); assertThat(ex.getCause(), equalTo(exception));
} }
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void throwsSetErrorWrappedInExecutionException() throws ExecutionException, InterruptedException { public void throwsSetErrorWrappedInExecutionException() throws ExecutionException, InterruptedException {
Throwable exception = new OutOfMemoryError(); Throwable exception = new OutOfMemoryError();
boolean wasSet = settableListenableFuture.setException(exception); assertTrue(settableListenableFuture.setException(exception));
assertTrue(wasSet);
try { try {
settableListenableFuture.get(); settableListenableFuture.get();
fail("Expected ExecutionException"); fail("Expected ExecutionException");
@@ -83,12 +89,16 @@ public class SettableListenableFutureTests {
catch (ExecutionException ex) { catch (ExecutionException ex) {
assertThat(ex.getCause(), equalTo(exception)); assertThat(ex.getCause(), equalTo(exception));
} }
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void setValueTriggersCallback() { public void setValueTriggersCallback() {
String string = "hello"; String string = "hello";
final String[] callbackHolder = new String[1]; final String[] callbackHolder = new String[1];
settableListenableFuture.addCallback(new ListenableFutureCallback<String>() { settableListenableFuture.addCallback(new ListenableFutureCallback<String>() {
@Override @Override
public void onSuccess(String result) { public void onSuccess(String result) {
@@ -99,14 +109,18 @@ public class SettableListenableFutureTests {
fail("Expected onSuccess() to be called"); fail("Expected onSuccess() to be called");
} }
}); });
settableListenableFuture.set(string); settableListenableFuture.set(string);
assertThat(callbackHolder[0], equalTo(string)); assertThat(callbackHolder[0], equalTo(string));
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void setValueTriggersCallbackOnlyOnce() { public void setValueTriggersCallbackOnlyOnce() {
String string = "hello"; String string = "hello";
final String[] callbackHolder = new String[1]; final String[] callbackHolder = new String[1];
settableListenableFuture.addCallback(new ListenableFutureCallback<String>() { settableListenableFuture.addCallback(new ListenableFutureCallback<String>() {
@Override @Override
public void onSuccess(String result) { public void onSuccess(String result) {
@@ -117,15 +131,19 @@ public class SettableListenableFutureTests {
fail("Expected onSuccess() to be called"); fail("Expected onSuccess() to be called");
} }
}); });
settableListenableFuture.set(string); settableListenableFuture.set(string);
assertFalse(settableListenableFuture.set("good bye")); assertFalse(settableListenableFuture.set("good bye"));
assertThat(callbackHolder[0], equalTo(string)); assertThat(callbackHolder[0], equalTo(string));
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void setExceptionTriggersCallback() { public void setExceptionTriggersCallback() {
Throwable exception = new RuntimeException(); Throwable exception = new RuntimeException();
final Throwable[] callbackHolder = new Throwable[1]; final Throwable[] callbackHolder = new Throwable[1];
settableListenableFuture.addCallback(new ListenableFutureCallback<String>() { settableListenableFuture.addCallback(new ListenableFutureCallback<String>() {
@Override @Override
public void onSuccess(String result) { public void onSuccess(String result) {
@@ -136,14 +154,18 @@ public class SettableListenableFutureTests {
callbackHolder[0] = ex; callbackHolder[0] = ex;
} }
}); });
settableListenableFuture.setException(exception); settableListenableFuture.setException(exception);
assertThat(callbackHolder[0], equalTo(exception)); assertThat(callbackHolder[0], equalTo(exception));
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void setExceptionTriggersCallbackOnlyOnce() { public void setExceptionTriggersCallbackOnlyOnce() {
Throwable exception = new RuntimeException(); Throwable exception = new RuntimeException();
final Throwable[] callbackHolder = new Throwable[1]; final Throwable[] callbackHolder = new Throwable[1];
settableListenableFuture.addCallback(new ListenableFutureCallback<String>() { settableListenableFuture.addCallback(new ListenableFutureCallback<String>() {
@Override @Override
public void onSuccess(String result) { public void onSuccess(String result) {
@@ -154,20 +176,26 @@ public class SettableListenableFutureTests {
callbackHolder[0] = ex; callbackHolder[0] = ex;
} }
}); });
settableListenableFuture.setException(exception); settableListenableFuture.setException(exception);
assertFalse(settableListenableFuture.setException(new IllegalArgumentException())); assertFalse(settableListenableFuture.setException(new IllegalArgumentException()));
assertThat(callbackHolder[0], equalTo(exception)); assertThat(callbackHolder[0], equalTo(exception));
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void nullIsAcceptedAsValueToSet() throws ExecutionException, InterruptedException { public void nullIsAcceptedAsValueToSet() throws ExecutionException, InterruptedException {
settableListenableFuture.set(null); settableListenableFuture.set(null);
assertNull(settableListenableFuture.get()); assertNull(settableListenableFuture.get());
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void getWaitsForCompletion() throws ExecutionException, InterruptedException { public void getWaitsForCompletion() throws ExecutionException, InterruptedException {
final String string = "hello"; final String string = "hello";
new Thread(new Runnable() { new Thread(new Runnable() {
@Override @Override
public void run() { public void run() {
@@ -180,8 +208,11 @@ public class SettableListenableFutureTests {
} }
} }
}).start(); }).start();
String value = settableListenableFuture.get(); String value = settableListenableFuture.get();
assertThat(value, equalTo(string)); assertThat(value, equalTo(string));
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
@@ -198,6 +229,7 @@ public class SettableListenableFutureTests {
@Test @Test
public void getWithTimeoutWaitsForCompletion() throws ExecutionException, InterruptedException, TimeoutException { public void getWithTimeoutWaitsForCompletion() throws ExecutionException, InterruptedException, TimeoutException {
final String string = "hello"; final String string = "hello";
new Thread(new Runnable() { new Thread(new Runnable() {
@Override @Override
public void run() { public void run() {
@@ -210,65 +242,74 @@ public class SettableListenableFutureTests {
} }
} }
}).start(); }).start();
String value = settableListenableFuture.get(100L, TimeUnit.MILLISECONDS);
String value = settableListenableFuture.get(500L, TimeUnit.MILLISECONDS);
assertThat(value, equalTo(string)); assertThat(value, equalTo(string));
assertFalse(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void cancelPreventsValueFromBeingSet() { public void cancelPreventsValueFromBeingSet() {
boolean wasCancelled = settableListenableFuture.cancel(true); assertTrue(settableListenableFuture.cancel(true));
assertTrue(wasCancelled); assertFalse(settableListenableFuture.set("hello"));
boolean wasSet = settableListenableFuture.set("hello"); assertTrue(settableListenableFuture.isCancelled());
assertFalse(wasSet); assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void cancelSetsFutureToDone() { public void cancelSetsFutureToDone() {
settableListenableFuture.cancel(true); settableListenableFuture.cancel(true);
assertTrue(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone()); assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void cancelWithMayInterruptIfRunningTrueCallsOverridenMethod() { public void cancelWithMayInterruptIfRunningTrueCallsOverriddenMethod() {
InterruptableSettableListenableFuture tested = new InterruptableSettableListenableFuture(); InterruptableSettableListenableFuture tested = new InterruptableSettableListenableFuture();
tested.cancel(true); assertTrue(tested.cancel(true));
assertTrue(tested.calledInterruptTask()); assertTrue(tested.calledInterruptTask());
assertTrue(tested.isCancelled());
assertTrue(tested.isDone());
} }
@Test @Test
public void cancelWithMayInterruptIfRunningFalseDoesNotCallOverridenMethod() { public void cancelWithMayInterruptIfRunningFalseDoesNotCallOverriddenMethod() {
InterruptableSettableListenableFuture tested = new InterruptableSettableListenableFuture(); InterruptableSettableListenableFuture tested = new InterruptableSettableListenableFuture();
tested.cancel(false); assertTrue(tested.cancel(false));
assertFalse(tested.calledInterruptTask()); assertFalse(tested.calledInterruptTask());
assertTrue(tested.isCancelled());
assertTrue(tested.isDone());
} }
@Test @Test
public void setPreventsCancel() { public void setPreventsCancel() {
boolean wasSet = settableListenableFuture.set("hello"); assertTrue(settableListenableFuture.set("hello"));
assertTrue(wasSet); assertFalse(settableListenableFuture.cancel(true));
boolean wasCancelled = settableListenableFuture.cancel(true); assertFalse(settableListenableFuture.isCancelled());
assertFalse(wasCancelled); assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void cancelPreventsExceptionFromBeingSet() { public void cancelPreventsExceptionFromBeingSet() {
boolean wasCancelled = settableListenableFuture.cancel(true); assertTrue(settableListenableFuture.cancel(true));
assertTrue(wasCancelled); assertFalse(settableListenableFuture.setException(new RuntimeException()));
boolean wasSet = settableListenableFuture.setException(new RuntimeException()); assertTrue(settableListenableFuture.isCancelled());
assertFalse(wasSet); assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void setExceptionPreventsCancel() { public void setExceptionPreventsCancel() {
boolean wasSet = settableListenableFuture.setException(new RuntimeException()); assertTrue(settableListenableFuture.setException(new RuntimeException()));
assertTrue(wasSet); assertFalse(settableListenableFuture.cancel(true));
boolean wasCancelled = settableListenableFuture.cancel(true); assertFalse(settableListenableFuture.isCancelled());
assertFalse(wasCancelled); assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
public void cancelStateThrowsExceptionWhenCallingGet() throws ExecutionException, InterruptedException { public void cancelStateThrowsExceptionWhenCallingGet() throws ExecutionException, InterruptedException {
settableListenableFuture.cancel(true); settableListenableFuture.cancel(true);
try { try {
settableListenableFuture.get(); settableListenableFuture.get();
fail("Expected CancellationException"); fail("Expected CancellationException");
@@ -276,6 +317,9 @@ public class SettableListenableFutureTests {
catch (CancellationException ex) { catch (CancellationException ex) {
// expected // expected
} }
assertTrue(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
@@ -292,13 +336,17 @@ public class SettableListenableFutureTests {
} }
} }
}).start(); }).start();
try { try {
settableListenableFuture.get(100L, TimeUnit.MILLISECONDS); settableListenableFuture.get(500L, TimeUnit.MILLISECONDS);
fail("Expected CancellationException"); fail("Expected CancellationException");
} }
catch (CancellationException ex) { catch (CancellationException ex) {
// expected // expected
} }
assertTrue(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
@@ -312,6 +360,9 @@ public class SettableListenableFutureTests {
settableListenableFuture.set("hello"); settableListenableFuture.set("hello");
verifyNoMoreInteractions(callback); verifyNoMoreInteractions(callback);
assertTrue(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }
@Test @Test
@@ -325,6 +376,9 @@ public class SettableListenableFutureTests {
settableListenableFuture.setException(new RuntimeException()); settableListenableFuture.setException(new RuntimeException());
verifyNoMoreInteractions(callback); verifyNoMoreInteractions(callback);
assertTrue(settableListenableFuture.isCancelled());
assertTrue(settableListenableFuture.isDone());
} }