- * 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. + * Activates a previously-registered {@link RetryDelayStrategy}, making it the + * strategy used for subsequent reconnect delay computations. *
- * 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. + * 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. *
- * 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()}. + * 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. *
- * At any other time, the value is undefined. + * This method is safe to call from any thread, but when called from a thread + * other than the one reading from this EventSource, the activation may not be + * observed for the impending reconnect if the reader thread has already begun + * computing its delay. In that case the swap takes effect on the following + * reconnect. * - * @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 || !registeredStrategiesState.containsKey(strategy)) { + return; + } + currentRetryDelayStrategy = strategy; } - + /** * Attempts to start the stream if it is not already active. *
@@ -228,8 +238,8 @@ public long getNextRetryDelayMillis() { *
* If the stream was previously active and then failed, {@link #start()} will sleep for * some amount of time-- the retry delay-- before trying to make the connection. The - * retry delay is determined by several factors: see {@link Builder#retryDelay(long, TimeUnit)}, - * {@link Builder#retryDelayStrategy(RetryDelayStrategy)}, and + * retry delay is determined by the configured + * {@link Builder#retryDelayStrategy(RetryDelayStrategy)} and * {@link Builder#retryDelayResetThreshold(long, TimeUnit)}. * @throws StreamException *
@@ -253,51 +263,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 interrupt(), stop(), or close() is called while we're waiting, we will
+ // trigger an early exit from this wait by calling sleepNotifier.notify().
+ 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();
}
+ } 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 +329,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 +618,14 @@ 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. Clamp against MAX_SERVER_DIRECTED_
+ // RETRY_DELAY_MILLIS to protect against misbehaving server issues.
+ serverDirectedInitialDelayMillis = Math.min(
+ ((SetRetryDelayEvent)event).getRetryMillis(),
+ MAX_SERVER_DIRECTED_RETRY_DELAY_MILLIS);
+ resetAllRegisteredStrategyState();
continue;
}
if (event instanceof MessageEvent) {
@@ -629,7 +647,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 +655,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 +663,46 @@ 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;
+ long connectionDurationMillis = disconnectedTime - 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 currentState = registeredStrategiesState.get(current);
+ RetryDelayStrategy next = currentState.getNext();
+ registeredStrategiesState.put(current, next != null ? next : currentState);
+ return currentState.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 registeredStrategiesState.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
- * The actual delay may be slightly less or greater, depending on the strategy specified by
- * {@link #retryDelayStrategy(RetryDelayStrategy)}. The default behavior is to increase the
- * delay exponentially from this base value on each attempt, up to a configured maximum,
- * substracting a random jitter; for more details, see {@link DefaultRetryDelayStrategy}.
+ * Whenever EventSource tries to start a new connection after a stream failure,
+ * it delays for an amount of time determined by the active
+ * {@link RetryDelayStrategy}. The default behavior is exponential backoff
+ * with jitter (see {@link RetryDelayStrategy#defaultStrategy()}).
*
- * If you set the base delay to zero, the backoff logic will not apply-- multiplying by
- * zero gives zero every time. Therefore, use a zero delay with caution since it could
- * cause a reconnect storm during a service interruption.
- *
- * @param retryDelay the base delay, in whatever time unit is specified by {@code timeUnit}
- * @param timeUnit the time unit, or {@code TimeUnit.MILLISECONDS} if null
- * @return the builder
- * @see EventSource#DEFAULT_RETRY_DELAY_MILLIS
- * @see #retryDelayStrategy(RetryDelayStrategy)
- * @see #retryDelayResetThreshold(long, TimeUnit)
- */
- public Builder retryDelay(long retryDelay, TimeUnit timeUnit) {
- retryDelayMillis = millisFromTimeUnit(retryDelay, timeUnit);
- return this;
- }
-
- /**
- * Specifies a strategy for determining the retry delay after an error.
+ * 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()}.
*
- * 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()}
+ * 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; must not be null
* @return the builder
- * @see #retryDelay(long, TimeUnit)
+ * @throws IllegalArgumentException if {@code retryDelayStrategy} is null
* @see #retryDelayResetThreshold(long, TimeUnit)
+ * @see EventSource#activateRetryDelayStrategy(RetryDelayStrategy)
* @since 4.0.0
*/
public Builder retryDelayStrategy(RetryDelayStrategy retryDelayStrategy) {
- this.retryDelayStrategy = retryDelayStrategy;
+ if (retryDelayStrategy == null) {
+ throw new IllegalArgumentException("retryDelayStrategy must not be null");
+ }
+ if (this.defaultRetryDelayStrategy == null) {
+ this.defaultRetryDelayStrategy = retryDelayStrategy;
+ } else {
+ 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..960abef 100644
--- a/src/main/java/com/launchdarkly/eventsource/RetryDelayStrategy.java
+++ b/src/main/java/com/launchdarkly/eventsource/RetryDelayStrategy.java
@@ -10,62 +10,55 @@
* 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. Returning {@code null} is treated as
+ * equivalent to returning {@code this}.
+ *
+ * @return the strategy to use next, or {@code null} to reuse this instance
+ * @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..ecee78f 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,71 @@ 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));
+ }
+
+ @Test
+ public void initialDelayAboveMaxDelayIsClamped() {
+ // The pre-PR apply(base) path pinned every attempt (including the first)
+ // against maxDelay. Post-PR, the max is enforced only in getNext(), so
+ // an initialDelay above maxDelay must still be clamped at construction
+ // for parity.
+ long max = 30_000;
+ RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy()
+ .initialDelay(1, TimeUnit.HOURS)
+ .backoffMultiplier(2).jitterMultiplier(0)
+ .maxDelay(max, TimeUnit.MILLISECONDS);
+
+ assertThat(s.getDelayMillis(), equalTo(max));
+ // Subsequent progression stays pinned.
+ assertThat(s.getNext().getDelayMillis(), equalTo(max));
+ }
+
+ @Test
+ public void withBaseDelayMillisAboveMaxDelayIsClamped() {
+ // A wire retry hint whose value exceeds the strategy's maxDelay must not
+ // bypass the max on the immediate reconnect. withBaseDelayMillis is the
+ // entry point for wire hints via EventSource.resetAllRegisteredStrategyState.
+ long max = 30_000;
+ RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy()
+ .initialDelay(1, TimeUnit.SECONDS)
+ .backoffMultiplier(2).jitterMultiplier(0)
+ .maxDelay(max, TimeUnit.MILLISECONDS)
+ .withBaseDelayMillis(60_000);
+
+ assertThat(s.getDelayMillis(), equalTo(max));
+ }
+
+ @Test
+ public void initialDelayIsPreservedWhenMaxDelayRaisedLater() {
+ // Reproduces the extended-regime configuration pattern from the PR
+ // description:
+ // .initialDelay(5, MINUTES).maxDelay(1, HOURS)
+ // The initialDelay call runs against defaultStrategy()'s current
+ // DEFAULT_MAX_DELAY_MILLIS (30 s) and would clamp base to 30 s if the
+ // constructor clamped baseDelayMillis eagerly. The subsequent maxDelay
+ // call raises the ceiling to 1 hour, so the caller's 5-minute base must
+ // survive to the final instance.
+ long expectedBase = 5 * 60 * 1000;
+ RetryDelayStrategy s = RetryDelayStrategy.defaultStrategy()
+ .initialDelay(5, TimeUnit.MINUTES)
+ .maxDelay(1, TimeUnit.HOURS)
+ .backoffMultiplier(2).jitterMultiplier(0);
+
+ assertThat(s.getDelayMillis(), equalTo(expectedBase));
}
}
diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceBuilderTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceBuilderTest.java
index c490e70..19f7cae 100644
--- a/src/test/java/com/launchdarkly/eventsource/EventSourceBuilderTest.java
+++ b/src/test/java/com/launchdarkly/eventsource/EventSourceBuilderTest.java
@@ -82,16 +82,21 @@ 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));
}
}
+
+ @Test(expected=IllegalArgumentException.class)
+ public void retryDelayStrategyRejectsNull() {
+ builder.retryDelayStrategy(null);
+ }
@Test
public void retryDelayResetThreshold() {
diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceErrorStrategyUsageTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceErrorStrategyUsageTest.java
index 2a0e7b0..fa7d90a 100644
--- a/src/test/java/com/launchdarkly/eventsource/EventSourceErrorStrategyUsageTest.java
+++ b/src/test/java/com/launchdarkly/eventsource/EventSourceErrorStrategyUsageTest.java
@@ -7,6 +7,7 @@
import org.junit.Test;
import java.io.IOException;
+import java.util.concurrent.TimeUnit;
import static com.launchdarkly.eventsource.MockConnectStrategy.rejectConnection;
import static com.launchdarkly.eventsource.MockConnectStrategy.respondWithDataAndThenStayOpen;
@@ -30,7 +31,7 @@ public class EventSourceErrorStrategyUsageTest {
private EventSource.Builder baseBuilder(MockConnectStrategy mock) {
return new EventSource.Builder(mock)
- .retryDelay(1, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.logger(testLogger.getLogger());
}
diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceReadingTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceReadingTest.java
index cf87245..3928e8e 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));
}
}
@@ -193,7 +195,7 @@ public void lastEventIdIsUpdatedFromEvent() throws Exception {
stream.provideData(body);
try (EventSource es = baseBuilder(mock)
- .retryDelay(10, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(10, TimeUnit.MILLISECONDS))
.build()) {
es.start();
@@ -217,14 +219,16 @@ public void lastEventIdIsUpdatedFromEvent() throws Exception {
public void initialRetryDelayIsSetFromBuilder() throws Exception {
MockConnectStrategy mock = new MockConnectStrategy();
- try (EventSource es = baseBuilder(mock).retryDelay(6, TimeUnit.SECONDS).build()) {
- assertEquals(6000, es.getBaseRetryDelayMillis());
+ try (EventSource es = baseBuilder(mock)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(6, TimeUnit.SECONDS))
+ .build()) {
+ assertEquals(6000, ((DefaultRetryDelayStrategy) es.defaultRetryDelayStrategy).baseDelayMillis);
}
}
@Test
public void retryDelayIsUpdatedFromEvent() throws Exception {
- String eventData = "some-data";
+ String eventData = "some-data";
String body = "retry: 300\n" + "\ndata: " + eventData + "\n\n";
MockConnectStrategy mock = new MockConnectStrategy();
@@ -233,7 +237,7 @@ public void retryDelayIsUpdatedFromEvent() throws Exception {
stream.provideData(body);
try (EventSource es = baseBuilder(mock)
- .retryDelay(10, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(10, TimeUnit.MILLISECONDS))
.build()) {
es.start();
@@ -242,8 +246,36 @@ 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);
+ }
+ }
+
+ @Test
+ public void retryDelayFromEventIsClampedToMax() throws Exception {
+ // Server sends a wire retry hint above MAX_SERVER_DIRECTED_RETRY_DELAY_MILLIS;
+ // the client applies the clamped value, not the raw hint. Use a strategy
+ // whose maxDelay comfortably exceeds MAX_SERVER_DIRECTED_RETRY_DELAY_MILLIS
+ // so the wire-side clamp is the load-bearing bound (not the strategy max).
+ long hugeHint = EventSource.MAX_SERVER_DIRECTED_RETRY_DELAY_MILLIS + 1_000_000L;
+ String body = "retry: " + hugeHint + "\n\ndata: x\n\n";
+
+ MockConnectStrategy mock = new MockConnectStrategy();
+ PipedStreamRequestHandler stream = respondWithStream();
+ mock.configureRequests(stream);
+ stream.provideData(body);
+
+ try (EventSource es = baseBuilder(mock)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy()
+ .initialDelay(10, TimeUnit.MILLISECONDS)
+ .maxDelay(24, TimeUnit.HOURS))
+ .build()) {
+ es.start();
+ assertThat(es.readAnyEvent(), equalTo(
+ new MessageEvent("message", "x", null, ORIGIN)));
+ assertEquals(EventSource.MAX_SERVER_DIRECTED_RETRY_DELAY_MILLIS,
+ ((DefaultRetryDelayStrategy) es.currentRetryStrategySnapshot()).baseDelayMillis);
}
}
@@ -371,7 +403,7 @@ public void canIterateMessagesWithAutoRetry() throws Exception {
new Thread(() -> {
es.set(baseBuilder(mock)
.errorStrategy(ErrorStrategy.alwaysContinue())
- .retryDelay(1, TimeUnit.MILLISECONDS)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.build());
for (MessageEvent m: es.get().messages()) {
queue.add(m);
@@ -418,7 +450,7 @@ public void canIterateEventsWithAutoRetry() throws Exception {
new Thread(() -> {
es.set(baseBuilder(mock)
.errorStrategy(ErrorStrategy.alwaysContinue())
- .retryDelay(1, TimeUnit.MILLISECONDS)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.build());
for (StreamEvent e: es.get().anyEvents()) {
queue.add(e);
diff --git a/src/test/java/com/launchdarkly/eventsource/EventSourceReconnectTest.java b/src/test/java/com/launchdarkly/eventsource/EventSourceReconnectTest.java
index 47b1589..d9f97fb 100644
--- a/src/test/java/com/launchdarkly/eventsource/EventSourceReconnectTest.java
+++ b/src/test/java/com/launchdarkly/eventsource/EventSourceReconnectTest.java
@@ -4,6 +4,7 @@
import org.junit.Rule;
import org.junit.Test;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static com.launchdarkly.eventsource.MockConnectStrategy.ORIGIN;
@@ -28,7 +29,6 @@ public class EventSourceReconnectTest {
private EventSource.Builder baseBuilder(MockConnectStrategy mock) {
return new EventSource.Builder(mock)
.errorStrategy(ErrorStrategy.alwaysContinue())
- .retryDelay(BRIEF_DELAY, null)
.logger(testLogger.getLogger());
}
@@ -41,15 +41,17 @@ public void eventSourceReconnectsAfterStreamClosedByServer() throws Exception {
respondWithDataAndThenEnd(message1),
respondWithDataAndThenStayOpen(message2));
- try (EventSource es = baseBuilder(mock).build()) {
+ try (EventSource es = baseBuilder(mock)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(BRIEF_DELAY, TimeUnit.MILLISECONDS))
+ .build()) {
assertThat(es.getState(), equalTo(ReadyState.RAW));
es.start();
-
+
assertThat(es.getState(), equalTo(ReadyState.OPEN));
assertThat(es.readAnyEvent(), equalTo(new MessageEvent("message", "first", null, ORIGIN)));
-
+
assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException())));
assertThat(es.getState(), equalTo(ReadyState.CLOSED));
@@ -70,15 +72,17 @@ public void eventSourceReconnectsAfterExternallyInterrupted() throws Exception {
mock.configureRequests(respondWithDataAndThenStayOpen(message1),
respondWithDataAndThenStayOpen(message2));
- try (EventSource es = baseBuilder(mock).build()) {
+ try (EventSource es = baseBuilder(mock)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(BRIEF_DELAY, TimeUnit.MILLISECONDS))
+ .build()) {
assertThat(es.getState(), equalTo(ReadyState.RAW));
-
+
es.start();
assertThat(es.getState(), equalTo(ReadyState.OPEN));
-
+
assertThat(es.readAnyEvent(), equalTo(new MessageEvent("message", "first", null, ORIGIN)));
-
+
interruptOnAnotherThread(es);
assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByCallerException())));
@@ -100,30 +104,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 +134,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 +148,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 +156,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..9c50c7d 100644
--- a/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java
+++ b/src/test/java/com/launchdarkly/eventsource/EventSourceRetryDelayStrategyUsageTest.java
@@ -7,6 +7,8 @@
import org.junit.Rule;
import org.junit.Test;
+import java.util.concurrent.TimeUnit;
+
import static com.launchdarkly.eventsource.MockConnectStrategy.ORIGIN;
import static com.launchdarkly.eventsource.MockConnectStrategy.respondWithDataAndThenEnd;
import static com.launchdarkly.eventsource.MockConnectStrategy.respondWithStream;
@@ -15,6 +17,7 @@
import static org.hamcrest.Matchers.endsWith;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.startsWith;
+import static org.junit.Assert.assertEquals;
/**
* These tests verify that EventSource interacts with the configured RetryDelayStrategy
@@ -32,14 +35,18 @@ public class EventSourceRetryDelayStrategyUsageTest {
private EventSource.Builder baseBuilder(MockConnectStrategy mock) {
return new EventSource.Builder(mock)
.errorStrategy(ErrorStrategy.alwaysContinue())
- .retryDelay(BRIEF_DELAY, null)
.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 +61,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 +78,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 +95,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 +111,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 +128,390 @@ 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 long getDelayMillis() {
+ return baseDelayMillis + (counter * increment);
+ }
+
+ @Override
+ public RetryDelayStrategy getNext() {
+ return new ArithmeticallyIncreasingRetryDelayStrategy(baseDelayMillis, increment, counter + 1);
+ }
+
@Override
- public Result apply(long baseDelayMillis) {
- return new Result(
- baseDelayMillis + (counter * increment),
- new ArithmeticallyIncreasingRetryDelayStrategy(increment, counter + 1)
- );
+ 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;
+ 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));
+ }
+ }
+
+ @Test
+ public void wireRetryHintAppliesToAllRegisteredStrategies() throws Exception {
+ // A server-directed retry hint received via the SSE "retry:" field is
+ // applied to every registered strategy's snapshot, not just the currently-
+ // active one. Verifies the "sticky-to-all" behavior called out in the PR
+ // description (and matching Go's ApplyRetryTime).
+ MockConnectStrategy mock = new MockConnectStrategy();
+ PipedStreamRequestHandler stream = respondWithStream();
+ mock.configureRequests(stream);
+ stream.provideData("retry: 500\n\ndata: x\n\n");
+
+ DefaultRetryDelayStrategy normal = RetryDelayStrategy.defaultStrategy()
+ .initialDelay(1000, TimeUnit.MILLISECONDS)
+ .jitterMultiplier(0);
+ DefaultRetryDelayStrategy extended = RetryDelayStrategy.defaultStrategy()
+ .initialDelay(60_000, TimeUnit.MILLISECONDS)
+ .jitterMultiplier(0);
+
+ try (EventSource es = baseBuilder(mock)
+ .retryDelayStrategy(normal) // default (active)
+ .retryDelayStrategy(extended) // additional
+ .build()) {
+ es.start();
+ assertThat(es.readAnyEvent(), equalTo(new MessageEvent("message", "x", null, ORIGIN)));
+
+ // Normal is the active strategy; its snapshot should reflect the wire hint.
+ assertEquals(500L,
+ ((DefaultRetryDelayStrategy) es.currentRetryStrategySnapshot()).baseDelayMillis);
+
+ // Extended was not active when the hint arrived, but the hint stampeded
+ // across every registered strategy's reset instance. Activate to peek.
+ es.activateRetryDelayStrategy(extended);
+ assertEquals(500L,
+ ((DefaultRetryDelayStrategy) es.currentRetryStrategySnapshot()).baseDelayMillis);
+ }
+ }
+
+ @Test
+ public void wireRetryHintIsStickyAcrossHealthyOpReset() throws Exception {
+ // A wire hint received on connection N stays sticky when a later healthy-op
+ // reset fires: the reset re-instantiates each registered strategy against
+ // the wire base, not the caller's originally-registered base. Verifies the
+ // WHATWG-sticky claim in the PR description.
+ MockConnectStrategy mock = new MockConnectStrategy();
+ PipedStreamRequestHandler stream1 = respondWithStream();
+ PipedStreamRequestHandler stream2 = respondWithStream();
+ mock.configureRequests(stream1, stream2);
+ stream1.provideData("retry: 500\n\ndata: x\n\n");
+
+ long threshold = 50;
+ // backoffMultiplier(1) keeps base flat across getNext() so the assertion
+ // reads the pure post-reset base.
+ DefaultRetryDelayStrategy normal = RetryDelayStrategy.defaultStrategy()
+ .initialDelay(1000, TimeUnit.MILLISECONDS)
+ .jitterMultiplier(0)
+ .backoffMultiplier(1);
+
+ try (EventSource es = baseBuilder(mock)
+ .retryDelayStrategy(normal)
+ .retryDelayResetThreshold(threshold, null)
+ .build()) {
+ es.start();
+ assertThat(es.readAnyEvent(), equalTo(new MessageEvent("message", "x", null, ORIGIN)));
+
+ // Let the connection live past the reset threshold, then fault it. The
+ // healthy-op reset should fire in computeReconnectDelay and re-apply the
+ // wire hint (500), not revert to the caller's original base (1000).
+ Thread.sleep(threshold + 10);
+ stream1.close();
+ assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException())));
+ assertThat(es.readAnyEvent(), equalTo(new StartedEvent()));
+ assertThat(readReconnectDelayFromLog(), equalTo(500L));
+ }
+ }
+
+ @Test
+ public void healthyOpResetIgnoresConsumerProcessingDelay() throws Exception {
+ // computeReconnectDelay runs at sleep-time (after the consumer has been
+ // handed a FaultEvent and looped back to readAnyEvent). The healthy-op
+ // threshold check must measure only the prior connection's duration
+ // (disconnectedTime - connectedTime), NOT (now - connectedTime), or a
+ // slow consumer can spuriously trip the reset for a connection that
+ // was actually short.
+ MockConnectStrategy mock = new MockConnectStrategy();
+ PipedStreamRequestHandler stream1 = respondWithStream();
+ PipedStreamRequestHandler stream2 = respondWithStream();
+ PipedStreamRequestHandler stream3 = respondWithStream();
+ mock.configureRequests(stream1, stream2, stream3);
+
+ long threshold = 200;
+ long initialDelay = 100;
+ DefaultRetryDelayStrategy strat = RetryDelayStrategy.defaultStrategy()
+ .initialDelay(initialDelay, TimeUnit.MILLISECONDS)
+ .jitterMultiplier(0); // backoffMultiplier default 2
+
+ try (EventSource es = baseBuilder(mock)
+ .retryDelayStrategy(strat)
+ .retryDelayResetThreshold(threshold, TimeUnit.MILLISECONDS)
+ .build()) {
+ es.start();
+
+ // Fault 1: brief connection, well below threshold. Delay = initialDelay.
+ stream1.close();
+ assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException())));
+ assertThat(es.readAnyEvent(), equalTo(new StartedEvent()));
+ assertThat(readReconnectDelayFromLog(), equalTo(initialDelay));
+
+ // Fault 2: same brief connection, but the consumer takes a long time
+ // between reading the FaultEvent and readAnyEvent'ing again -- long
+ // enough that (now - connectedTime) crosses the threshold even though
+ // the connection itself did not.
+ stream2.close();
+ assertThat(es.readAnyEvent(), equalTo(new FaultEvent(new StreamClosedByServerException())));
+ Thread.sleep(threshold + 100);
+ assertThat(es.readAnyEvent(), equalTo(new StartedEvent()));
+
+ // Correct behavior: no reset (actual connection duration << threshold),
+ // so the strategy's counter advances and delay is initialDelay * 2.
+ // Bug (pre-fix): reset fires because now - connectedTime >= threshold,
+ // giving delay = initialDelay again.
+ assertThat(readReconnectDelayFromLog(), equalTo(initialDelay * 2));
+ }
+ }
}
diff --git a/src/test/java/com/launchdarkly/eventsource/HttpConnectStrategyWithEventSourceTest.java b/src/test/java/com/launchdarkly/eventsource/HttpConnectStrategyWithEventSourceTest.java
index 2ce7493..741bc2b 100644
--- a/src/test/java/com/launchdarkly/eventsource/HttpConnectStrategyWithEventSourceTest.java
+++ b/src/test/java/com/launchdarkly/eventsource/HttpConnectStrategyWithEventSourceTest.java
@@ -9,6 +9,8 @@
import org.hamcrest.Matchers;
+import java.util.concurrent.TimeUnit;
+
import static com.launchdarkly.eventsource.TestUtils.interruptOnAnotherThread;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
@@ -104,7 +106,7 @@ public void eventSourceReconnectsAfterSocketClosed() throws Exception {
try (HttpServer server = HttpServer.start(allResponses)) {
try (EventSource es = new EventSource.Builder(server.getUri())
.errorStrategy(ErrorStrategy.alwaysContinue())
- .retryDelay(1, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.build()) {
es.start();
@@ -136,7 +138,7 @@ public void eventSourceReconnectsAfterExternallyInterrupted() throws Exception {
try (HttpServer server = HttpServer.start(allResponses)) {
try (EventSource es = new EventSource.Builder(server.getUri())
.errorStrategy(ErrorStrategy.alwaysContinue())
- .retryDelay(1, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.build()) {
es.start();
@@ -237,7 +239,7 @@ public void messageEventsIncludeHeadersFromReconnectedConnection() throws Except
try (HttpServer server = HttpServer.start(allResponses)) {
try (EventSource es = new EventSource.Builder(server.getUri())
.errorStrategy(ErrorStrategy.alwaysContinue())
- .retryDelay(1, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.build()) {
es.start();
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));
+ }
+}
diff --git a/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceBasicTest.java b/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceBasicTest.java
index 2be5b79..7a06e2d 100644
--- a/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceBasicTest.java
+++ b/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceBasicTest.java
@@ -2,11 +2,14 @@
import com.launchdarkly.eventsource.EventSource;
import com.launchdarkly.eventsource.MockConnectStrategy;
+import com.launchdarkly.eventsource.RetryDelayStrategy;
import com.launchdarkly.eventsource.StreamClosedByServerException;
import com.launchdarkly.eventsource.TestScopedLoggerRule;
import com.launchdarkly.eventsource.background.Stubs.LogItem;
import com.launchdarkly.eventsource.background.Stubs.TestHandler;
+import java.util.concurrent.TimeUnit;
+
import org.junit.Rule;
import org.junit.Test;
@@ -23,7 +26,7 @@ public class BackgroundEventSourceBasicTest {
private EventSource.Builder baseEventSourceBuilder() {
return new EventSource.Builder(mockConnect)
- .retryDelay(1, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.logger(testLogger.getLogger());
}
diff --git a/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceErrorHandlingTest.java b/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceErrorHandlingTest.java
index 759a4aa..d8bf495 100644
--- a/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceErrorHandlingTest.java
+++ b/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceErrorHandlingTest.java
@@ -4,6 +4,7 @@
import com.launchdarkly.eventsource.MessageEvent;
import com.launchdarkly.eventsource.MockConnectStrategy;
import com.launchdarkly.eventsource.ReadyState;
+import com.launchdarkly.eventsource.RetryDelayStrategy;
import com.launchdarkly.eventsource.StreamClosedByServerException;
import com.launchdarkly.eventsource.StreamException;
import com.launchdarkly.eventsource.StreamHttpErrorException;
@@ -40,7 +41,7 @@ private void verifyErrorLogged(Throwable t) {
private EventSource.Builder baseEventSourceBuilder() {
return new EventSource.Builder(mockConnect)
- .retryDelay(1, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.logger(testLogger.getLogger());
}
diff --git a/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceThreadingTest.java b/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceThreadingTest.java
index e95ab0e..e2f97d9 100644
--- a/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceThreadingTest.java
+++ b/src/test/java/com/launchdarkly/eventsource/background/BackgroundEventSourceThreadingTest.java
@@ -4,6 +4,7 @@
import com.launchdarkly.eventsource.MessageEvent;
import com.launchdarkly.eventsource.MockConnectStrategy;
import com.launchdarkly.eventsource.MockConnectStrategy.PipedStreamRequestHandler;
+import com.launchdarkly.eventsource.RetryDelayStrategy;
import com.launchdarkly.eventsource.StreamClosedByServerException;
import com.launchdarkly.eventsource.TestScopedLoggerRule;
import com.launchdarkly.eventsource.background.Stubs.LogItem;
@@ -36,7 +37,7 @@ public class BackgroundEventSourceThreadingTest {
private EventSource.Builder baseEventSourceBuilder() {
return new EventSource.Builder(mockConnect)
- .retryDelay(1, null)
+ .retryDelayStrategy(RetryDelayStrategy.defaultStrategy().initialDelay(1, TimeUnit.MILLISECONDS))
.logger(testLogger.getLogger());
}