From 8fc09bd52bcbd9f960183ee0bf296b4e0522958f Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Thu, 20 Aug 2026 09:55:47 -0400 Subject: [PATCH] feat!: multi-strategy retry delay API for regime switching Introduces a multi-strategy retry-delay API on EventSource so SDKs adopting the LaunchDarkly RETRY specification can register normal- and extended-regime strategies at build time and activate between them at runtime. Server-directed retry: hints from the SSE wire remain sticky across activations, matching WHATWG "reconnection time is set until updated" semantics. RetryDelayStrategy is now a snapshot-oriented immutable value: each instance exposes getDelayMillis() for the current retry's delay and getNext() for the successor instance. The single previous method apply(long) is removed along with the Result wrapper class; the base delay is no longer an out-of-band parameter but lives on the strategy itself and is updated via the new withBaseDelayMillis(long) method (default no-op for custom strategies without a base concept). DefaultRetryDelayStrategy gains initialDelay(long, TimeUnit) so each strategy can carry its own initial delay, letting normal- and extended-regime strategies coexist with different starting points. EventSource.activateRetryDelayStrategy(RetryDelayStrategy) swaps the active strategy at runtime; each registered strategy retains its own backoff progression state across activations. Passing null or an unregistered strategy is a silent no-op. Builder.retryDelayStrategy has additive semantics: the first call sets the default (initially active and reset target); subsequent calls register additional strategies. The reconnect-delay compute is deferred to sleep time so activation and wire-hint changes received during the fault window take effect on the impending reconnect, not the one after. BREAKING CHANGE: RetryDelayStrategy.apply(long) is replaced by getDelayMillis() + getNext() + withBaseDelayMillis(long). The Result class is removed. Custom RetryDelayStrategy implementations must migrate to the new abstract shape. EventSource no longer exposes getBaseRetryDelayMillis() or getNextRetryDelayMillis(); observability is via the "Waiting X milliseconds before reconnecting" log message. --- .../DefaultRetryDelayStrategy.java | 117 +++++--- .../launchdarkly/eventsource/EventSource.java | 261 ++++++++++------- .../eventsource/RetryDelayStrategy.java | 82 +++--- .../DefaultRetryDelayStrategyTest.java | 182 +++++++----- .../eventsource/EventSourceBuilderTest.java | 4 +- .../eventsource/EventSourceReadingTest.java | 11 +- .../eventsource/EventSourceReconnectTest.java | 59 ++-- ...ventSourceRetryDelayStrategyUsageTest.java | 268 +++++++++++++++--- .../eventsource/RetryDelayStrategyTest.java | 22 ++ 9 files changed, 669 insertions(+), 337 deletions(-) create mode 100644 src/test/java/com/launchdarkly/eventsource/RetryDelayStrategyTest.java diff --git a/src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java b/src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java index 239438a..4cef38a 100644 --- a/src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java +++ b/src/main/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategy.java @@ -1,6 +1,6 @@ package com.launchdarkly.eventsource; -import java.security.SecureRandom; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import static com.launchdarkly.eventsource.Helpers.millisFromTimeUnit; @@ -9,59 +9,72 @@ * Default implementation of the retry delay strategy, providing exponential backoff * and jitter. *

- * The algorithm is as follows: - *

+ * Each instance is immutable: {@link #getDelayMillis()} returns the delay for this + * instance, and {@link #getNext()} returns the successor instance with the base + * delay multiplied by the backoff multiplier (pinned at the max delay). Jitter is + * rolled once per instance at construction so {@link #getDelayMillis()} is + * deterministic on a given instance. *

* This class is immutable. {@link RetryDelayStrategy#defaultStrategy()} returns the * default instance. To change any parameters, call methods which return a modified * instance: * 


  *     RetryDelayStrategy strategy = RetryDelayStrategy.defaultStrategy()
- *       .jitterMultiplier(0.25)
+ *       .initialDelay(1, TimeUnit.SECONDS)
+ *       .jitterMultiplier(0.25f)
  *       .maxDelay(20, TimeUnit.SECONDS);
  *
* * @since 4.0.0 */ public class DefaultRetryDelayStrategy extends RetryDelayStrategy { + /** + * The default value for {@link #initialDelay(long, TimeUnit)}: 1 second. + */ + public static final long DEFAULT_INITIAL_DELAY_MILLIS = 1000; + /** * The default value for {@link #maxDelay(long, TimeUnit)}: 30 seconds. */ public static final long DEFAULT_MAX_DELAY_MILLIS = 30000; - + /** * The default value for {@link #backoffMultiplier(float)}: 2. */ public static final float DEFAULT_BACKOFF_MULTIPLIER = 2; - + /** * The default value for {@link #jitterMultiplier(float)}: 0.5. */ public static final float DEFAULT_JITTER_MULTIPLIER = 0.5f; - static DefaultRetryDelayStrategy INSTANCE = new DefaultRetryDelayStrategy(0, + static final DefaultRetryDelayStrategy INSTANCE = new DefaultRetryDelayStrategy( + DEFAULT_INITIAL_DELAY_MILLIS, DEFAULT_MAX_DELAY_MILLIS, DEFAULT_BACKOFF_MULTIPLIER, DEFAULT_JITTER_MULTIPLIER); - - private final long lastBaseDelayMillis; + + final long baseDelayMillis; private final long maxDelayMillis; private final float backoffMultiplier; private final float jitterMultiplier; - private static final SecureRandom random = new SecureRandom(); - + private final long delayMillis; + + /** + * Returns a modified strategy with a specific initial (base) delay. The returned + * instance is fresh — its backoff progression is reset. + * + * @param initialDelay the initial delay in whatever time unit is specified by {@code timeUnit} + * @param timeUnit the time unit, or {@code TimeUnit.MILLISECONDS} if null + * @return a new instance with the specified initial delay + * @since 5.0.0 + * @see #DEFAULT_INITIAL_DELAY_MILLIS + */ + public DefaultRetryDelayStrategy initialDelay(long initialDelay, TimeUnit timeUnit) { + return new DefaultRetryDelayStrategy(millisFromTimeUnit(initialDelay, timeUnit), + this.maxDelayMillis, this.backoffMultiplier, this.jitterMultiplier); + } + /** * Returns a modified strategy with a specific maximum delay. * @@ -71,7 +84,7 @@ public class DefaultRetryDelayStrategy extends RetryDelayStrategy { * @see #DEFAULT_MAX_DELAY_MILLIS */ public DefaultRetryDelayStrategy maxDelay(long maxDelay, TimeUnit timeUnit) { - return new DefaultRetryDelayStrategy(lastBaseDelayMillis, + return new DefaultRetryDelayStrategy(this.baseDelayMillis, millisFromTimeUnit(maxDelay, timeUnit), this.backoffMultiplier, this.jitterMultiplier @@ -81,58 +94,68 @@ public DefaultRetryDelayStrategy maxDelay(long maxDelay, TimeUnit timeUnit) { /** * Returns a modified strategy with a specific backoff multipler. A multipler of 1 * means the base delay never changes, 2 means it doubles each time, etc. - * + * * @param newBackoffMultiplier the backoff multipler * @return a new instance with the specified backoff multiplier * @see #DEFAULT_BACKOFF_MULTIPLIER */ public DefaultRetryDelayStrategy backoffMultiplier(float newBackoffMultiplier) { - return new DefaultRetryDelayStrategy(0, this.maxDelayMillis, newBackoffMultiplier, this.jitterMultiplier); + return new DefaultRetryDelayStrategy(this.baseDelayMillis, this.maxDelayMillis, + newBackoffMultiplier, this.jitterMultiplier); } /** * Returns a modified strategy with a specific jitter multipler. A multipler of 0.5 * means each delay is reduced randomly by up to 50%, 0.25 means it is reduced * randomly by up to 25%, etc. Zero means there is no jitter. - * + * * @param newJitterMultiplier the jigger multipler * @return a new instance with the specified jitter multipler * @see #DEFAULT_JITTER_MULTIPLIER */ public DefaultRetryDelayStrategy jitterMultiplier(float newJitterMultiplier) { - return new DefaultRetryDelayStrategy(0, this.maxDelayMillis, this.backoffMultiplier, newJitterMultiplier); + return new DefaultRetryDelayStrategy(this.baseDelayMillis, this.maxDelayMillis, + this.backoffMultiplier, newJitterMultiplier); } - + private DefaultRetryDelayStrategy( - long lastBaseDelayMillis, + long baseDelayMillis, long maxDelayMillis, float backoffMultiplier, float jitterMultiplier ) { - this.lastBaseDelayMillis = lastBaseDelayMillis; + this.baseDelayMillis = baseDelayMillis; this.maxDelayMillis = maxDelayMillis; this.backoffMultiplier = backoffMultiplier; this.jitterMultiplier = jitterMultiplier; - } - - @Override - public Result apply(long baseDelayMillis) { - long nextBaseDelay = lastBaseDelayMillis == 0 ? baseDelayMillis : - (long)(lastBaseDelayMillis * backoffMultiplier); - if (maxDelayMillis > 0 && nextBaseDelay > maxDelayMillis) { - nextBaseDelay = maxDelayMillis; - } - long adjustedDelay = nextBaseDelay; - if (jitterMultiplier > 0) { + long adjustedDelay = baseDelayMillis; + if (jitterMultiplier > 0 && baseDelayMillis > 0) { // 2^31 milliseconds is much longer than any reconnect time we would reasonably want to use, so we can pin this to int - int maxTimeInt = nextBaseDelay > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int)nextBaseDelay; + int maxTimeInt = baseDelayMillis > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int)baseDelayMillis; int jitterRange = Math.round(maxTimeInt * jitterMultiplier); if (jitterRange > 0) { - adjustedDelay -= random.nextInt(jitterRange); + adjustedDelay -= ThreadLocalRandom.current().nextInt(jitterRange); } } - RetryDelayStrategy updatedStrategy = - new DefaultRetryDelayStrategy(nextBaseDelay, maxDelayMillis, backoffMultiplier, jitterMultiplier); - return new Result(adjustedDelay, updatedStrategy); + this.delayMillis = adjustedDelay; + } + + @Override + public long getDelayMillis() { + return delayMillis; + } + + @Override + public RetryDelayStrategy getNext() { + long nextBase = (long)(baseDelayMillis * backoffMultiplier); + if (maxDelayMillis > 0 && nextBase > maxDelayMillis) { + nextBase = maxDelayMillis; + } + return new DefaultRetryDelayStrategy(nextBase, maxDelayMillis, backoffMultiplier, jitterMultiplier); + } + + @Override + public DefaultRetryDelayStrategy withBaseDelayMillis(long millis) { + return new DefaultRetryDelayStrategy(millis, maxDelayMillis, backoffMultiplier, jitterMultiplier); } } diff --git a/src/main/java/com/launchdarkly/eventsource/EventSource.java b/src/main/java/com/launchdarkly/eventsource/EventSource.java index f5c6f94..0fe19bb 100644 --- a/src/main/java/com/launchdarkly/eventsource/EventSource.java +++ b/src/main/java/com/launchdarkly/eventsource/EventSource.java @@ -7,8 +7,12 @@ import java.io.IOException; import java.net.URI; import java.net.URL; +import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; @@ -72,27 +76,30 @@ public class EventSource implements Closeable { public static final int DEFAULT_READ_BUFFER_SIZE = 1000; // Note that some fields have package-private visibility for tests. - + private final Object sleepNotifier = new Object(); - + // The following final fields are set from the configuration builder. private final ConnectStrategy.Client client; final int readBufferSize; final ErrorStrategy baseErrorStrategy; - final RetryDelayStrategy baseRetryDelayStrategy; final long retryDelayResetThresholdMillis; final boolean streamEventData; final Set expectFields; - + // The following mutable fields are not volatile because they should only be // 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; + private final Map registeredStrategies; + final RetryDelayStrategy defaultRetryDelayStrategy; + volatile RetryDelayStrategy currentRetryDelayStrategy; + private volatile Long serverDirectedInitialDelayMillis = null; + // 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. @@ -104,13 +111,9 @@ public class EventSource implements Closeable { // and are read by the thread that is reading the stream. private volatile boolean deliberatelyClosedConnection; private volatile boolean calledStop; - - // These fields are written by the thread that is reading the stream, and can - // be read by other threads to inspect the state of the stream. - volatile long baseRetryDelayMillis; // set at config time but may be changed by a "retry:" value + private volatile String lastEventId; private volatile URI origin; - private volatile long nextReconnectDelayMillis; EventSource(Builder builder) { this.logger = builder.logger == null ? LDLogger.none() : builder.logger; @@ -119,10 +122,33 @@ public class EventSource implements Closeable { this.lastEventId = builder.lastEventId; this.baseErrorStrategy = this.currentErrorStrategy = builder.errorStrategy == null ? ErrorStrategy.alwaysThrow() : builder.errorStrategy; - this.baseRetryDelayStrategy = this.currentRetryDelayStrategy = - (builder.retryDelayStrategy == null ? RetryDelayStrategy.defaultStrategy() : - builder.retryDelayStrategy); - this.baseRetryDelayMillis = builder.retryDelayMillis; + // Assemble the retry-strategy registry from the builder. The default is always + // registered; additional strategies from the builder are also registered. + // + // Builder.retryDelay(long, TimeUnit) is a legacy shortcut that only applies when + // no caller-provided strategy is present. When the caller provides their own + // strategy, we preserve its reference identity (so activateRetryDelayStrategy + // works with the caller's original object) rather than mutating it via + // withBaseDelayMillis. Callers who want a specific initial on their custom + // strategy should set it directly on the strategy instance. + RetryDelayStrategy defaultStrategy; + if (builder.defaultRetryDelayStrategy != null) { + defaultStrategy = builder.defaultRetryDelayStrategy; + } else { + defaultStrategy = RetryDelayStrategy.defaultStrategy(); + if (builder.retryDelayMillisSet) { + defaultStrategy = defaultStrategy.withBaseDelayMillis(builder.retryDelayMillis); + } + } + this.defaultRetryDelayStrategy = defaultStrategy; + this.registeredStrategies = new HashMap<>(); + // the key is the original strategy, the value is the mutated strategy / state as + // operations progress + this.registeredStrategies.put(defaultStrategy, defaultStrategy); + for (RetryDelayStrategy s : builder.additionalRetryDelayStrategies) { + this.registeredStrategies.put(s, s); + } + this.currentRetryDelayStrategy = defaultStrategy; this.retryDelayResetThresholdMillis = builder.retryDelayResetThresholdMillis; this.streamEventData = builder.streamEventData; this.expectFields = builder.expectFields; @@ -176,43 +202,28 @@ public String getLastEventId() { } /** - * Returns the current base retry delay. - *

- * This is initially set by {@link Builder#retryDelay(long, TimeUnit)}, or - * {@link #DEFAULT_RETRY_DELAY_MILLIS} if not specified. It can be overriden by the - * stream provider if the stream contains a "retry:" line. - *

- * The actual retry delay for any given reconnection is computed by applying the - * configured {@link RetryDelayStrategy} to this value. - * - * @return the base retry delay in milliseconds - * @see #getNextRetryDelayMillis() - * @since 4.0.0 - */ - public long getBaseRetryDelayMillis() { - return baseRetryDelayMillis; - } - - /** - * Returns the retry delay that will be used for the next reconnection, if the - * stream has failed. + * Activates a previously-registered {@link RetryDelayStrategy}, making it the + * strategy used for subsequent reconnect delay computations. *

- * If you have just received a {@link StreamException} or {@link FaultEvent}, this - * value tells you how long EventSource will sleep before reconnecting, if you tell - * it to reconnect by calling {@link #start()} or by trying to read another event. - * The value is computed by applying the configured {@link RetryDelayStrategy} to - * the current value of {@link #getBaseRetryDelayMillis()}. + * The strategy must have been registered on the {@link Builder} via + * {@link Builder#retryDelayStrategy(RetryDelayStrategy)}. A null value or a + * strategy that was not registered on this EventSource is silently ignored. *

- * At any other time, the value is undefined. + * Each registered strategy carries its own backoff progression state. + * Activation is a pointer swap and does not reset the newly-activated + * strategy's counter; a strategy's state persists across activations. * - * @return the next retry delay in milliseconds - * @see #getBaseRetryDelayMillis() - * @since 4.0.0 + * @param strategy a strategy previously registered on the builder; a null value + * or a strategy not registered on this EventSource is treated as a no-op + * @since 5.0.0 */ - public long getNextRetryDelayMillis() { - return nextReconnectDelayMillis; + public void activateRetryDelayStrategy(RetryDelayStrategy strategy) { + if (strategy == null || !registeredStrategies.containsKey(strategy)) { + return; + } + currentRetryDelayStrategy = strategy; } - + /** * Attempts to start the stream if it is not already active. *

@@ -253,51 +264,54 @@ private FaultEvent tryStart(boolean canReturnFaultEvent) throws StreamException while (true) { StreamException exception = null; - - if (nextReconnectDelayMillis > 0) { - long delayNow = disconnectedTime == 0 ? nextReconnectDelayMillis : - (nextReconnectDelayMillis - (System.currentTimeMillis() - disconnectedTime)); - if (delayNow > 0) { - logger.info("Waiting {} milliseconds before reconnecting", delayNow); - try { - synchronized (sleepNotifier) { - if (!deliberatelyClosedConnection) { - sleepNotifier.wait(delayNow); - } - // If interrupt(), stop(), or close() is called while we're waiting, we will - // trigger an early exit from this wait by calling sleepNotifier.notify(). + + // Compute the reconnect delay just before sleep (rather than eagerly at + // fault time). Any strategy activation or wire retry hint received in + // the window between fault delivery and reconnect is honored on this + // reconnect, not the next one. + long reconnectDelayMillis = disconnectedTime != 0 ? computeReconnectDelay() : 0; + if (reconnectDelayMillis > 0) { + logger.info("Waiting {} milliseconds before reconnecting", reconnectDelayMillis); + try { + synchronized (sleepNotifier) { + if (!deliberatelyClosedConnection) { + sleepNotifier.wait(reconnectDelayMillis); } - } catch (InterruptedException e) { - // Thread.interrupt() should also have the effect of making us stop waiting - logger.debug("EventSource thread was interrupted during start()"); - deliberatelyClosedConnection = true; - Thread.interrupted(); // clear interrupted state - } - // Check if deliberatelyClosedConnection might have been set during that wait - if (deliberatelyClosedConnection) { - exception = new StreamClosedByCallerException(); + // If interrupt(), stop(), or close() is called while we're waiting, we will + // trigger an early exit from this wait by calling sleepNotifier.notify(). } + } catch (InterruptedException e) { + // Thread.interrupt() should also have the effect of making us stop waiting + logger.debug("EventSource thread was interrupted during start()"); + deliberatelyClosedConnection = true; + Thread.interrupted(); // clear interrupted state + } + // Check if deliberatelyClosedConnection might have been set during that wait. + // If so, the sleep was aborted by interrupt() -- we've observed it, so clear the + // flag so the next retry iteration is free to proceed to the connect attempt. + if (deliberatelyClosedConnection) { + exception = new StreamClosedByCallerException(); + deliberatelyClosedConnection = false; } } - + ConnectStrategy.Client.Result clientResult = null; - + if (exception == null) { readyState.set(ReadyState.CONNECTING); - + connectedTime = 0; deliberatelyClosedConnection = calledStop = false; - + try { clientResult = client.connect(lastEventId); } catch (StreamException e) { exception = e; } } - + if (exception != null) { disconnectedTime = System.currentTimeMillis(); - computeReconnectDelay(); if (applyErrorStrategy(exception) == ErrorStrategy.Action.CONTINUE) { // The ErrorStrategy told us to CONTINUE rather than throwing an exception. if (canReturnFaultEvent) { @@ -316,8 +330,8 @@ private FaultEvent tryStart(boolean canReturnFaultEvent) throws StreamException // The ErrorStrategy told us to THROW rather than CONTINUE. throw exception; } - - + + connectionCloser.set(clientResult.getCloser()); origin = clientResult.getOrigin() == null ? client.getOrigin() : clientResult.getOrigin(); connectedTime = System.currentTimeMillis(); @@ -605,9 +619,11 @@ private StreamEvent requireEvent() throws StreamException { StreamEvent event = eventParser.nextEvent(); if (event instanceof SetRetryDelayEvent) { // SetRetryDelayEvent means the stream contained a "retry:" line. We don't - // surface this to the caller, we just apply the new delay and move on. - baseRetryDelayMillis = ((SetRetryDelayEvent)event).getRetryMillis(); - resetRetryDelayStrategy(); + // surface this to the caller, we just apply the new base and move on. + // The new base is sticky across any subsequent activation via + // serverDirectedInitialDelayMillis. + serverDirectedInitialDelayMillis = ((SetRetryDelayEvent)event).getRetryMillis(); + resetAllRegisteredStrategyState(); continue; } if (event instanceof MessageEvent) { @@ -629,7 +645,6 @@ private StreamEvent requireEvent() throws StreamException { disconnectedTime = System.currentTimeMillis(); closeCurrentStream(false, false); eventParser = null; - computeReconnectDelay(); if (applyErrorStrategy(e) == ErrorStrategy.Action.CONTINUE) { // At this point we're handling errors from reading the stream (not initial connection), // so we never have HTTP response headers available (headers is always null) @@ -638,11 +653,6 @@ private StreamEvent requireEvent() throws StreamException { throw e; } } - - private void resetRetryDelayStrategy() { - logger.debug("Resetting retry delay strategy to initial state"); - currentRetryDelayStrategy = baseRetryDelayStrategy; - } private ErrorStrategy.Action applyErrorStrategy(StreamException e) { ErrorStrategy.Result errorStrategyResult = currentErrorStrategy.apply(e); @@ -651,19 +661,45 @@ private ErrorStrategy.Action applyErrorStrategy(StreamException e) { } return errorStrategyResult.getAction(); } - - private void computeReconnectDelay() { + + // Called just before sleeping at the top of tryStart()'s retry loop. Returns + // the delay for the impending reconnect and advances the active strategy's + // state via getNext() for the next fault. + private long computeReconnectDelay() { if (retryDelayResetThresholdMillis > 0 && connectedTime != 0) { long connectionDurationMillis = System.currentTimeMillis() - connectedTime; if (connectionDurationMillis >= retryDelayResetThresholdMillis) { - resetRetryDelayStrategy(); + // Healthy-op reset: the connection lasted long enough that we consider + // ourselves back to a "fresh" state. Revert active to the designated + // default strategy and zero every registered strategy's counter state. + logger.debug("Resetting retry delay strategy to initial state"); + currentRetryDelayStrategy = defaultRetryDelayStrategy; + resetAllRegisteredStrategyState(); } } - RetryDelayStrategy.Result result = - currentRetryDelayStrategy.apply(baseRetryDelayMillis); - nextReconnectDelayMillis = result.getDelayMillis(); - if (result.getNext() != null) { - currentRetryDelayStrategy = result.getNext(); + RetryDelayStrategy current = currentRetryDelayStrategy; + RetryDelayStrategy advanced = registeredStrategies.get(current); + registeredStrategies.put(current, advanced.getNext()); + return advanced.getDelayMillis(); + } + + // Package-private accessor: returns the current advanced instance for the + // currently-active registered strategy. Used by tests to observe the retry state. + RetryDelayStrategy currentRetryStrategySnapshot() { + return registeredStrategies.get(currentRetryDelayStrategy); + } + + // Reset each registered strategy's backoff progression. If a wire retry hint has + // been received, the reset instance uses the wire base; otherwise it reverts to + // the caller's originally-registered instance. This preserves the WHATWG-sticky + // wire override across healthy-op resets. + private void resetAllRegisteredStrategyState() { + for (Map.Entry e : registeredStrategies.entrySet()) { + RetryDelayStrategy fresh = e.getKey(); + if (serverDirectedInitialDelayMillis != null) { + fresh = fresh.withBaseDelayMillis(serverDirectedInitialDelayMillis); + } + e.setValue(fresh); } } @@ -708,8 +744,10 @@ private boolean closeCurrentStream(boolean deliberatelyInterrupted, boolean shou public static final class Builder { private final ConnectStrategy connectStrategy; // final because it's mandatory, set at constructor time private ErrorStrategy errorStrategy; - private RetryDelayStrategy retryDelayStrategy; - private long retryDelayMillis = DEFAULT_RETRY_DELAY_MILLIS; + RetryDelayStrategy defaultRetryDelayStrategy; + final List additionalRetryDelayStrategies = new ArrayList<>(); + long retryDelayMillis = DEFAULT_RETRY_DELAY_MILLIS; + boolean retryDelayMillisSet = false; private long retryDelayResetThresholdMillis = DEFAULT_RETRY_DELAY_RESET_THRESHOLD_MILLIS; private String lastEventId; private int readBufferSize = DEFAULT_READ_BUFFER_SIZE; @@ -860,29 +898,40 @@ public Builder lastEventId(String lastEventId) { */ public Builder retryDelay(long retryDelay, TimeUnit timeUnit) { retryDelayMillis = millisFromTimeUnit(retryDelay, timeUnit); + retryDelayMillisSet = true; return this; } /** - * Specifies a strategy for determining the retry delay after an error. + * Configures the retry-delay strategies available to the EventSource. *

* Whenever EventSource tries to start a new connection after a stream failure, - * it delays for an amount of time that is determined by two parameters: the - * base retry delay ({@link #retryDelay(long, TimeUnit)}), and the retry delay - * strategy which transforms the base retry delay in some way. The default behavior - * is to apply an exponential backoff and jitter. You may instead use a modified - * version of {@link DefaultRetryDelayStrategy} to customize the backoff and - * jitter, or a custom implementation with any other logic. - * - * @param retryDelayStrategy the object that will control retry delays; if null, - * defaults to {@link RetryDelayStrategy#defaultStrategy()} + * it delays for an amount of time determined by two parameters: the base + * retry delay ({@link #retryDelay(long, TimeUnit)}) and a + * {@link RetryDelayStrategy} that transforms it. The default behavior is + * exponential backoff with jitter. + *

+ * First call sets the default strategy — the strategy that + * is initially active and that the healthy-op reset returns to. If never + * called, the default is {@link RetryDelayStrategy#defaultStrategy()}. + *

+ * Subsequent calls register additional strategies. These are not + * initially active; they become available for runtime activation via + * {@link EventSource#activateRetryDelayStrategy(RetryDelayStrategy)}. + * + * @param retryDelayStrategy the strategy to configure * @return the builder * @see #retryDelay(long, TimeUnit) * @see #retryDelayResetThreshold(long, TimeUnit) + * @see EventSource#activateRetryDelayStrategy(RetryDelayStrategy) * @since 4.0.0 */ public Builder retryDelayStrategy(RetryDelayStrategy retryDelayStrategy) { - this.retryDelayStrategy = retryDelayStrategy; + if (this.defaultRetryDelayStrategy == null) { + this.defaultRetryDelayStrategy = retryDelayStrategy; + } else if (retryDelayStrategy != null) { + this.additionalRetryDelayStrategies.add(retryDelayStrategy); + } return this; } diff --git a/src/main/java/com/launchdarkly/eventsource/RetryDelayStrategy.java b/src/main/java/com/launchdarkly/eventsource/RetryDelayStrategy.java index d417ee0..2cef615 100644 --- a/src/main/java/com/launchdarkly/eventsource/RetryDelayStrategy.java +++ b/src/main/java/com/launchdarkly/eventsource/RetryDelayStrategy.java @@ -10,62 +10,54 @@ * generally a best practice to use backoff and jitter, to avoid a reconnect storm * during a service interruption. *

- * Implementations of this interface should be immutable. To implement strategies where - * the delay uses different parameters on each subsequent retry (such as exponential - * backoff), the strategy should return a new instance of its own class in - * {@link RetryDelayStrategy.Result#getNext()}, rather than modifying the state of the - * existing instance. This makes it easy for EventSource to reset to the original delay - * state when appropriate by simply reusing the original instance. + * Implementations should be immutable. Each instance represents a single state in the + * retry-delay sequence: {@link #getDelayMillis()} returns the delay to use for the + * impending retry, and {@link #getNext()} returns the strategy instance to use for + * the retry after that. Strategies with a base-delay concept may also implement + * {@link #withBaseDelayMillis(long)} to accept server-directed reconnection-time + * overrides from the SSE {@code retry:} field. * * @since 4.0.0 */ public abstract class RetryDelayStrategy { /** - * The return type of {@link RetryDelayStrategy#apply(long)}. + * Returns the retry delay this instance represents, in milliseconds. Pure and + * deterministic on a given instance. + * + * @return the delay in milliseconds + * @since 5.0.0 */ - public static class Result { - private final long delayMillis; - private final RetryDelayStrategy next; - - /** - * Constructs an instance. - * - * @param delayMillis the computed delay in milliseconds - * @param next a {@link RetryDelayStrategy} instance to be used for the next retry; - * null means to use the same instance as last time - */ - public Result(long delayMillis, RetryDelayStrategy next) { - this.delayMillis = delayMillis; - this.next = next; - } + public abstract long getDelayMillis(); - /** - * Returns the computed delay. - * @return the delay in milliseconds - */ - public long getDelayMillis() { - return delayMillis; - } + /** + * Returns the strategy instance to use for the retry after this one. Does not + * modify this instance. + *

+ * Strategies that never advance (e.g., a constant-delay strategy) return + * {@code this}. Strategies with backoff progression return a new instance + * carrying the advanced state. + * + * @return the strategy to use next + * @since 5.0.0 + */ + public abstract RetryDelayStrategy getNext(); - /** - * Returns the strategy instance to be used for the next retry, or null to use the - * same instance as last time. - * @return a new instance or null - */ - public RetryDelayStrategy getNext() { - return next; - } - } - /** - * Applies the strategy to compute the appropriate retry delay. + * Returns a fresh instance of this strategy with its base delay set to the given + * value and any backoff progression reset. + *

+ * The default implementation returns {@code this}. Strategies without a base-delay + * concept opt out of wire-directed base overrides by not implementing this method. * - * @param baseDelayMillis the initial configured base delay as set by - * {@link EventSource.Builder#retryDelay(long, java.util.concurrent.TimeUnit)} - * @return the computed delay + * @param millis the new base delay in milliseconds + * @return a fresh instance with the given base, or {@code this} if the strategy + * does not honor base overrides + * @since 5.0.0 */ - public abstract Result apply(long baseDelayMillis); - + public RetryDelayStrategy withBaseDelayMillis(long millis) { + return this; + } + /** * Returns the default implementation, configured to use the default backoff and * jitter. diff --git a/src/test/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategyTest.java b/src/test/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategyTest.java index 7b15cb1..e6f207c 100644 --- a/src/test/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategyTest.java +++ b/src/test/java/com/launchdarkly/eventsource/DefaultRetryDelayStrategyTest.java @@ -17,23 +17,23 @@ public void backoffWithNoJitterAndNoMax() { long base = 4; RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(base, TimeUnit.MILLISECONDS) .backoffMultiplier(2).jitterMultiplier(0) .maxDelay(0, null); - RetryDelayStrategy.Result r1 = s.apply(base); - assertThat(r1.getDelayMillis(), equalTo(base)); - - RetryDelayStrategy.Result r2 = r1.getNext().apply(base); - assertThat(r2.getDelayMillis(), equalTo(base * 2)); - - RetryDelayStrategy.Result r3 = r2.getNext().apply(base); - assertThat(r3.getDelayMillis(), equalTo(base * 4)); - - RetryDelayStrategy.Result r4 = r3.getNext().apply(base); - assertThat(r4.getDelayMillis(), equalTo(base * 8)); - - RetryDelayStrategy.Result r5 = r4.getNext().apply(base); - assertThat(r5.getDelayMillis(), equalTo(base * 16)); + assertThat(s.getDelayMillis(), equalTo(base)); + + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(base * 2)); + + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(base * 4)); + + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(base * 8)); + + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(base * 16)); } @Test @@ -42,38 +42,36 @@ public void backoffWithNoJitterAndMax() { long max = base * 4 + 3; RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(base, TimeUnit.MILLISECONDS) .backoffMultiplier(2).jitterMultiplier(0) .maxDelay(max, TimeUnit.MILLISECONDS); - RetryDelayStrategy.Result r1 = s.apply(base); - assertThat(r1.getDelayMillis(), equalTo(base)); - - RetryDelayStrategy.Result r2 = r1.getNext().apply(base); - assertThat(r2.getDelayMillis(), equalTo(base * 2)); - - RetryDelayStrategy.Result r3 = r2.getNext().apply(base); - assertThat(r3.getDelayMillis(), equalTo(base * 4)); - - RetryDelayStrategy.Result r4 = r3.getNext().apply(base); - assertThat(r4.getDelayMillis(), equalTo(max)); + assertThat(s.getDelayMillis(), equalTo(base)); + + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(base * 2)); + + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(base * 4)); + + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(max)); } - + @Test public void noBackoffAndNoJitter() { long base = 4; RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(base, TimeUnit.MILLISECONDS) .backoffMultiplier(1).jitterMultiplier(0) .maxDelay(0, null); - RetryDelayStrategy.Result r1 = s.apply(base); - assertThat(r1.getDelayMillis(), equalTo(base)); - - RetryDelayStrategy.Result r2 = r1.getNext().apply(base); - assertThat(r2.getDelayMillis(), equalTo(base)); - - RetryDelayStrategy.Result r3 = r2.getNext().apply(base); - assertThat(r3.getDelayMillis(), equalTo(base)); + assertThat(s.getDelayMillis(), equalTo(base)); + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(base)); + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(base)); } @Test @@ -84,69 +82,99 @@ public void backoffWithJitter() { float specifiedJitter = 0.25f; RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(base, TimeUnit.MILLISECONDS) .backoffMultiplier(specifiedBackoff).jitterMultiplier(specifiedJitter) .maxDelay(max, TimeUnit.MILLISECONDS); - - RetryDelayStrategy.Result r1 = - verifyJitter(s, base, base, specifiedJitter); - - RetryDelayStrategy.Result r2 = - verifyJitter(r1.getNext(), base, base * specifiedBackoff, specifiedJitter); - RetryDelayStrategy.Result r3 = - verifyJitter(r2.getNext(), base, base * specifiedBackoff * specifiedBackoff, specifiedJitter); - - verifyJitter(r3.getNext(), base, max, specifiedJitter); + s = verifyJitter(s, base, specifiedJitter); + s = verifyJitter(s, base * specifiedBackoff, specifiedJitter); + s = verifyJitter(s, base * specifiedBackoff * specifiedBackoff, specifiedJitter); + verifyJitter(s, max, specifiedJitter); } @Test public void zeroBaseDelayAlwaysProducesZero() { - RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy(); + RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(0, TimeUnit.MILLISECONDS); for (int i = 0; i < 5; i++) { - RetryDelayStrategy.Result r = s.apply(0); - assertThat(r.getDelayMillis(), equalTo(0L)); - s = r.getNext(); + assertThat(s.getDelayMillis(), equalTo(0L)); + s = s.getNext(); } } - - private RetryDelayStrategy.Result verifyJitter( + + @Test + public void withBaseDelayMillisOverridesAndResetsProgression() { + long initialBase = 100; + long overrideBase = 500; + + RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(initialBase, TimeUnit.MILLISECONDS) + .backoffMultiplier(2).jitterMultiplier(0) + .maxDelay(0, null); + + // Advance a few steps. + s = s.getNext(); + s = s.getNext(); + // Now at 400 (100 * 2 * 2). + assertThat(s.getDelayMillis(), equalTo(initialBase * 4)); + + // Override the base; expect a fresh snapshot at the new base. + s = s.withBaseDelayMillis(overrideBase); + assertThat(s.getDelayMillis(), equalTo(overrideBase)); + + // Advance from the fresh snapshot. + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo(overrideBase * 2)); + } + + // Verifies that a strategy's getDelayMillis() sits in the expected jitter range + // around baseWithBackoff, and returns the getNext() strategy for chained + // verification. Because each snapshot's jitter is rolled once at construction + // (deterministic per instance), we sample 100 fresh withBaseDelayMillis + // reconstructions to confirm the range and that the values aren't all identical. + private RetryDelayStrategy verifyJitter( RetryDelayStrategy s, - long base, long baseWithBackoff, float expectedJitterRatio ) { - // We can't 100% prove that it's using the expected jitter ratio, since the result - // is pseudo-random, but we can at least prove that repeated computations don't - // fall outside the expected range and aren't all equal. - RetryDelayStrategy.Result lastResult = null; + long firstDelay = s.getDelayMillis(); + assertThat(firstDelay, allOf( + greaterThanOrEqualTo((long)(baseWithBackoff * expectedJitterRatio)), + lessThanOrEqualTo(baseWithBackoff) + )); + + // Sample additional jittered values via withBaseDelayMillis() (each call + // reconstructs with a fresh jitter roll). boolean atLeastOneWasDifferent = false; for (int i = 0; i < 100; i++) { - RetryDelayStrategy.Result result = s.apply(base); - assertThat(result.getDelayMillis(), allOf( + RetryDelayStrategy sampled = s.withBaseDelayMillis(baseWithBackoff); + long delay = sampled.getDelayMillis(); + assertThat(delay, allOf( greaterThanOrEqualTo((long)(baseWithBackoff * expectedJitterRatio)), lessThanOrEqualTo(baseWithBackoff) - )); - if (lastResult != null && !atLeastOneWasDifferent) { - atLeastOneWasDifferent = result.getDelayMillis() != lastResult.getDelayMillis(); + )); + if (delay != firstDelay) { + atLeastOneWasDifferent = true; } - lastResult = result; } - return lastResult; + // (Not asserting atLeastOneWasDifferent strictly to avoid flakes on very small + // baseWithBackoff values, but it should virtually always be true.) + return s.getNext(); } - + @Test public void defaultBackoff() { long base = 4; RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(base, TimeUnit.MILLISECONDS) .jitterMultiplier(0).maxDelay(100, TimeUnit.SECONDS); - - RetryDelayStrategy.Result r1 = s.apply(base); - assertThat(r1.getDelayMillis(), equalTo(base)); - - RetryDelayStrategy.Result r2 = r1.getNext().apply(base); - assertThat(r2.getDelayMillis(), equalTo((long) + + assertThat(s.getDelayMillis(), equalTo(base)); + + s = s.getNext(); + assertThat(s.getDelayMillis(), equalTo((long) (base * DefaultRetryDelayStrategy.DEFAULT_BACKOFF_MULTIPLIER))); } @@ -155,8 +183,20 @@ public void defaultJitter() { long base = 4; RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(base, TimeUnit.MILLISECONDS) .maxDelay(100, TimeUnit.SECONDS); - - verifyJitter(s, base, base, DefaultRetryDelayStrategy.DEFAULT_JITTER_MULTIPLIER); + + verifyJitter(s, base, DefaultRetryDelayStrategy.DEFAULT_JITTER_MULTIPLIER); + } + + @Test + public void tinyBaseWithSmallJitterProducesNoJitter() { + // When base * jitterMultiplier rounds below 1, jitter is effectively disabled + // (the jitter subtraction would be zero). Verify this edge is handled without + // throwing (SecureRandom.nextInt(0) would throw IllegalArgumentException). + RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy() + .initialDelay(1, TimeUnit.MILLISECONDS) + .jitterMultiplier(0.4f); + assertThat(s.getDelayMillis(), equalTo(1L)); } } diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceBuilderTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceBuilderTest.java index c490e70..53c6fd1 100644 --- a/src/test/java/com/launchdarkly/eventsource/EventSourceBuilderTest.java +++ b/src/test/java/com/launchdarkly/eventsource/EventSourceBuilderTest.java @@ -82,13 +82,13 @@ public void httpUrlCannotBeNull() { @Test public void retryDelayStrategy() { try (EventSource es = builder.build()) { - assertThat(es.baseRetryDelayStrategy, sameInstance(RetryDelayStrategy.defaultStrategy())); + assertThat(es.defaultRetryDelayStrategy, sameInstance(RetryDelayStrategy.defaultStrategy())); assertThat(es.currentRetryDelayStrategy, sameInstance(RetryDelayStrategy.defaultStrategy())); } RetryDelayStrategy customStrategy = RetryDelayStrategy.defaultStrategy().backoffMultiplier(3); try (EventSource es = builder.retryDelayStrategy(customStrategy).build()) { - assertThat(es.baseRetryDelayStrategy, sameInstance(customStrategy)); + assertThat(es.defaultRetryDelayStrategy, sameInstance(customStrategy)); assertThat(es.currentRetryDelayStrategy, sameInstance(customStrategy)); } } diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceReadingTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceReadingTest.java index cf87245..2555571 100644 --- a/src/test/java/com/launchdarkly/eventsource/EventSourceReadingTest.java +++ b/src/test/java/com/launchdarkly/eventsource/EventSourceReadingTest.java @@ -44,7 +44,8 @@ public void expectedStateBeforeStart() throws Exception { assertThat(es.getState(), equalTo(ReadyState.RAW)); assertThat(es.getOrigin(), equalTo(ORIGIN)); assertThat(es.getLastEventId(), nullValue()); - assertThat(es.getBaseRetryDelayMillis(), equalTo(EventSource.DEFAULT_RETRY_DELAY_MILLIS)); + assertThat(((DefaultRetryDelayStrategy) es.defaultRetryDelayStrategy).baseDelayMillis, + equalTo(EventSource.DEFAULT_RETRY_DELAY_MILLIS)); } } @@ -59,7 +60,8 @@ public void expectedStateAfterStart() throws Exception { assertThat(es.getState(), equalTo(ReadyState.OPEN)); assertThat(es.getOrigin(), equalTo(ORIGIN)); assertThat(es.getLastEventId(), nullValue()); - assertThat(es.getBaseRetryDelayMillis(), equalTo(EventSource.DEFAULT_RETRY_DELAY_MILLIS)); + assertThat(((DefaultRetryDelayStrategy) es.defaultRetryDelayStrategy).baseDelayMillis, + equalTo(EventSource.DEFAULT_RETRY_DELAY_MILLIS)); } } @@ -218,7 +220,7 @@ public void initialRetryDelayIsSetFromBuilder() throws Exception { MockConnectStrategy mock = new MockConnectStrategy(); try (EventSource es = baseBuilder(mock).retryDelay(6, TimeUnit.SECONDS).build()) { - assertEquals(6000, es.getBaseRetryDelayMillis()); + assertEquals(6000, ((DefaultRetryDelayStrategy) es.defaultRetryDelayStrategy).baseDelayMillis); } } @@ -243,7 +245,8 @@ public void retryDelayIsUpdatedFromEvent() throws Exception { assertThat(es.readAnyEvent(), equalTo( new MessageEvent("message", eventData, null, ORIGIN))); - assertEquals(300, es.getBaseRetryDelayMillis()); + // Wire retry hint updates the current active strategy's snapshot with the new base. + assertEquals(300, ((DefaultRetryDelayStrategy) es.currentRetryStrategySnapshot()).baseDelayMillis); } } diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceReconnectTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceReconnectTest.java index 47b1589..098b60a 100644 --- a/src/test/java/com/launchdarkly/eventsource/EventSourceReconnectTest.java +++ b/src/test/java/com/launchdarkly/eventsource/EventSourceReconnectTest.java @@ -100,30 +100,20 @@ public void retryDelayIsTerminatedEarlyIfEventSourceInterruptIsCalled() throws E respondWithDataAndThenEnd("data: first\n\n"), respondWithStream()); - AtomicInteger counter = new AtomicInteger(0); long longDelay = 5000, tinyDelay = 1; - RetryDelayStrategy longDelayForFirstRetryOnly = new RetryDelayStrategy() { - @Override - public Result apply(long baseDelayMillis) { - return new Result( - counter.getAndIncrement() == 0 ? longDelay : tinyDelay, - null); - } - }; - + RetryDelayStrategy longDelayForFirstRetryOnly = new TwoStageDelayStrategy(longDelay, tinyDelay); + try (EventSource es = baseBuilder(mock) .retryDelayStrategy(longDelayForFirstRetryOnly) .build()) { assertThat(es.getState(), equalTo(ReadyState.RAW)); - + es.start(); - + assertThat(es.readAnyEvent(), equalTo(new MessageEvent("message", "first", null, ORIGIN))); assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); - assertThat(es.getNextRetryDelayMillis(), equalTo(longDelay)); - long timeBeforeRetrying = System.currentTimeMillis(); interruptOnAnotherThreadAfterDelay(es, 100); es.start(); @@ -140,16 +130,8 @@ public void retryDelayIsTerminatedEarlyIfThreadInterruptIsCalled() throws Except respondWithDataAndThenEnd("data: first\n\n"), respondWithStream()); - AtomicInteger counter = new AtomicInteger(0); long longDelay = 5000, tinyDelay = 1; - RetryDelayStrategy longDelayForFirstRetryOnly = new RetryDelayStrategy() { - @Override - public Result apply(long baseDelayMillis) { - return new Result( - counter.getAndIncrement() == 0 ? longDelay : tinyDelay, - null); - } - }; + RetryDelayStrategy longDelayForFirstRetryOnly = new TwoStageDelayStrategy(longDelay, tinyDelay); try (EventSource es = baseBuilder(mock) .retryDelayStrategy(longDelayForFirstRetryOnly) @@ -162,8 +144,6 @@ public Result apply(long baseDelayMillis) { assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); - assertThat(es.getNextRetryDelayMillis(), equalTo(longDelay)); - long timeBeforeRetrying = System.currentTimeMillis(); interruptThisThreadFromAnotherThreadAfterDelay(100); es.start(); @@ -172,4 +152,33 @@ public Result apply(long baseDelayMillis) { assertThat(actualDuration, Matchers.lessThan(longDelay)); } } + + // Test strategy that returns one delay on the first attempt and a different + // delay for all subsequent attempts. Immutable — the "first vs subsequent" + // distinction is captured by two chained snapshots. + private static final class TwoStageDelayStrategy extends RetryDelayStrategy { + private final long firstDelay; + private final long subsequentDelay; + private final boolean isFirst; + + TwoStageDelayStrategy(long firstDelay, long subsequentDelay) { + this(firstDelay, subsequentDelay, true); + } + + private TwoStageDelayStrategy(long firstDelay, long subsequentDelay, boolean isFirst) { + this.firstDelay = firstDelay; + this.subsequentDelay = subsequentDelay; + this.isFirst = isFirst; + } + + @Override + public long getDelayMillis() { + return isFirst ? firstDelay : subsequentDelay; + } + + @Override + public RetryDelayStrategy getNext() { + return new TwoStageDelayStrategy(firstDelay, subsequentDelay, false); + } + } } diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java index 30de94b..352fbb5 100644 --- a/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java +++ b/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java @@ -36,10 +36,15 @@ private EventSource.Builder baseBuilder(MockConnectStrategy mock) { .logger(testLogger.getLogger()); } - private void expectReconnectingLogMessage() { + // Consumes the "Waiting X milliseconds before reconnecting" log message emitted + // by EventSource just before the sleep, and returns the value of X. Since the + // log now emits the strategy's computed delay (no elapsed-time subtraction), + // tests can assert exact equality against the strategy's expected delay. + private long readReconnectDelayFromLog() { LogCapture.Message m = testLogger.getLogCapture().requireMessage(LDLogLevel.INFO, 1000); assertThat(m.getText(), allOf( - startsWith("Waiting"), endsWith(("milliseconds before reconnecting")))); + startsWith("Waiting"), endsWith("milliseconds before reconnecting"))); + return Long.parseLong(m.getText().split(" ")[1]); } @Test @@ -54,15 +59,15 @@ public void nextRetryDelayStrategyIsAppliedEachTime() throws Exception { mock.configureRequests(respondWithStream()); // leave stream open after last retry int increment = 3; - RetryDelayStrategy retryDelayStrategy = new ArithmeticallyIncreasingRetryDelayStrategy(increment, 0); + RetryDelayStrategy retryDelayStrategy = + new ArithmeticallyIncreasingRetryDelayStrategy(initialDelay, increment, 0); try (EventSource es = baseBuilder(mock) .retryDelayStrategy(retryDelayStrategy) - .retryDelay(initialDelay, null) .logger(testLogger.getLogger()) .build()) { es.start(); - + for (int i = 0; i < attempts; i++) { assertThat(es.readAnyEvent(), equalTo(new MessageEvent( "message", "event" + i, null, ORIGIN))); @@ -71,9 +76,7 @@ public void nextRetryDelayStrategyIsAppliedEachTime() throws Exception { assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); - expectReconnectingLogMessage(); - - assertThat(es.getNextRetryDelayMillis(), equalTo(initialDelay + (increment * i))); + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + (increment * i))); } } } @@ -90,11 +93,10 @@ public void sameRetryDelayStrategyIsReusedIfItReturnsNoNextOne() throws Exceptio mock.configureRequests(respondWithStream()); // leave stream open after last retry int increment = 3; - RetryDelayStrategy retryDelayStrategy = new FixedRetryDelayStrategy(increment); + RetryDelayStrategy retryDelayStrategy = new FixedRetryDelayStrategy(initialDelay, increment); try (EventSource es = baseBuilder(mock) .retryDelayStrategy(retryDelayStrategy) - .retryDelay(initialDelay, null) .logger(testLogger.getLogger()) .build()) { es.start(); @@ -107,9 +109,7 @@ public void sameRetryDelayStrategyIsReusedIfItReturnsNoNextOne() throws Exceptio assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); - expectReconnectingLogMessage(); - - assertThat(es.getNextRetryDelayMillis(), equalTo(initialDelay + increment)); + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + increment)); } } } @@ -126,74 +126,268 @@ public void retryDelayStrategyIsResetAfterThreshold() throws Exception { long initialDelay = 10; long threshold = 50; int increment = 3; - RetryDelayStrategy retryDelayStrategy = new ArithmeticallyIncreasingRetryDelayStrategy(increment, 0); + RetryDelayStrategy retryDelayStrategy = + new ArithmeticallyIncreasingRetryDelayStrategy(initialDelay, increment, 0); try (EventSource es = baseBuilder(mock) .retryDelayStrategy(retryDelayStrategy) - .retryDelay(initialDelay, null) .retryDelayResetThreshold(threshold, null) .logger(testLogger.getLogger()) .build()) { es.start(); stream1.close(); - + // On first failure, the delay is the initial delay assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); - assertThat(es.getNextRetryDelayMillis(), equalTo(initialDelay)); assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay)); stream2.close(); // On second failure, the delay is incremented because it happened sooner than the threshold assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); - assertThat(es.getNextRetryDelayMillis(), equalTo(initialDelay + increment)); assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + increment)); Thread.sleep(threshold + 10); stream3.close(); // This time, the stream lasted longer than the threshold so we reset to the initial delay assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); - assertThat(es.getNextRetryDelayMillis(), equalTo(initialDelay)); assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); - + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay)); + stream4.close(); - + // And now this time, the stream did not last long enough so the delay gets incremented assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); - assertThat(es.getNextRetryDelayMillis(), equalTo(initialDelay + increment)); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + increment)); } } private static class ArithmeticallyIncreasingRetryDelayStrategy extends RetryDelayStrategy { + private final long baseDelayMillis; private final int increment; private final int counter; - - ArithmeticallyIncreasingRetryDelayStrategy(int increment, int counter) { + + ArithmeticallyIncreasingRetryDelayStrategy(long baseDelayMillis, int increment, int counter) { + this.baseDelayMillis = baseDelayMillis; this.increment = increment; this.counter = counter; } - + + ArithmeticallyIncreasingRetryDelayStrategy(int increment) { + this(0, increment, 0); + } + @Override - public Result apply(long baseDelayMillis) { - return new Result( - baseDelayMillis + (counter * increment), - new ArithmeticallyIncreasingRetryDelayStrategy(increment, counter + 1) - ); + public long getDelayMillis() { + return baseDelayMillis + (counter * increment); + } + + @Override + public RetryDelayStrategy getNext() { + return new ArithmeticallyIncreasingRetryDelayStrategy(baseDelayMillis, increment, counter + 1); + } + + @Override + public RetryDelayStrategy withBaseDelayMillis(long millis) { + return new ArithmeticallyIncreasingRetryDelayStrategy(millis, increment, 0); } } - + private static class FixedRetryDelayStrategy extends RetryDelayStrategy { + private final long baseDelayMillis; private final int increment; - - FixedRetryDelayStrategy(int increment) { + + FixedRetryDelayStrategy(long baseDelayMillis, int increment) { + this.baseDelayMillis = baseDelayMillis; this.increment = increment; } - + + FixedRetryDelayStrategy(int increment) { + this(0, increment); + } + @Override - public Result apply(long baseDelayMillis) { - return new Result(baseDelayMillis + increment, null); + public long getDelayMillis() { + return baseDelayMillis + increment; } - } + + @Override + public RetryDelayStrategy getNext() { + return this; + } + + @Override + public RetryDelayStrategy withBaseDelayMillis(long millis) { + return new FixedRetryDelayStrategy(millis, increment); + } + } + + // Tests for activateRetryDelayStrategy: the SDK-side entry point for regime + // switching per the LaunchDarkly RETRY spec. Strategies are registered at build + // time via repeated calls to retryDelayStrategy(); the first call sets the + // default (reset target) and subsequent calls register additional strategies + // available for runtime activation. + + @Test + public void activateRetryDelayStrategyNullIsNoOp() throws Exception { + MockConnectStrategy mock = new MockConnectStrategy(); + mock.configureRequests(respondWithStream()); + try (EventSource es = baseBuilder(mock).build()) { + es.start(); + RetryDelayStrategy before = es.currentRetryDelayStrategy; + es.activateRetryDelayStrategy(null); + // No throw, no state change. + assertThat(es.currentRetryDelayStrategy, equalTo(before)); + } + } + + @Test + public void activateRetryDelayStrategyUnregisteredIsNoOp() throws Exception { + MockConnectStrategy mock = new MockConnectStrategy(); + mock.configureRequests(respondWithStream()); + try (EventSource es = baseBuilder(mock).build()) { + es.start(); + RetryDelayStrategy before = es.currentRetryDelayStrategy; + es.activateRetryDelayStrategy(new FixedRetryDelayStrategy(100)); + // No throw, no state change. + assertThat(es.currentRetryDelayStrategy, equalTo(before)); + } + } + + @Test + public void activateRetryDelayStrategySwapsTheActiveStrategy() throws Exception { + MockConnectStrategy mock = new MockConnectStrategy(); + PipedStreamRequestHandler stream1 = respondWithStream(); + PipedStreamRequestHandler stream2 = respondWithStream(); + PipedStreamRequestHandler stream3 = respondWithStream(); + mock.configureRequests(stream1, stream2, stream3); + + long initialDelay = 10; + int normalIncrement = 3, extendedIncrement = 100; + // Bake initial into each strategy explicitly, since Builder.retryDelay(x) applies + // only to the default strategy (via withBaseDelayMillis at build time). + RetryDelayStrategy normal = new FixedRetryDelayStrategy(initialDelay, normalIncrement); + RetryDelayStrategy extended = new FixedRetryDelayStrategy(initialDelay, extendedIncrement); + try (EventSource es = baseBuilder(mock) + .retryDelayStrategy(normal) // first call = default + .retryDelayStrategy(extended) // second call = additional + .build()) { + es.start(); + + stream1.close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + // Default (normal) is active. + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + normalIncrement)); + + // Swap to the extended strategy. + es.activateRetryDelayStrategy(extended); + + stream2.close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + extendedIncrement)); + } + } + + @Test + public void healthyOpResetRevertsToDefaultStrategy() throws Exception { + // After healthy-op reset threshold elapses, active reverts to the default + // (first-registered) strategy. + MockConnectStrategy mock = new MockConnectStrategy(); + PipedStreamRequestHandler stream1 = respondWithStream(); + PipedStreamRequestHandler stream2 = respondWithStream(); + PipedStreamRequestHandler stream3 = respondWithStream(); + mock.configureRequests(stream1, stream2, stream3); + + long initialDelay = 10; + long threshold = 50; + int normalIncrement = 3, extendedIncrement = 100; + RetryDelayStrategy normal = new FixedRetryDelayStrategy(initialDelay, normalIncrement); + RetryDelayStrategy extended = new FixedRetryDelayStrategy(initialDelay, extendedIncrement); + try (EventSource es = baseBuilder(mock) + .retryDelayStrategy(normal) + .retryDelayStrategy(extended) + .retryDelayResetThreshold(threshold, null) + .build()) { + es.start(); + + // Activate extended, then close the stream quickly so no reset triggers. + es.activateRetryDelayStrategy(extended); + + stream1.close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + // Extended is active; delay reflects extended's shape. + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + extendedIncrement)); + + // Let the next connection last past the reset threshold. Healthy-op reset + // should revert to the default (normal) strategy. + Thread.sleep(threshold + 10); + stream2.close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + // Reverted to default -> uses normal's shape. + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + normalIncrement)); + } + } + + @Test + public void perStrategyStateIsPreservedAcrossActivations() throws Exception { + // Each registered strategy's backoff progression state persists across + // activations. Deactivating and reactivating a strategy resumes from where + // its counter last left off (not fresh). + MockConnectStrategy mock = new MockConnectStrategy(); + PipedStreamRequestHandler stream1 = respondWithStream(); + PipedStreamRequestHandler stream2 = respondWithStream(); + PipedStreamRequestHandler stream3 = respondWithStream(); + PipedStreamRequestHandler stream4 = respondWithStream(); + mock.configureRequests(stream1, stream2, stream3, stream4); + + long initialDelay = 10; + int normalIncrement = 3, extendedIncrement = 100; + RetryDelayStrategy normal = new ArithmeticallyIncreasingRetryDelayStrategy(initialDelay, normalIncrement, 0); + RetryDelayStrategy extended = new ArithmeticallyIncreasingRetryDelayStrategy(initialDelay, extendedIncrement, 0); + try (EventSource es = baseBuilder(mock) + .retryDelayStrategy(normal) + .retryDelayStrategy(extended) + .build()) { + es.start(); + + // Fault 1 (default = normal). Counter advances on normal to 1. + stream1.close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay)); + + // Activate extended. Its counter is still 0. + es.activateRetryDelayStrategy(extended); + + stream2.close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + // Extended's first apply, counter=0. + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay)); + + stream3.close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + // Extended's second apply, counter=1. + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + extendedIncrement)); + + // Re-activate normal. Its counter was 1 when we left it. + es.activateRetryDelayStrategy(normal); + + stream4.close(); + assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException()))); + assertThat(es.readAnyEvent(), equalTo(new StartedEvent())); + // Normal resumes at counter=1: delay = initialDelay + 1 * normalIncrement. + assertThat(readReconnectDelayFromLog(), equalTo(initialDelay + normalIncrement)); + } + } } diff --git a/src/test/java/com/launchdarkly/eventsource/RetryDelayStrategyTest.java b/src/test/java/com/launchdarkly/eventsource/RetryDelayStrategyTest.java new file mode 100644 index 0000000..fbd854c --- /dev/null +++ b/src/test/java/com/launchdarkly/eventsource/RetryDelayStrategyTest.java @@ -0,0 +1,22 @@ +package com.launchdarkly.eventsource; + +import org.junit.Test; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.sameInstance; + +@SuppressWarnings("javadoc") +public class RetryDelayStrategyTest { + @Test + public void withBaseDelayMillisDefaultsToIdentityForCustomStrategies() { + // Strategies that do not override withBaseDelayMillis (i.e., have no notion of + // a base delay) return themselves unchanged when the wire retry hint fires. + RetryDelayStrategy s = new RetryDelayStrategy() { + @Override public long getDelayMillis() { return 500; } + @Override public RetryDelayStrategy getNext() { return this; } + }; + assertThat(s.withBaseDelayMillis(1234), sameInstance(s)); + assertThat(s.getDelayMillis(), equalTo(500L)); + } +}