Improve synchronization in ReactiveClientTest

Author: rschmitt

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/reactive/ReactiveClientTest.java

@@ -41,6 +41,7 @@
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -75,7 +76,6 @@
 import org.apache.hc.core5.util.Timeout;
 import org.hamcrest.CoreMatchers;
 import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.reactivestreams.Publisher;
@@ -212,17 +212,17 @@ void testRequestError() throws Exception {
 
         final ExecutionException exception = Assertions.assertThrows(ExecutionException.class, () ->
                 future.get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit()));
-        Assertions.assertTrue(exception.getCause() instanceof HttpStreamResetException);
+        Assertions.assertInstanceOf(HttpStreamResetException.class, exception.getCause());
         Assertions.assertSame(exceptionThrown, exception.getCause().getCause());
     }
 
     @Test
     void testRequestTimeout() throws Exception {
         final InetSocketAddress address = startServer();
         final HttpAsyncRequester requester = clientResource.start();
-        final AtomicBoolean requestPublisherWasCancelled = new AtomicBoolean();
+        final CountDownLatch requestPublisherCancellation = new CountDownLatch(1);
         final Publisher<ByteBuffer> publisher = Flowable.<ByteBuffer>never()
-                .doOnCancel(() -> requestPublisherWasCancelled.set(true));
+                .doOnCancel(requestPublisherCancellation::countDown);
         final ReactiveEntityProducer producer = new ReactiveEntityProducer(publisher, -1, null, null);
         final BasicRequestProducer request = getRequestProducer(address, producer);
 
@@ -231,7 +231,7 @@ void testRequestTimeout() throws Exception {
 
         final ExecutionException exception = Assertions.assertThrows(ExecutionException.class, () ->
                 future.get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit()));
-        Assertions.assertTrue(requestPublisherWasCancelled.get());
+        Assertions.assertTrue(requestPublisherCancellation.await(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit()));
         final Throwable cause = exception.getCause();
         if (versionPolicy == HttpVersionPolicy.FORCE_HTTP_1) {
             Assertions.assertTrue(cause instanceof SocketTimeoutException, "Expected SocketTimeoutException, but got " + cause.getClass().getName());
@@ -243,14 +243,13 @@ void testRequestTimeout() throws Exception {
     }
 
     @Test
-    @Disabled("Fails intermittently in GitHub Actions")
     void testResponseCancellation() throws Exception {
         final InetSocketAddress address = startServer();
         final HttpAsyncRequester requester = clientResource.start();
-        final AtomicBoolean requestPublisherWasCancelled = new AtomicBoolean();
+        final CountDownLatch requestPublisherCancellation = new CountDownLatch(1);
         final AtomicReference<Throwable> requestStreamError = new AtomicReference<>();
         final Publisher<ByteBuffer> stream = Reactive3TestUtils.produceStream(Long.MAX_VALUE, 1024, null)
-                .doOnCancel(() -> requestPublisherWasCancelled.set(true))
+                .doOnCancel(requestPublisherCancellation::countDown)
                 .doOnError(requestStreamError::set);
         final ReactiveEntityProducer producer = new ReactiveEntityProducer(stream, -1, null, null);
         final BasicRequestProducer request = getRequestProducer(address, producer);
@@ -273,8 +272,8 @@ void testResponseCancellation() throws Exception {
         assertThat(exception, CoreMatchers.anyOf(
                 CoreMatchers.instanceOf(CancellationException.class),
                 CoreMatchers.instanceOf(ExecutionException.class)));
-        Assertions.assertTrue(exception.getCause() instanceof HttpStreamResetException);
-        Assertions.assertTrue(requestPublisherWasCancelled.get());
+        Assertions.assertInstanceOf(HttpStreamResetException.class, exception.getCause());
+        Assertions.assertTrue(requestPublisherCancellation.await(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit()));
         Assertions.assertNull(requestStreamError.get());
     }