This is an automated email from the ASF dual-hosted git repository. reta pushed a commit to branch 3.2.x-fixes in repository https://gitbox.apache.org/repos/asf/cxf.git
commit ef418f08285a88a64bb17c3610308cf5e2c8b68f 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. (cherry picked from commit 4f23c3f5456f2486493e623ce3c3844887e37255) (cherry picked from commit 2d43308da43874163d0b42ceaf358c3ffe4cc3c7) --- .../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 {
