From 0207fc2690aacf945c7ddf8d536dc8e963e04147 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Tue, 11 Aug 2026 17:37:48 -0400 Subject: [PATCH 1/2] feat: expose setInitialRetryDelayMillis and setMaxRetryDelayMillis on EventSource MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two narrow public setter methods on EventSource for SDK-driven regime switching. Motivating use case is RETRY-spec conformance in server SDKs: on classification of a failure as "unexpected" (per RETRY §1.6 / §1.7), the SDK's data source needs to shift the retry timing into an extended regime (e.g. initial 5 min, max 1 hr), and shift back after healthy operation. Cross-referenced in launchdarkly/sdk-scratchpad's server-sdk-guide.md, and tracked as SDK-2789 (Java) under the RETRY-conformance epic SDK-2775. API: - setInitialRetryDelayMillis(long) updates the existing volatile baseRetryDelayMillis field (the same field wire-side SetRetryDelayEvent already updates). If the current strategy is a DefaultRetryDelayStrategy, the exponent counter is also reset so the first subsequent apply() uses the new base directly. Non-Default strategies just see the new base on the next apply() call. - setMaxRetryDelayMillis(long) constructs a new DefaultRetryDelayStrategy with the specified max delay and the exponent counter reset to 0, preserving the current backoff multiplier and jitter multiplier, and atomically swaps the reference. No-op for custom strategy impls (which don't expose a max-delay concept via the abstract interface). Both setters realize the "reset n when delays change" invariant from the LaunchDarkly server-SDK implementation guide: the first attempt in a new regime uses the new initial delay directly rather than newBase * 2^oldN. Under the hood: DefaultRetryDelayStrategy gains two package-private helpers (withResetCounter and withMaxDelayMillisAndResetCounter) that build copies with a fresh counter. The public builder methods (maxDelay, backoffMultiplier, jitterMultiplier) are unchanged. currentRetryDelayStrategy is now volatile to support the "caller can invoke setters from any thread" contract. Tests: 6 new tests covering direct setter effects, counter-reset behavior, composed extended-regime sequence, wire-hint interaction, and no-op behavior on non-Default strategies. Full unit suite and sse-contract-tests both green. --- .../DefaultRetryDelayStrategy.java | 17 ++ .../launchdarkly/eventsource/EventSource.java | 88 ++++++- ...ventSourceRetryDelayStrategyUsageTest.java | 242 ++++++++++++++++++ 3 files changed, 345 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java b/src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java index 239438a..a01dcfd 100644 --- a/src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java +++ b/src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java @@ -114,6 +114,23 @@ private DefaultRetryDelayStrategy( this.backoffMultiplier = backoffMultiplier; this.jitterMultiplier = jitterMultiplier; } + + // Package-private helper used by EventSource.setInitialRetryDelayMillis. Returns a + // copy with the exponent counter reset to 0 but the max delay, backoff multiplier, + // and jitter multiplier preserved. Ensures the next apply() uses the base delay + // directly, matching the "first attempt in a new regime uses the initial delay" + // invariant from the RETRY specification and the server-SDK implementation guide. + DefaultRetryDelayStrategy withResetCounter() { + return new DefaultRetryDelayStrategy(0, this.maxDelayMillis, this.backoffMultiplier, this.jitterMultiplier); + } + + // Package-private helper used by EventSource.setMaxRetryDelayMillis. Returns a + // copy with a new max delay AND the exponent counter reset to 0. Preserves the + // backoff multiplier and jitter multiplier. Same reset-on-delay-change invariant + // as withResetCounter(). + DefaultRetryDelayStrategy withMaxDelayMillisAndResetCounter(long newMaxDelayMillis) { + return new DefaultRetryDelayStrategy(0, newMaxDelayMillis, this.backoffMultiplier, this.jitterMultiplier); + } @Override public Result apply(long baseDelayMillis) { diff --git a/src/main/java/com/launchdarkly/eventsource/EventSource.java b/src/main/java/com/launchdarkly/eventsource/EventSource.java index f5c6f94..6610522 100644 --- a/src/main/java/com/launchdarkly/eventsource/EventSource.java +++ b/src/main/java/com/launchdarkly/eventsource/EventSource.java @@ -88,11 +88,18 @@ public class EventSource implements Closeable { // accessed from the thread that is reading from EventSource. private EventParser eventParser; ErrorStrategy currentErrorStrategy; - RetryDelayStrategy currentRetryDelayStrategy; private long connectedTime; private long disconnectedTime; private StreamEvent nextEvent; + // currentRetryDelayStrategy is volatile because it can be updated via the public + // setMaxRetryDelayMillis / setInitialRetryDelayMillis setters from any thread + // (e.g., an SDK error handler that runs on the reading thread, or a caller that + // wants to swap regimes from a different thread). Updates are always full-object + // replacements of the immutable strategy; volatile provides the necessary + // publication semantics. + volatile RetryDelayStrategy currentRetryDelayStrategy; + // These fields are set by the thread that is reading the stream, but can // be modified from other threads if they call stop() or interrupt(). We // use AtomicReference because we need atomicity in updates. @@ -212,7 +219,84 @@ public long getBaseRetryDelayMillis() { public long getNextRetryDelayMillis() { return nextReconnectDelayMillis; } - + + /** + * Updates the base retry delay used for computing subsequent reconnect delays. + *

+ * This is the SDK-side entry point for RETRY-spec regime switching: for example, + * a data source that has classified a failure as "unexpected" per the RETRY + * specification and wants to transition into an extended-regime backoff calls + * this method with the extended-regime initial delay (e.g., 5 minutes). + *

+ * Semantics match the SSE {@code retry:} field: the new value becomes the base + * used by the current {@link RetryDelayStrategy} on each subsequent {@code apply()} + * call, and the strategy's internal exponent counter is reset so the first + * subsequent reconnect uses the new base directly (rather than the new base + * multiplied by the current backoff exponent). If the current strategy is a + * {@link DefaultRetryDelayStrategy}, the counter reset preserves the strategy's + * max delay, backoff multiplier, and jitter multiplier. For any other custom + * {@link RetryDelayStrategy} implementation, only the base delay field is + * updated; the strategy's own state is left untouched (custom strategies can + * observe the new base via the {@code baseDelayMillis} argument to their + * {@code apply()} method). + *

+ * This method is thread-safe. + * + * @param millis the new base retry delay in milliseconds + * @since 4.1.0 + * @see #setMaxRetryDelayMillis(long) + * @see #getBaseRetryDelayMillis() + */ + public void setInitialRetryDelayMillis(long millis) { + baseRetryDelayMillis = millis; + RetryDelayStrategy current = currentRetryDelayStrategy; + if (current instanceof DefaultRetryDelayStrategy) { + currentRetryDelayStrategy = ((DefaultRetryDelayStrategy) current).withResetCounter(); + } + // For non-Default strategies we can't force a counter reset; the strategy's + // apply() will see the new baseDelayMillis on the next call and behave accordingly. + } + + /** + * Updates the maximum retry delay used by the current retry delay strategy. + *

+ * This is the SDK-side entry point for RETRY-spec regime switching: paired with + * {@link #setInitialRetryDelayMillis(long)}, a data source calls this method with + * the extended-regime maximum (e.g., 1 hour) when transitioning into the extended + * regime, and with the normal-regime maximum when transitioning back after a + * healthy-operation reset. + *

+ * Internally: constructs a new {@link DefaultRetryDelayStrategy} instance with the + * given max delay and the exponent counter reset to 0, preserving the current + * backoff multiplier and jitter multiplier, and atomically swaps the reference. + * The counter reset ensures the first subsequent reconnect uses the current base + * delay directly rather than the base multiplied by the current backoff exponent + * (matching the "reset {@code n} when delays change" invariant from the RETRY + * specification's server-SDK implementation guide). + *

+ * Currently only supported when the active strategy is a + * {@link DefaultRetryDelayStrategy} — the standard case for callers that + * configured retry via {@link Builder#retryDelay(long, TimeUnit)} or the + * default strategy. If the active strategy is a custom implementation, this + * method is a silent no-op (custom strategies define their own timing shape + * and don't expose a "max delay" concept via the abstract + * {@link RetryDelayStrategy} interface). + *

+ * This method is thread-safe. + * + * @param millis the new max retry delay in milliseconds + * @since 4.1.0 + * @see #setInitialRetryDelayMillis(long) + */ + public void setMaxRetryDelayMillis(long millis) { + RetryDelayStrategy current = currentRetryDelayStrategy; + if (current instanceof DefaultRetryDelayStrategy) { + currentRetryDelayStrategy = + ((DefaultRetryDelayStrategy) current).withMaxDelayMillisAndResetCounter(millis); + } + // else: silent no-op for custom strategies (see Javadoc). + } + /** * Attempts to start the stream if it is not already active. *

diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java index 30de94b..5fb23bd 100644 --- a/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java +++ b/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java @@ -166,6 +166,248 @@ public void retryDelayStrategyIsResetAfterThreshold() throws Exception { } } + // Tests for setInitialRetryDelayMillis and setMaxRetryDelayMillis. These are the + // SDK-side entry points for RETRY-spec regime switching (see LaunchDarkly's + // server-SDK implementation guide, SDK-2775). + + @Test + public void setInitialRetryDelayMillisUpdatesGetBaseRetryDelayMillis() throws Exception { + MockConnectStrategy mock = new MockConnectStrategy(); + mock.configureRequests(respondWithStream()); // stays connected + + try (EventSource es = baseBuilder(mock).retryDelay(1000, null).build()) { + es.start(); + assertThat(es.getBaseRetryDelayMillis(), equalTo(1000L)); + + es.setInitialRetryDelayMillis(5000L); + assertThat(es.getBaseRetryDelayMillis(), equalTo(5000L)); + + es.setInitialRetryDelayMillis(300000L); + assertThat(es.getBaseRetryDelayMillis(), equalTo(300000L)); + } + } + + @Test + public void setInitialRetryDelayMillisResetsExponentCounter() throws Exception { + // After some retries have advanced the exponent counter, calling + // setInitialRetryDelayMillis(newBase) must reset the counter so the next retry + // uses the new base directly rather than newBase * multiplier^currentCounter. + // This is the "reset n when delays change" invariant. + MockConnectStrategy mock = new MockConnectStrategy(); + PipedStreamRequestHandler[] streams = new PipedStreamRequestHandler[4]; + for (int i = 0; i < streams.length; i++) { + streams[i] = respondWithStream(); + } + mock.configureRequests(streams); + + long normalBase = 100; + long extendedBase = 5000; + // Zero jitter so retry delays are deterministic. + RetryDelayStrategy strategy = RetryDelayStrategy.defaultStrategy() + .jitterMultiplier(0) + .backoffMultiplier(2) + .maxDelay(1_000_000, null); // effectively no cap for this test + + try (EventSource es = baseBuilder(mock) + .retryDelayStrategy(strategy) + .retryDelay(normalBase, null) + .build()) { + es.start(); + + // Trigger fault #1: exponent counter starts at 0, delay = normalBase * 2^0 = 100. + streams[0].close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(es.getNextRetryDelayMillis(), equalTo(normalBase)); + + // Trigger fault #2: counter advances, delay = normalBase * 2^1 = 200. + streams[1].close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(es.getNextRetryDelayMillis(), equalTo(normalBase * 2)); + + // SDK-side regime switch: set the new base and expect the counter to reset. + es.setInitialRetryDelayMillis(extendedBase); + + // Trigger fault #3: delay should be extendedBase * 2^0 = 5000, NOT + // extendedBase * 2^2 = 20000 (which would happen if the counter wasn't reset). + streams[2].close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(es.getNextRetryDelayMillis(), equalTo(extendedBase)); + } + } + + @Test + public void setMaxRetryDelayMillisClampsAndResetsExponentCounter() throws Exception { + // After the exponent counter has advanced, calling setMaxRetryDelayMillis with a + // new max should both (a) reset the counter to 0 so the next retry uses the base + // delay directly, and (b) apply the new max as the ceiling for subsequent + // exponential progression. + MockConnectStrategy mock = new MockConnectStrategy(); + PipedStreamRequestHandler[] streams = new PipedStreamRequestHandler[5]; + for (int i = 0; i < streams.length; i++) { + streams[i] = respondWithStream(); + } + mock.configureRequests(streams); + + long base = 1000; + RetryDelayStrategy strategy = RetryDelayStrategy.defaultStrategy() + .jitterMultiplier(0) + .backoffMultiplier(2) + .maxDelay(30000, null); // normal-regime cap + + try (EventSource es = baseBuilder(mock) + .retryDelayStrategy(strategy) + .retryDelay(base, null) + .build()) { + es.start(); + + // Advance counter: 1000, 2000, 4000. + long[] initialProgression = new long[] { base, base * 2, base * 4 }; + for (int i = 0; i < initialProgression.length; i++) { + streams[i].close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(es.getNextRetryDelayMillis(), equalTo(initialProgression[i])); + } + + // Change max to something large (extended-regime cap: 1hr). Counter resets. + es.setMaxRetryDelayMillis(3_600_000L); + + // Next fault: with counter reset to 0, delay = base * 2^0 = base = 1000 + // (NOT base * 2^3 = 8000 which would happen if counter wasn't reset). + streams[3].close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(es.getNextRetryDelayMillis(), equalTo(base)); + + // Following fault: counter=1, delay = base * 2 = 2000. Under the new max + // (1hr), no clamping occurs. + streams[4].close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(es.getNextRetryDelayMillis(), equalTo(base * 2)); + } + } + + @Test + public void setInitialAndSetMaxComposeForExtendedRegimeSequence() throws Exception { + // Simulates a full SDK-side transition into an extended regime: caller invokes + // both setters, then observes the RETRY-spec extended-regime doubling shape + // (5min, 10min, 20min, 40min, then clamp to 60min) — realized here at ms-scale + // (10, 20, 40, 80, 120 clamp) so the test doesn't have to actually wall-clock + // wait through 5-minute retry sleeps. Same doubling+clamp shape. + MockConnectStrategy mock = new MockConnectStrategy(); + PipedStreamRequestHandler[] streams = new PipedStreamRequestHandler[7]; + for (int i = 0; i < streams.length; i++) { + streams[i] = respondWithStream(); + } + mock.configureRequests(streams); + + // Normal-regime initial (1s) and cap (30s), just so the builder is happy. + // Deterministic jitter=0. + RetryDelayStrategy strategy = RetryDelayStrategy.defaultStrategy() + .jitterMultiplier(0) + .backoffMultiplier(2) + .maxDelay(30_000, null); + + try (EventSource es = baseBuilder(mock) + .retryDelayStrategy(strategy) + .retryDelay(1000, null) + .build()) { + es.start(); + + // Regime switch to extended, scaled to ms so the test wall-clock stays under 1s. + long extendedInitial = 10; + long extendedMax = 120; + es.setInitialRetryDelayMillis(extendedInitial); + es.setMaxRetryDelayMillis(extendedMax); + + long[] expected = { + extendedInitial, // 10 ms (analog of 5 min) + extendedInitial * 2, // 20 ms (analog of 10 min) + extendedInitial * 4, // 40 ms (analog of 20 min) + extendedInitial * 8, // 80 ms (analog of 40 min) + extendedMax, // clamp (analog of 60 min) + extendedMax, // still clamped + extendedMax, // still clamped + }; + for (int i = 0; i < expected.length; i++) { + streams[i].close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(es.getNextRetryDelayMillis(), equalTo(expected[i])); + } + } + } + + @Test + public void wireRetryHintStillTakesEffectAfterSdkSideSetters() throws Exception { + // The server-directed retry: hint in the SSE wire remains authoritative for the + // base delay. This test pins that behavior after the SDK-side setters have been + // used to transition into an extended regime — a subsequent wire hint should + // override the SDK's chosen initial delay. + MockConnectStrategy mock = new MockConnectStrategy(); + // First: a stream that emits a retry: hint of 750ms, then closes. + mock.configureRequests(respondWithDataAndThenEnd("retry: 750\n\n")); + mock.configureRequests(respondWithStream()); + + RetryDelayStrategy strategy = RetryDelayStrategy.defaultStrategy() + .jitterMultiplier(0) + .backoffMultiplier(2) + .maxDelay(3_600_000, null); + + try (EventSource es = baseBuilder(mock) + .retryDelayStrategy(strategy) + .retryDelay(1000, null) + .build()) { + es.start(); + + // SDK transitions to extended: initial=5min, max=1hr. + es.setInitialRetryDelayMillis(300_000L); + es.setMaxRetryDelayMillis(3_600_000L); + assertThat(es.getBaseRetryDelayMillis(), equalTo(300_000L)); + + // Read from the stream — it will emit a retry: 750 line, which updates + // baseRetryDelayMillis and resets the strategy. The retry: line is consumed + // internally by EventSource (not surfaced as an event); the FaultEvent from + // the stream ending follows. + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + + // Base delay is now the wire-hinted 750ms, not the SDK-set 5min. + assertThat(es.getBaseRetryDelayMillis(), equalTo(750L)); + // And the computed retry delay reflects the wire hint (counter reset by + // resetRetryDelayStrategy which the wire-hint path calls). + assertThat(es.getNextRetryDelayMillis(), equalTo(750L)); + } + } + + @Test + public void settersOnCustomRetryDelayStrategyDoNotThrow() throws Exception { + // Non-DefaultRetryDelayStrategy: setInitialRetryDelayMillis still updates the + // base delay field (observable via getBaseRetryDelayMillis), and + // setMaxRetryDelayMillis is a silent no-op. + MockConnectStrategy mock = new MockConnectStrategy(); + mock.configureRequests(respondWithStream()); + + RetryDelayStrategy custom = new FixedRetryDelayStrategy(0); + + try (EventSource es = baseBuilder(mock) + .retryDelayStrategy(custom) + .retryDelay(1000, null) + .build()) { + es.start(); + + es.setInitialRetryDelayMillis(5000L); + assertThat(es.getBaseRetryDelayMillis(), equalTo(5000L)); + + // No exception. + es.setMaxRetryDelayMillis(60000L); + } + } + private static class ArithmeticallyIncreasingRetryDelayStrategy extends RetryDelayStrategy { private final int increment; private final int counter; From 933c93932e3aa4b0d95a5cafca63cd509c25ecb6 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Wed, 12 Aug 2026 10:01:29 -0400 Subject: [PATCH 2/2] fix: recompute pending reconnect delay when regime changes mid-fault An SDK error handler running under ErrorStrategy.alwaysContinue calls setInitialRetryDelayMillis / setMaxRetryDelayMillis inside its handleError callback, in response to a fault that has just been classified as UNEXPECTED per the RETRY specification. The immediately-prior computeReconnectDelay() call had already stored nextReconnectDelayMillis using the pre-transition strategy, so without a recompute the upcoming reconnect would use the OLD regime's timing (e.g., 1 ms normal-regime delay) and the extended-regime backoff would kick in only starting from the NEXT fault. Fix: - Track a pendingReconnectWait flag: set by computeReconnectDelay after a fault, cleared by tryStart on successful reconnect. - setInitialRetryDelayMillis and setMaxRetryDelayMillis, if pendingReconnectWait is true, recompute nextReconnectDelayMillis with the just-updated strategy. Do NOT advance the strategy (prior computeReconnectDelay already did that; advancing here would double-increment the counter for the next fault). Verified via the sdk-test-harness RETRY-conformance streaming/retry test "enters extended-regime backoff after unexpected HTTP error", which is now green (was failing prior to this fix because the SDK reconnected at normal-regime timing after the first 401). --- .../launchdarkly/eventsource/EventSource.java | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/src/main/java/com/launchdarkly/eventsource/EventSource.java b/src/main/java/com/launchdarkly/eventsource/EventSource.java index 6610522..3af5c08 100644 --- a/src/main/java/com/launchdarkly/eventsource/EventSource.java +++ b/src/main/java/com/launchdarkly/eventsource/EventSource.java @@ -100,6 +100,12 @@ public class EventSource implements Closeable { // publication semantics. volatile RetryDelayStrategy currentRetryDelayStrategy; + // Set to true by computeReconnectDelay() (following a fault), and cleared by + // tryStart() on successful reconnect. Used by setInitialRetryDelayMillis / + // setMaxRetryDelayMillis to know whether they should recompute the pending + // reconnect delay (SDK-driven mid-flight regime change). + private volatile boolean pendingReconnectWait = false; + // These fields are set by the thread that is reading the stream, but can // be modified from other threads if they call stop() or interrupt(). We // use AtomicReference because we need atomicity in updates. @@ -255,6 +261,11 @@ public void setInitialRetryDelayMillis(long millis) { } // For non-Default strategies we can't force a counter reset; the strategy's // apply() will see the new baseDelayMillis on the next call and behave accordingly. + // If a reconnect delay was already computed by the immediately-prior fault + // (which used the OLD strategy), recompute it now so the pending wait + // reflects the new regime. Without this, an SDK-driven regime transition + // from inside an error handler would take effect only on the NEXT fault. + recomputeNextReconnectDelayForRegimeSwitch(); } /** @@ -295,6 +306,30 @@ public void setMaxRetryDelayMillis(long millis) { ((DefaultRetryDelayStrategy) current).withMaxDelayMillisAndResetCounter(millis); } // else: silent no-op for custom strategies (see Javadoc). + // Same rationale as setInitialRetryDelayMillis: recompute the pending + // reconnect delay so an SDK-driven regime transition takes effect on the + // upcoming reconnect (not the NEXT fault's). + recomputeNextReconnectDelayForRegimeSwitch(); + } + + // Recompute nextReconnectDelayMillis using the current strategy + base delay. + // Called from the SDK-facing setters so a regime transition inside an error + // handler affects the pending reconnect wait, not just the one after. Only + // meaningful when a reconnect wait is pending (nextReconnectDelayMillis > 0). + // Deliberately does NOT advance the strategy: this recompute overwrites the + // just-computed delay from the fault-handling path; if we advanced the + // strategy here it would double-advance the counter for the next fault. + private void recomputeNextReconnectDelayForRegimeSwitch() { + // Only meaningful when we're in the between-fault-and-reconnect window. + if (!pendingReconnectWait) { + return; + } + RetryDelayStrategy.Result result = currentRetryDelayStrategy.apply(baseRetryDelayMillis); + nextReconnectDelayMillis = result.getDelayMillis(); + // Intentionally do NOT assign result.getNext() to currentRetryDelayStrategy. + // The prior fault's computeReconnectDelay() already advanced the counter + // for the impending reconnect; our job here is just to overwrite the + // stored delay value with what it should have been under the new regime. } /** @@ -405,6 +440,11 @@ private FaultEvent tryStart(boolean canReturnFaultEvent) throws StreamException connectionCloser.set(clientResult.getCloser()); origin = clientResult.getOrigin() == null ? client.getOrigin() : clientResult.getOrigin(); connectedTime = System.currentTimeMillis(); + // Clear the pending-reconnect flag now that the reconnect succeeded. The + // SDK-facing setInitialRetryDelayMillis / setMaxRetryDelayMillis methods + // use this flag to detect an in-flight reconnect wait and re-run the delay + // computation under a mid-flight regime change. + pendingReconnectWait = false; logger.debug("Connected to SSE stream"); ResponseHeaders headers = clientResult.getHeaders(); @@ -749,6 +789,7 @@ private void computeReconnectDelay() { if (result.getNext() != null) { currentRetryDelayStrategy = result.getNext(); } + pendingReconnectWait = true; } private boolean closeCurrentStream(boolean deliberatelyInterrupted, boolean shouldStopIterating) {