- * The algorithm is as follows: - *
* 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);
* - * 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
* 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));
+ }
+}