This is an automated email from the ASF dual-hosted git repository.
reta pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/master by this push:
new 4f23c3f CXF-8242: Stop blocking executor thread on microprofile rest
asynchronous call. Fixing handler invocation flow.
4f23c3f is described below
commit 4f23c3f5456f2486493e623ce3c3844887e37255
Author: reta <[email protected]>
AuthorDate: Tue Mar 31 21:07:15 2020 -0400
CXF-8242: Stop blocking executor thread on microprofile rest asynchronous
call. Fixing handler invocation flow.
---
.../cxf/jaxrs/client/JaxrsClientCallback.java | 28 +++++++++++++++-----
.../cxf/jaxrs/client/JaxrsClientCallbackTest.java | 30 +++++++++++++++++++---
2 files changed, 48 insertions(+), 10 deletions(-)
diff --git
a/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java
b/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java
index 5db63bc..5c123cd 100644
---
a/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java
+++
b/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/JaxrsClientCallback.java
@@ -57,11 +57,21 @@ public class JaxrsClientCallback<T> extends ClientCallback {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
- boolean result = super.cancel(mayInterruptIfRunning);
- if (result && handler != null) {
- handler.failed(new CancellationException());
+ if (!started) {
+ // The handler has to be called *before* future completes
+ if (handler != null) {
+ handler.failed(new CancellationException());
+ }
+
+ delegate.cancel(mayInterruptIfRunning);
+ synchronized (this) {
+ notifyAll();
+ }
+
+ return true;
}
- return result;
+
+ return false;
}
public Future<T> createFuture() {
@@ -71,11 +81,13 @@ public class JaxrsClientCallback<T> extends ClientCallback {
@SuppressWarnings("unchecked")
public void handleResponse(Map<String, Object> ctx, Object[] res) {
context = ctx;
- delegate.complete(res);
+
+ // The handler has to be called *before* future completes
if (handler != null) {
handler.completed((T)res[0]);
}
+ delegate.complete(res);
synchronized (this) {
notifyAll();
}
@@ -84,11 +96,13 @@ public class JaxrsClientCallback<T> extends ClientCallback {
@Override
public void handleException(Map<String, Object> ctx, final Throwable ex) {
context = ctx;
- delegate.completeExceptionally(ex);
+
+ // The handler has to be called *before* future completes
if (handler != null) {
handler.failed(ex);
}
-
+
+ delegate.completeExceptionally(ex);
synchronized (this) {
notifyAll();
}
diff --git
a/rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java
b/rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java
index 4f1b124..e7618b1 100644
---
a/rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java
+++
b/rt/rs/client/src/test/java/org/apache/cxf/jaxrs/client/JaxrsClientCallbackTest.java
@@ -21,12 +21,14 @@ package org.apache.cxf.jaxrs.client;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
import javax.ws.rs.client.InvocationCallback;
@@ -38,6 +40,7 @@ import org.junit.Before;
import org.junit.Test;
import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertThrows;
@@ -48,17 +51,22 @@ public class JaxrsClientCallbackTest {
private InvocationCallback<String> handler;
private ScheduledExecutorService executor;
private JaxrsResponseFuture<String> future;
+ private AtomicReference<Object> state;
@Before
public void setUp() {
+ state = new AtomicReference<>();
+
executor = Executors.newSingleThreadScheduledExecutor();
handler = new InvocationCallback<String>() {
@Override
public void failed(Throwable throwable) {
+ state.set(throwable);
}
@Override
public void completed(String response) {
+ state.set(response);
}
};
@@ -85,8 +93,9 @@ public class JaxrsClientCallbackTest {
assertThat(future.get(10, TimeUnit.MILLISECONDS), equalTo("results"));
assertThat(future.isCancelled(), equalTo(false));
assertThat(future.isDone(), equalTo(true));
+ assertThat(state.get(), equalTo("results"));
}
-
+
@Test
public void testGetResponseContextOnSuccessCallback() throws Exception {
final CyclicBarrier barrier = new CyclicBarrier(2);
@@ -100,6 +109,7 @@ public class JaxrsClientCallbackTest {
assertThat(future.get(10, TimeUnit.MILLISECONDS), equalTo("results"));
assertThat(future.isCancelled(), equalTo(false));
assertThat(future.isDone(), equalTo(true));
+ assertThat(state.get(), equalTo("results"));
}
@Test
@@ -111,6 +121,7 @@ public class JaxrsClientCallbackTest {
assertThrows(ExecutionException.class, () -> future.get(10,
TimeUnit.MILLISECONDS));
assertThat(future.isCancelled(), equalTo(false));
assertThat(future.isDone(), equalTo(true));
+ assertThat(state.get(), instanceOf(RuntimeException.class));
}
@Test
@@ -124,7 +135,7 @@ public class JaxrsClientCallbackTest {
}
@Test
- public void testHandleCancellationCallback() throws Exception {
+ public void testHandleCancellationCallbackWithFuture() throws Exception {
final CyclicBarrier barrier = new CyclicBarrier(2);
schedule(barrier, () -> future.cancel(true));
barrier.await(5, TimeUnit.SECONDS);
@@ -133,8 +144,21 @@ public class JaxrsClientCallbackTest {
assertThrows(InterruptedException.class, () -> future.get(10,
TimeUnit.MILLISECONDS));
assertThat(future.isCancelled(), equalTo(true));
assertThat(future.isDone(), equalTo(true));
+ assertThat(state.get(), instanceOf(InterruptedException.class));
+ }
+
+ @Test
+ public void testHandleCancellationCallback() throws Exception {
+ final CyclicBarrier barrier = new CyclicBarrier(2);
+ schedule(barrier, () -> callback.cancel(true));
+ barrier.await(5, TimeUnit.SECONDS);
+
+ assertThrows(InterruptedException.class, () -> callback.get());
+ assertThrows(InterruptedException.class, () -> callback.get(10,
TimeUnit.MILLISECONDS));
+ assertThat(callback.isCancelled(), equalTo(true));
+ assertThat(callback.isDone(), equalTo(true));
+ assertThat(state.get(), instanceOf(CancellationException.class));
}
-
@Test
public void testHandleCancellationCallbackWhenStarted() throws Exception {
