From 667ed4ca1b4ef34aac4fd1a126afe282c55ecebd Mon Sep 17 00:00:00 2001 From: Andrew Toivonen Date: Wed, 1 Jul 2026 14:35:38 -0500 Subject: [PATCH] Stop HealthPoller from polling a closed client after cancellation --- build.sbt | 1 + .../dwolla/rsocket/consul/HealthPoller.java | 64 ++++++++++++------- .../dwolla/rsocket/HealthPollerShould.java | 61 ++++++++++++++++++ 3 files changed, 102 insertions(+), 24 deletions(-) diff --git a/build.sbt b/build.sbt index c014dda..d364cec 100644 --- a/build.sbt +++ b/build.sbt @@ -58,6 +58,7 @@ lazy val `rsocket-consul-java` = (project in file(".")) "org.junit.platform" % "junit-platform-launcher" % "1.14.4" % Test, "net.aichler" % "jupiter-interface" % JupiterKeys.jupiterVersion.value % Test, "org.mockito" % "mockito-core" % "5.23.0" % Test, + "io.projectreactor" % "reactor-test" % "3.4.41" % Test, "org.slf4j" % "slf4j-nop" % "2.0.17" % Test, ) }, diff --git a/src/main/java/com/dwolla/rsocket/consul/HealthPoller.java b/src/main/java/com/dwolla/rsocket/consul/HealthPoller.java index aed1f84..8444093 100644 --- a/src/main/java/com/dwolla/rsocket/consul/HealthPoller.java +++ b/src/main/java/com/dwolla/rsocket/consul/HealthPoller.java @@ -4,6 +4,8 @@ import com.google.gson.Gson; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import reactor.core.Disposable; +import reactor.core.Disposables; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -19,6 +21,7 @@ public class HealthPoller implements AutoCloseable { private final int StartingIndex = 0; + private final Duration RetryDelay = Duration.ofSeconds(5); private final String healthUrl = "%s/v1/health/service/%s?passing=true&index=%d&wait=1m"; private final Gson gson = new Gson(); @@ -35,32 +38,45 @@ public HealthPoller(HttpClient client, String consulHost) { public Flux> start(String serviceName) { return Flux.create(emitter -> { + Disposable.Swap backoff = Disposables.swap(); + emitter.onDispose(backoff); + Recursive>> r = new Recursive<>(); r.func = - index -> - client - .get(String.format(healthUrl, consulHost, serviceName, index)) - .thenApply( - res -> { - int nextIdx = getNextIdxFrom(res); - - if (nextIdx > index) { - lastResponse = getAddressesFrom(res.getBody()); - emitter.next(lastResponse); - } - - return nextIdx; - }) - .whenComplete( - (nextId, th) -> { - if (th != null) { - logger.warn("Got an exception while long polling Consul.", th); - Mono.delay(Duration.ofSeconds(5)) - .subscribe(i -> r.func.apply(StartingIndex)); - } else { - r.func.apply(nextId); - } - }); + index -> { + if (emitter.isCancelled()) { + return CompletableFuture.completedFuture(index); + } + + return client + .get(String.format(healthUrl, consulHost, serviceName, index)) + .thenApply( + res -> { + int nextIdx = getNextIdxFrom(res); + + if (nextIdx > index) { + lastResponse = getAddressesFrom(res.getBody()); + emitter.next(lastResponse); + } + + return nextIdx; + }) + .whenComplete( + (nextId, th) -> { + if (emitter.isCancelled()) { + return; + } + + if (th != null) { + logger.warn("Got an exception while long polling Consul.", th); + backoff.update( + Mono.delay(RetryDelay) + .subscribe(i -> r.func.apply(StartingIndex))); + } else { + r.func.apply(nextId); + } + }); + }; r.func.apply(StartingIndex); }); diff --git a/src/test/java/com/dwolla/rsocket/HealthPollerShould.java b/src/test/java/com/dwolla/rsocket/HealthPollerShould.java index 8930e06..18d5518 100644 --- a/src/test/java/com/dwolla/rsocket/HealthPollerShould.java +++ b/src/test/java/com/dwolla/rsocket/HealthPollerShould.java @@ -5,7 +5,9 @@ import com.dwolla.rsocket.consul.SimpleResponse; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import reactor.core.Disposable; import reactor.core.publisher.Flux; +import reactor.test.scheduler.VirtualTimeScheduler; import java.time.Duration; import java.util.*; @@ -14,7 +16,11 @@ import java.util.stream.Collectors; import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; class HealthPollerShould { @@ -171,4 +177,59 @@ void startOverAtIndexZeroAfterFiveSecondsIfAnExceptionOccurs() throws ExecutionE assertTrue(output.contains(host1)); assertTrue(output.contains(host2)); } + + @Test + void stopPollingOnceTheSubscriptionIsCancelled() { + final String firstUrl = + CONSUL_HOST + "/v1/health/service/" + SERVICE_NAME + "?passing=true&index=0&wait=1m"; + final String secondUrl = + CONSUL_HOST + + "/v1/health/service/" + + SERVICE_NAME + + "?passing=true&index=" + + firstIndex + + "&wait=1m"; + final SimpleResponse res = new SimpleResponse(JSON_RES, new ArrayList<>(firstHeaders.entrySet())); + final CompletableFuture firstFuture = new CompletableFuture<>(); + + when(client.get(firstUrl)).thenReturn(firstFuture); + + // The initial request is issued synchronously on subscribe. + final Disposable subscription = poller.start(SERVICE_NAME).subscribe(); + + // Cancel before the in-flight request resolves. Completing it now runs the + // dependent stages synchronously, so whenComplete observes the cancellation + // and returns instead of rescheduling the next poll. + subscription.dispose(); + firstFuture.complete(res); + + verify(client, times(1)).get(anyString()); + verify(client, never()).get(secondUrl); + } + + @Test + void notRestartAfterTheBackoffWhenCancelledDuringTheDelay() { + final VirtualTimeScheduler scheduler = VirtualTimeScheduler.getOrSet(); + try { + final String firstUrl = + CONSUL_HOST + "/v1/health/service/" + SERVICE_NAME + "?passing=true&index=0&wait=1m"; + final CompletableFuture failedFuture = new CompletableFuture<>(); + + when(client.get(firstUrl)).thenReturn(failedFuture); + + final Disposable subscription = poller.start(SERVICE_NAME).subscribe(); + + // Exception schedules the 5s backoff on the virtual scheduler. + failedFuture.completeExceptionally(new RuntimeException("Something happened")); + + // Cancel during the backoff window, then advance past when it would have fired. + subscription.dispose(); + scheduler.advanceTimeBy(Duration.ofSeconds(6)); + + // The loop stays dead: only the initial index=0 request was ever issued. + verify(client, times(1)).get(anyString()); + } finally { + VirtualTimeScheduler.reset(); + } + } }