@@ -97,10 +97,16 @@ open class EventStreamClient(
97 * Read SSE events from [events], racing each read against an idle-timeout watchdog.
97 * Read SSE events from [events], racing each read against an idle-timeout watchdog.
98 * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
98 * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
99 * clean stream close or scope cancellation.
99 * clean stream close or scope cancellation.
101+ * If the idle watchdog tripped because a [triggerReconnect] signal was consumed
102+ * (rather than a genuine idle timeout), [onReconnectConsumed] is invoked so the outer
103+ * reconnect loop can reconnect immediately instead of waiting out the backoff — the
104+ * signal is already consumed, so the backoff `select` couldn't be cut short.
101 private suspend fun readSseEvents(
106 private suspend fun readSseEvents(
102 scope: CoroutineScope,
107 scope: CoroutineScope,
103 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
108 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
109+ onReconnectConsumed: () -> Unit,
104 send: suspend (BusEvent) -> Unit,
110 send: suspend (BusEvent) -> Unit,
106 while (scope.isActive) {
112 while (scope.isActive) {
@@ -108,17 +114,20 @@ open class EventStreamClient(
108 // read), so a socket the peer never closed would otherwise hang until OS TCP
114 // read), so a socket the peer never closed would otherwise hang until OS TCP
109 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
115 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
110 // within the idle timeout, treat the connection as half-open and drop it.
116 // within the idle timeout, treat the connection as half-open and drop it.
117+ var consumedReconnectSignal = false
111 val result: ChannelResult<ServerSentEvent>? = select {
118 val result: ChannelResult<ServerSentEvent>? = select {
112 events.onReceiveCatching { it }
119 events.onReceiveCatching { it }
113 onTimeout(idleTimeoutMs) { null }
120 onTimeout(idleTimeoutMs) { null }
114 // A triggerReconnect() issued while a read is active must interrupt it now:
121 // A triggerReconnect() issued while a read is active must interrupt it now:
115 // return the same null sentinel as the idle timeout so a half-open socket
122 // return the same null sentinel as the idle timeout so a half-open socket
116 // (no TCP RST) is dropped and reconnected promptly instead of waiting out
123 // (no TCP RST) is dropped and reconnected promptly instead of waiting out
117− // idleTimeoutMs (~90s). Consuming the signal here also stops the backoff
124+ // idleTimeoutMs (~90s).
118− // select below from re-triggering on the stale request.
119− reconnectSignal.onReceive { null }
125+ reconnectSignal.onReceive { consumedReconnectSignal = true; null }
127+ if (result == null) {
128+ if (consumedReconnectSignal) onReconnectConsumed()
129+ throw IOException("SSE read dropped (idle timeout or reconnect requested), reconnecting")
121− if (result == null) throw IOException("SSE read dropped (idle timeout or reconnect requested), reconnecting")
122 val sse = result.getOrNull()
131 val sse = result.getOrNull()
123 if (sse == null) {
132 if (sse == null) {
124 // The producer channel is closed. A non-null cause means `incoming`
133 // The producer channel is closed. A non-null cause means `incoming`
@@ -158,6 +167,12 @@ open class EventStreamClient(
158 var backoffMs = initialBackoffMs
167 var backoffMs = initialBackoffMs
159 while (isActive) {
168 while (isActive) {
160 _state.value = ConnectionState.Connecting
169 _state.value = ConnectionState.Connecting
170+ // Set by readSseEvents when a triggerReconnect() signal interrupted an active
171+ // read. The signal is consumed to drop the socket, so the backoff select below
172+ // can't also see it — without this flag the reconnect would wait the full
173+ // backoff despite the caller asking for an immediate reconnect (network
174+ // handoff, manual refresh, ...).
175+ var reconnectConsumedDuringRead = false
162 // The SSE stream is long-lived: disable the request-level and socket
177 // The SSE stream is long-lived: disable the request-level and socket
163 // timeouts (the client default is 60s for REST) in the request config
178 // timeouts (the client default is 60s for REST) in the request config
@@ -221,7 +236,7 @@ open class EventStreamClient(
221 // never escalate the exponential ramp.
236 // never escalate the exponential ramp.
222 var receivedAny = false
237 var receivedAny = false
224− readSseEvents(scope, events) {
239+ readSseEvents(scope, events, { reconnectConsumedDuringRead = true }) {
225 if (!receivedAny) {
240 if (!receivedAny) {
226 receivedAny = true
241 receivedAny = true
227 backoffMs = initialBackoffMs
242 backoffMs = initialBackoffMs
@@ -272,14 +287,30 @@ open class EventStreamClient(
273 _state.value = ConnectionState.Disconnected
288 _state.value = ConnectionState.Disconnected
274 if (!isActive) break
289 if (!isActive) break
275− // Wait for the backoff, but allow a reconnect signal to cut it short.
290+ // Wait for the backoff, but allow a reconnect signal to cut it short. When
276− var signaled = false
291+ // readSseEvents consumed a triggerReconnect() signal to drop an active read,
277− val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
292+ // the signal is already spent so the select below can't receive it — pass a
278− select {
293+ // zero backoff in that case so the reconnect is immediate (the whole point of
279− reconnectSignal.onReceive { signaled = true }
294+ // triggerReconnect: network handoff, manual refresh, ...).
280− onTimeout((backoffMs + jitter).coerceAtLeast(0)) { /* normal backoff elapsed */ }
295+ val signaled = waitForBackoffOrSignal(reconnectConsumedDuringRead, backoffMs)
282 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
296 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
300+ /** Suspend for the backoff unless [reconnectConsumedDuringRead] (immediate) or a new
301+ * [triggerReconnect] arrives during the wait. Returns true when the reconnect should
302+ * skip the backoff ramp (signal-driven), false when the full backoff elapsed. */
303+ private suspend fun waitForBackoffOrSignal(
304+ reconnectConsumedDuringRead: Boolean,
307+ if (reconnectConsumedDuringRead) return true
308+ var signaled = false
309+ val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
311+ reconnectSignal.onReceive { signaled = true }
312+ onTimeout((backoffMs + jitter).coerceAtLeast(0)) { /* normal backoff elapsed */ }