Skip to content

feat: add resilient SSE session with reconnection/backoff#43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dogo wants to merge 4 commits into
base:main
Choose a base branch
from
Open

feat: add resilient SSE session with reconnection/backoff #43

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
97e320b
feat: add resilient SSE session with reconnection/backoff
dogo Oct 1, 2025
38d9edb
Merge remote-tracking branch 'upstream/main' into feature/reconnect
dogo Nov 12, 2025
0a21302
Merge branch 'main' of github.com:Recouse/EventSource into feature/re…
dogo Feb 21, 2026
9b134e8
Use Task.sleep for reconnect delay and reorder methods after init
dogo Feb 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 94 additions & 39 deletions Sources/EventSource/EventSource.swift
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -43,9 +43,12 @@ public struct EventSource: Sendable {
private let eventParser: @Sendable () -> EventParser

public var timeoutIntervalForRequest: TimeInterval

public var timeoutIntervalForResource: TimeInterval

public var maxReconnectAttempts: Int
public var reconnectInitialDelay: TimeInterval
public var reconnectBackoffFactor: Double

public init(
mode: Mode = .default,
timeoutIntervalForRequest: TimeInterval = 60,
Expand All@@ -63,20 +66,29 @@ public struct EventSource: Sendable {
mode: Mode = .default,
eventParser: @autoclosure @escaping @Sendable () -> EventParser,
timeoutIntervalForRequest: TimeInterval = 60,
timeoutIntervalForResource: TimeInterval = 300
timeoutIntervalForResource: TimeInterval = 300,
maxReconnectAttempts: Int = 5,
reconnectInitialDelay: TimeInterval = 1.0,
reconnectBackoffFactor: Double = 2.0
) {
self.mode = mode
self.eventParser = eventParser
self.timeoutIntervalForRequest = timeoutIntervalForRequest
self.timeoutIntervalForResource = timeoutIntervalForResource
self.maxReconnectAttempts = maxReconnectAttempts
self.reconnectInitialDelay = reconnectInitialDelay
self.reconnectBackoffFactor = reconnectBackoffFactor
}

public func dataTask(for urlRequest: URLRequest) -> DataTask {
DataTask(
urlRequest: urlRequest,
eventParser: eventParser(),
timeoutIntervalForRequest: timeoutIntervalForRequest,
timeoutIntervalForResource: timeoutIntervalForResource
timeoutIntervalForResource: timeoutIntervalForResource,
maxReconnectAttempts: maxReconnectAttempts,
reconnectInitialDelay: reconnectInitialDelay,
reconnectBackoffFactor: reconnectBackoffFactor
)
}
}
Expand All@@ -88,8 +100,24 @@ public extension EventSource {
/// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
/// it can be started by iterating event stream returned by ``DataTask/events()``.
final class DataTask: Sendable {

private let _readyState: Mutex<ReadyState> = Mutex(.none)

// Reconnection properties
private let maxReconnectAttempts: Int
private let reconnectInitialDelay: TimeInterval
private let reconnectBackoffFactor: Double

private let _reconnectAttempts: Mutex<Int> = Mutex(0)
private var reconnectAttempts: Int {
get {
_reconnectAttempts.withLock { $0 }
}
set {
_reconnectAttempts.withLock { $0 = newValue }
}
}

/// A value representing the state of the connection.
public var readyState: ReadyState {
get {
Expand DownExpand Up@@ -170,12 +198,18 @@ public extension EventSource {
urlRequest: URLRequest,
eventParser: EventParser,
timeoutIntervalForRequest: TimeInterval,
timeoutIntervalForResource: TimeInterval
timeoutIntervalForResource: TimeInterval,
maxReconnectAttempts: Int,
reconnectInitialDelay: TimeInterval,
reconnectBackoffFactor: Double
) {
self.urlRequest = urlRequest
self._eventParser = Mutex(eventParser)
self.timeoutIntervalForRequest = timeoutIntervalForRequest
self.timeoutIntervalForResource = timeoutIntervalForResource
self.maxReconnectAttempts = maxReconnectAttempts
self.reconnectInitialDelay = reconnectInitialDelay
self.reconnectBackoffFactor = reconnectBackoffFactor
}

/// Creates and returns event stream.
Expand All@@ -188,42 +222,56 @@ public extension EventSource {
}

return AsyncStream { continuation in
let sessionDelegate = SessionDelegate()
let urlSession = URLSession(
configuration: urlSessionConfiguration,
delegate: sessionDelegate,
delegateQueue: nil
)
let urlSessionDataTask = urlSession.dataTask(with: urlRequest)

let sessionDelegateTask = Task { [weak self] in
for await event in sessionDelegate.eventStream {
guard let self else { return }

switch event {
case let .didCompleteWithError(error):
handleSessionError(error, stream: continuation, urlSession: urlSession)
case let .didReceiveResponse(response, completionHandler):
handleSessionResponse(
response,
stream: continuation,
urlSession: urlSession,
completionHandler: completionHandler
)
case let .didReceiveData(data):
parseMessages(from: data, stream: continuation, urlSession: urlSession)
}
startSession(stream: continuation)
readyState = .connecting
consumed = true
}
}

/// Initializes or reinitializes the SSE session
private func startSession(stream continuation: AsyncStream<EventType>.Continuation) {
let sessionDelegate = SessionDelegate()
let urlSession = URLSession(
configuration: urlSessionConfiguration,
delegate: sessionDelegate,
delegateQueue: nil
)
let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
let sessionDelegateTask = Task { [weak self] in
for await event in sessionDelegate.eventStream {
guard let self else { return }
switch event {
case let .didCompleteWithError(error):
self.handleSessionError(error, stream: continuation, urlSession: urlSession)
case let .didReceiveResponse(response, completionHandler):
self.handleSessionResponse(
response,
stream: continuation,
urlSession: urlSession,
completionHandler: completionHandler
)
case let .didReceiveData(data):
self.parseMessages(from: data, stream: continuation, urlSession: urlSession)
}
}
}

continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}
continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}

urlSessionDataTask.resume()
readyState = .connecting
consumed = true
urlSessionDataTask.resume()
}

/// Helper method for reconnection
private func attemptReconnect(stream continuation: AsyncStream<EventType>.Continuation) {
let delay = reconnectInitialDelay * pow(reconnectBackoffFactor, Double(reconnectAttempts - 1))
Task { [weak self] in
try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
self?.startSession(stream: continuation)
self?.readyState = .connecting
self?.consumed = true
}
}

Expand All@@ -241,9 +289,15 @@ public extension EventSource {
if let error {
sendErrorEvent(with: error, stream: continuation)
}

// Close connection
close(stream: continuation, urlSession: urlSession)

// Attempts to reconnect if the limit has not been exceeded
if reconnectAttempts < maxReconnectAttempts {
Comment on lines +293 to +294

@RecouseRecouseFeb 24, 2026

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
// Attempts to reconnect if the limit has not been exceeded
if reconnectAttempts < maxReconnectAttempts {
// If an error occurred, an error object will indicate how the request failed, otherwise nil.
// Attempts to reconnect on error if the limit has not been exceeded.
if error !=nil, reconnectAttempts < maxReconnectAttempts {

reconnectAttempts += 1
attemptReconnect(stream: continuation)
} else {
// Close connection if attempts exceeded
close(stream: continuation, urlSession: urlSession)
}
}

private func handleSessionResponse(
Expand DownExpand Up@@ -321,6 +375,7 @@ public extension EventSource {

private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
readyState = .open
reconnectAttempts = 0 // reset attempts when opening
continuation.yield(.open)
}

Expand Down
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
feat: add resilient SSE session with reconnection/backoff by dogo · Pull Request #43 · Recouse/EventSource · GitHub
Skip to content
feat: add resilient SSE session with reconnection/backoff#43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

dogo
wants to merge
4
commits into

base:main
Choose a base branch

from

 Open


feat: add resilient SSE session with reconnection/backoff
#43
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
97e320b

feat: add resilient SSE session with reconnection/backoff


dogo Oct 1, 2025
38d9edb

Merge remote-tracking branch 'upstream/main' into feature/reconnect


dogo Nov 12, 2025
0a21302

Merge branch 'main' of github.com:Recouse/EventSource into feature/re…


dogo Feb 21, 2026
9b134e8

Use Task.sleep for reconnect delay and reorder methods after init


dogo Feb 23, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


133 changes: 94 additions & 39 deletions
Sources/EventSource/EventSource.swift

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -43,9 +43,12 @@ public struct EventSource: Sendable {
 private let eventParser: @Sendable () -> EventParser

 public var timeoutIntervalForRequest: TimeInterval

 public var timeoutIntervalForResource: TimeInterval

 public var maxReconnectAttempts: Int
 public var reconnectInitialDelay: TimeInterval
 public var reconnectBackoffFactor: Double

 public init(
 mode: Mode = .default,
 timeoutIntervalForRequest: TimeInterval = 60,
Expand All@@ -63,20 +66,29 @@ public struct EventSource: Sendable {
 mode: Mode = .default,
 eventParser: @autoclosure @escaping @Sendable () -> EventParser,
 timeoutIntervalForRequest: TimeInterval = 60,
 timeoutIntervalForResource: TimeInterval = 300
 timeoutIntervalForResource: TimeInterval = 300,
 maxReconnectAttempts: Int = 5,
 reconnectInitialDelay: TimeInterval = 1.0,
 reconnectBackoffFactor: Double = 2.0
 ) {
 self.mode = mode
 self.eventParser = eventParser
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 public func dataTask(for urlRequest: URLRequest) -> DataTask {
 DataTask(
 urlRequest: urlRequest,
 eventParser: eventParser(),
 timeoutIntervalForRequest: timeoutIntervalForRequest,
 timeoutIntervalForResource: timeoutIntervalForResource
 timeoutIntervalForResource: timeoutIntervalForResource,
 maxReconnectAttempts: maxReconnectAttempts,
 reconnectInitialDelay: reconnectInitialDelay,
 reconnectBackoffFactor: reconnectBackoffFactor
 )
 }
}
Expand All@@ -88,8 +100,24 @@ public extension EventSource {
 /// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
 /// it can be started by iterating event stream returned by ``DataTask/events()``.
 final class DataTask: Sendable {

 private let _readyState: Mutex<ReadyState> = Mutex(.none)

 // Reconnection properties
 private let maxReconnectAttempts: Int
 private let reconnectInitialDelay: TimeInterval
 private let reconnectBackoffFactor: Double

 private let _reconnectAttempts: Mutex<Int> = Mutex(0)
 private var reconnectAttempts: Int {
 get {
 _reconnectAttempts.withLock { $0 }
 }
 set {
 _reconnectAttempts.withLock { $0 = newValue }
 }
 }

 /// A value representing the state of the connection.
 public var readyState: ReadyState {
 get {
Expand DownExpand Up@@ -170,12 +198,18 @@ public extension EventSource {
 urlRequest: URLRequest,
 eventParser: EventParser,
 timeoutIntervalForRequest: TimeInterval,
 timeoutIntervalForResource: TimeInterval
 timeoutIntervalForResource: TimeInterval,
 maxReconnectAttempts: Int,
 reconnectInitialDelay: TimeInterval,
 reconnectBackoffFactor: Double
 ) {
 self.urlRequest = urlRequest
 self._eventParser = Mutex(eventParser)
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 /// Creates and returns event stream.
Expand All@@ -188,42 +222,56 @@ public extension EventSource {
 }

 return AsyncStream { continuation in
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)

 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }

 switch event {
 case let .didCompleteWithError(error):
 handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 startSession(stream: continuation)
 readyState = .connecting
 consumed = true
 }
 }

 /// Initializes or reinitializes the SSE session
 private func startSession(stream continuation: AsyncStream<EventType>.Continuation) {
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }
 switch event {
 case let .didCompleteWithError(error):
 self.handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 self.handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 self.parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 }
 }

continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}
 continuation.onTermination = { @Sendable [weak self] _ in
 sessionDelegateTask.cancel()
 Task { self?.close(stream: continuation, urlSession: urlSession) }
 }

 urlSessionDataTask.resume()
 readyState = .connecting
 consumed = true
 urlSessionDataTask.resume()
 }

 /// Helper method for reconnection
 private func attemptReconnect(stream continuation: AsyncStream<EventType>.Continuation) {
 let delay = reconnectInitialDelay * pow(reconnectBackoffFactor, Double(reconnectAttempts - 1))
 Task { [weak self] in
 try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
 self?.startSession(stream: continuation)
 self?.readyState = .connecting
 self?.consumed = true
 }
 }

Expand All@@ -241,9 +289,15 @@ public extension EventSource {
 if let error {
 sendErrorEvent(with: error, stream: continuation)
 }

 // Close connection
 close(stream: continuation, urlSession: urlSession)

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {

Comment on lines

+293
 to 
+294

@RecouseRecouseFeb 24, 2026

Copy link


Copy Markdown

Owner
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Suggested change

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {
 // If an error occurred, an error object will indicate how the request failed, otherwise nil.
 // Attempts to reconnect on error if the limit has not been exceeded.
 if error !=nil, reconnectAttempts < maxReconnectAttempts {
 reconnectAttempts += 1
 attemptReconnect(stream: continuation)
 } else {
 // Close connection if attempts exceeded
 close(stream: continuation, urlSession: urlSession)
 }
 }

 private func handleSessionResponse(
Expand DownExpand Up@@ -321,6 +375,7 @@ public extension EventSource {

 private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
 readyState = .open
 reconnectAttempts = 0 // reset attempts when opening
 continuation.yield(.open)
 }

Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
feat: add resilient SSE session with reconnection/backoff by dogo · Pull Request #43 · Recouse/EventSource · GitHub
Skip to content
feat: add resilient SSE session with reconnection/backoff#43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

dogo
wants to merge
4
commits into

base:main
Choose a base branch

from

 Open


feat: add resilient SSE session with reconnection/backoff
#43
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
97e320b

feat: add resilient SSE session with reconnection/backoff


dogo Oct 1, 2025
38d9edb

Merge remote-tracking branch 'upstream/main' into feature/reconnect


dogo Nov 12, 2025
0a21302

Merge branch 'main' of github.com:Recouse/EventSource into feature/re…


dogo Feb 21, 2026
9b134e8

Use Task.sleep for reconnect delay and reorder methods after init


dogo Feb 23, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


133 changes: 94 additions & 39 deletions
Sources/EventSource/EventSource.swift

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -43,9 +43,12 @@ public struct EventSource: Sendable {
 private let eventParser: @Sendable () -> EventParser

 public var timeoutIntervalForRequest: TimeInterval

 public var timeoutIntervalForResource: TimeInterval

 public var maxReconnectAttempts: Int
 public var reconnectInitialDelay: TimeInterval
 public var reconnectBackoffFactor: Double

 public init(
 mode: Mode = .default,
 timeoutIntervalForRequest: TimeInterval = 60,
Expand All@@ -63,20 +66,29 @@ public struct EventSource: Sendable {
 mode: Mode = .default,
 eventParser: @autoclosure @escaping @Sendable () -> EventParser,
 timeoutIntervalForRequest: TimeInterval = 60,
 timeoutIntervalForResource: TimeInterval = 300
 timeoutIntervalForResource: TimeInterval = 300,
 maxReconnectAttempts: Int = 5,
 reconnectInitialDelay: TimeInterval = 1.0,
 reconnectBackoffFactor: Double = 2.0
 ) {
 self.mode = mode
 self.eventParser = eventParser
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 public func dataTask(for urlRequest: URLRequest) -> DataTask {
 DataTask(
 urlRequest: urlRequest,
 eventParser: eventParser(),
 timeoutIntervalForRequest: timeoutIntervalForRequest,
 timeoutIntervalForResource: timeoutIntervalForResource
 timeoutIntervalForResource: timeoutIntervalForResource,
 maxReconnectAttempts: maxReconnectAttempts,
 reconnectInitialDelay: reconnectInitialDelay,
 reconnectBackoffFactor: reconnectBackoffFactor
 )
 }
}
Expand All@@ -88,8 +100,24 @@ public extension EventSource {
 /// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
 /// it can be started by iterating event stream returned by ``DataTask/events()``.
 final class DataTask: Sendable {

 private let _readyState: Mutex<ReadyState> = Mutex(.none)

 // Reconnection properties
 private let maxReconnectAttempts: Int
 private let reconnectInitialDelay: TimeInterval
 private let reconnectBackoffFactor: Double

 private let _reconnectAttempts: Mutex<Int> = Mutex(0)
 private var reconnectAttempts: Int {
 get {
 _reconnectAttempts.withLock { $0 }
 }
 set {
 _reconnectAttempts.withLock { $0 = newValue }
 }
 }

 /// A value representing the state of the connection.
 public var readyState: ReadyState {
 get {
Expand DownExpand Up@@ -170,12 +198,18 @@ public extension EventSource {
 urlRequest: URLRequest,
 eventParser: EventParser,
 timeoutIntervalForRequest: TimeInterval,
 timeoutIntervalForResource: TimeInterval
 timeoutIntervalForResource: TimeInterval,
 maxReconnectAttempts: Int,
 reconnectInitialDelay: TimeInterval,
 reconnectBackoffFactor: Double
 ) {
 self.urlRequest = urlRequest
 self._eventParser = Mutex(eventParser)
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 /// Creates and returns event stream.
Expand All@@ -188,42 +222,56 @@ public extension EventSource {
 }

 return AsyncStream { continuation in
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)

 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }

 switch event {
 case let .didCompleteWithError(error):
 handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 startSession(stream: continuation)
 readyState = .connecting
 consumed = true
 }
 }

 /// Initializes or reinitializes the SSE session
 private func startSession(stream continuation: AsyncStream<EventType>.Continuation) {
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }
 switch event {
 case let .didCompleteWithError(error):
 self.handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 self.handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 self.parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 }
 }

continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}
 continuation.onTermination = { @Sendable [weak self] _ in
 sessionDelegateTask.cancel()
 Task { self?.close(stream: continuation, urlSession: urlSession) }
 }

 urlSessionDataTask.resume()
 readyState = .connecting
 consumed = true
 urlSessionDataTask.resume()
 }

 /// Helper method for reconnection
 private func attemptReconnect(stream continuation: AsyncStream<EventType>.Continuation) {
 let delay = reconnectInitialDelay * pow(reconnectBackoffFactor, Double(reconnectAttempts - 1))
 Task { [weak self] in
 try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
 self?.startSession(stream: continuation)
 self?.readyState = .connecting
 self?.consumed = true
 }
 }

Expand All@@ -241,9 +289,15 @@ public extension EventSource {
 if let error {
 sendErrorEvent(with: error, stream: continuation)
 }

 // Close connection
 close(stream: continuation, urlSession: urlSession)

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {

Comment on lines

+293
 to 
+294

@RecouseRecouseFeb 24, 2026

Copy link


Copy Markdown

Owner
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Suggested change

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {
 // If an error occurred, an error object will indicate how the request failed, otherwise nil.
 // Attempts to reconnect on error if the limit has not been exceeded.
 if error !=nil, reconnectAttempts < maxReconnectAttempts {
 reconnectAttempts += 1
 attemptReconnect(stream: continuation)
 } else {
 // Close connection if attempts exceeded
 close(stream: continuation, urlSession: urlSession)
 }
 }

 private func handleSessionResponse(
Expand DownExpand Up@@ -321,6 +375,7 @@ public extension EventSource {

 private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
 readyState = .open
 reconnectAttempts = 0 // reset attempts when opening
 continuation.yield(.open)
 }

Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
feat: add resilient SSE session with reconnection/backoff by dogo · Pull Request #43 · Recouse/EventSource · GitHub
Skip to content
feat: add resilient SSE session with reconnection/backoff#43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

dogo
wants to merge
4
commits into

base:main
Choose a base branch

from

 Open


feat: add resilient SSE session with reconnection/backoff
#43
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
97e320b

feat: add resilient SSE session with reconnection/backoff


dogo Oct 1, 2025
38d9edb

Merge remote-tracking branch 'upstream/main' into feature/reconnect


dogo Nov 12, 2025
0a21302

Merge branch 'main' of github.com:Recouse/EventSource into feature/re…


dogo Feb 21, 2026
9b134e8

Use Task.sleep for reconnect delay and reorder methods after init


dogo Feb 23, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


133 changes: 94 additions & 39 deletions
Sources/EventSource/EventSource.swift

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -43,9 +43,12 @@ public struct EventSource: Sendable {
 private let eventParser: @Sendable () -> EventParser

 public var timeoutIntervalForRequest: TimeInterval

 public var timeoutIntervalForResource: TimeInterval

 public var maxReconnectAttempts: Int
 public var reconnectInitialDelay: TimeInterval
 public var reconnectBackoffFactor: Double

 public init(
 mode: Mode = .default,
 timeoutIntervalForRequest: TimeInterval = 60,
Expand All@@ -63,20 +66,29 @@ public struct EventSource: Sendable {
 mode: Mode = .default,
 eventParser: @autoclosure @escaping @Sendable () -> EventParser,
 timeoutIntervalForRequest: TimeInterval = 60,
 timeoutIntervalForResource: TimeInterval = 300
 timeoutIntervalForResource: TimeInterval = 300,
 maxReconnectAttempts: Int = 5,
 reconnectInitialDelay: TimeInterval = 1.0,
 reconnectBackoffFactor: Double = 2.0
 ) {
 self.mode = mode
 self.eventParser = eventParser
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 public func dataTask(for urlRequest: URLRequest) -> DataTask {
 DataTask(
 urlRequest: urlRequest,
 eventParser: eventParser(),
 timeoutIntervalForRequest: timeoutIntervalForRequest,
 timeoutIntervalForResource: timeoutIntervalForResource
 timeoutIntervalForResource: timeoutIntervalForResource,
 maxReconnectAttempts: maxReconnectAttempts,
 reconnectInitialDelay: reconnectInitialDelay,
 reconnectBackoffFactor: reconnectBackoffFactor
 )
 }
}
Expand All@@ -88,8 +100,24 @@ public extension EventSource {
 /// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
 /// it can be started by iterating event stream returned by ``DataTask/events()``.
 final class DataTask: Sendable {

 private let _readyState: Mutex<ReadyState> = Mutex(.none)

 // Reconnection properties
 private let maxReconnectAttempts: Int
 private let reconnectInitialDelay: TimeInterval
 private let reconnectBackoffFactor: Double

 private let _reconnectAttempts: Mutex<Int> = Mutex(0)
 private var reconnectAttempts: Int {
 get {
 _reconnectAttempts.withLock { $0 }
 }
 set {
 _reconnectAttempts.withLock { $0 = newValue }
 }
 }

 /// A value representing the state of the connection.
 public var readyState: ReadyState {
 get {
Expand DownExpand Up@@ -170,12 +198,18 @@ public extension EventSource {
 urlRequest: URLRequest,
 eventParser: EventParser,
 timeoutIntervalForRequest: TimeInterval,
 timeoutIntervalForResource: TimeInterval
 timeoutIntervalForResource: TimeInterval,
 maxReconnectAttempts: Int,
 reconnectInitialDelay: TimeInterval,
 reconnectBackoffFactor: Double
 ) {
 self.urlRequest = urlRequest
 self._eventParser = Mutex(eventParser)
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 /// Creates and returns event stream.
Expand All@@ -188,42 +222,56 @@ public extension EventSource {
 }

 return AsyncStream { continuation in
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)

 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }

 switch event {
 case let .didCompleteWithError(error):
 handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 startSession(stream: continuation)
 readyState = .connecting
 consumed = true
 }
 }

 /// Initializes or reinitializes the SSE session
 private func startSession(stream continuation: AsyncStream<EventType>.Continuation) {
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }
 switch event {
 case let .didCompleteWithError(error):
 self.handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 self.handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 self.parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 }
 }

continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}
 continuation.onTermination = { @Sendable [weak self] _ in
 sessionDelegateTask.cancel()
 Task { self?.close(stream: continuation, urlSession: urlSession) }
 }

 urlSessionDataTask.resume()
 readyState = .connecting
 consumed = true
 urlSessionDataTask.resume()
 }

 /// Helper method for reconnection
 private func attemptReconnect(stream continuation: AsyncStream<EventType>.Continuation) {
 let delay = reconnectInitialDelay * pow(reconnectBackoffFactor, Double(reconnectAttempts - 1))
 Task { [weak self] in
 try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
 self?.startSession(stream: continuation)
 self?.readyState = .connecting
 self?.consumed = true
 }
 }

Expand All@@ -241,9 +289,15 @@ public extension EventSource {
 if let error {
 sendErrorEvent(with: error, stream: continuation)
 }

 // Close connection
 close(stream: continuation, urlSession: urlSession)

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {

Comment on lines

+293
 to 
+294

@RecouseRecouseFeb 24, 2026

Copy link


Copy Markdown

Owner
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Suggested change

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {
 // If an error occurred, an error object will indicate how the request failed, otherwise nil.
 // Attempts to reconnect on error if the limit has not been exceeded.
 if error !=nil, reconnectAttempts < maxReconnectAttempts {
 reconnectAttempts += 1
 attemptReconnect(stream: continuation)
 } else {
 // Close connection if attempts exceeded
 close(stream: continuation, urlSession: urlSession)
 }
 }

 private func handleSessionResponse(
Expand DownExpand Up@@ -321,6 +375,7 @@ public extension EventSource {

 private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
 readyState = .open
 reconnectAttempts = 0 // reset attempts when opening
 continuation.yield(.open)
 }

Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
feat: add resilient SSE session with reconnection/backoff by dogo · Pull Request #43 · Recouse/EventSource · GitHub
Skip to content
feat: add resilient SSE session with reconnection/backoff#43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

dogo
wants to merge
4
commits into

base:main
Choose a base branch

from

 Open


feat: add resilient SSE session with reconnection/backoff
#43
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
97e320b

feat: add resilient SSE session with reconnection/backoff


dogo Oct 1, 2025
38d9edb

Merge remote-tracking branch 'upstream/main' into feature/reconnect


dogo Nov 12, 2025
0a21302

Merge branch 'main' of github.com:Recouse/EventSource into feature/re…


dogo Feb 21, 2026
9b134e8

Use Task.sleep for reconnect delay and reorder methods after init


dogo Feb 23, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


133 changes: 94 additions & 39 deletions
Sources/EventSource/EventSource.swift

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -43,9 +43,12 @@ public struct EventSource: Sendable {
 private let eventParser: @Sendable () -> EventParser

 public var timeoutIntervalForRequest: TimeInterval

 public var timeoutIntervalForResource: TimeInterval

 public var maxReconnectAttempts: Int
 public var reconnectInitialDelay: TimeInterval
 public var reconnectBackoffFactor: Double

 public init(
 mode: Mode = .default,
 timeoutIntervalForRequest: TimeInterval = 60,
Expand All@@ -63,20 +66,29 @@ public struct EventSource: Sendable {
 mode: Mode = .default,
 eventParser: @autoclosure @escaping @Sendable () -> EventParser,
 timeoutIntervalForRequest: TimeInterval = 60,
 timeoutIntervalForResource: TimeInterval = 300
 timeoutIntervalForResource: TimeInterval = 300,
 maxReconnectAttempts: Int = 5,
 reconnectInitialDelay: TimeInterval = 1.0,
 reconnectBackoffFactor: Double = 2.0
 ) {
 self.mode = mode
 self.eventParser = eventParser
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 public func dataTask(for urlRequest: URLRequest) -> DataTask {
 DataTask(
 urlRequest: urlRequest,
 eventParser: eventParser(),
 timeoutIntervalForRequest: timeoutIntervalForRequest,
 timeoutIntervalForResource: timeoutIntervalForResource
 timeoutIntervalForResource: timeoutIntervalForResource,
 maxReconnectAttempts: maxReconnectAttempts,
 reconnectInitialDelay: reconnectInitialDelay,
 reconnectBackoffFactor: reconnectBackoffFactor
 )
 }
}
Expand All@@ -88,8 +100,24 @@ public extension EventSource {
 /// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
 /// it can be started by iterating event stream returned by ``DataTask/events()``.
 final class DataTask: Sendable {

 private let _readyState: Mutex<ReadyState> = Mutex(.none)

 // Reconnection properties
 private let maxReconnectAttempts: Int
 private let reconnectInitialDelay: TimeInterval
 private let reconnectBackoffFactor: Double

 private let _reconnectAttempts: Mutex<Int> = Mutex(0)
 private var reconnectAttempts: Int {
 get {
 _reconnectAttempts.withLock { $0 }
 }
 set {
 _reconnectAttempts.withLock { $0 = newValue }
 }
 }

 /// A value representing the state of the connection.
 public var readyState: ReadyState {
 get {
Expand DownExpand Up@@ -170,12 +198,18 @@ public extension EventSource {
 urlRequest: URLRequest,
 eventParser: EventParser,
 timeoutIntervalForRequest: TimeInterval,
 timeoutIntervalForResource: TimeInterval
 timeoutIntervalForResource: TimeInterval,
 maxReconnectAttempts: Int,
 reconnectInitialDelay: TimeInterval,
 reconnectBackoffFactor: Double
 ) {
 self.urlRequest = urlRequest
 self._eventParser = Mutex(eventParser)
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 /// Creates and returns event stream.
Expand All@@ -188,42 +222,56 @@ public extension EventSource {
 }

 return AsyncStream { continuation in
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)

 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }

 switch event {
 case let .didCompleteWithError(error):
 handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 startSession(stream: continuation)
 readyState = .connecting
 consumed = true
 }
 }

 /// Initializes or reinitializes the SSE session
 private func startSession(stream continuation: AsyncStream<EventType>.Continuation) {
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }
 switch event {
 case let .didCompleteWithError(error):
 self.handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 self.handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 self.parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 }
 }

continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}
 continuation.onTermination = { @Sendable [weak self] _ in
 sessionDelegateTask.cancel()
 Task { self?.close(stream: continuation, urlSession: urlSession) }
 }

 urlSessionDataTask.resume()
 readyState = .connecting
 consumed = true
 urlSessionDataTask.resume()
 }

 /// Helper method for reconnection
 private func attemptReconnect(stream continuation: AsyncStream<EventType>.Continuation) {
 let delay = reconnectInitialDelay * pow(reconnectBackoffFactor, Double(reconnectAttempts - 1))
 Task { [weak self] in
 try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
 self?.startSession(stream: continuation)
 self?.readyState = .connecting
 self?.consumed = true
 }
 }

Expand All@@ -241,9 +289,15 @@ public extension EventSource {
 if let error {
 sendErrorEvent(with: error, stream: continuation)
 }

 // Close connection
 close(stream: continuation, urlSession: urlSession)

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {

Comment on lines

+293
 to 
+294

@RecouseRecouseFeb 24, 2026

Copy link


Copy Markdown

Owner
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Suggested change

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {
 // If an error occurred, an error object will indicate how the request failed, otherwise nil.
 // Attempts to reconnect on error if the limit has not been exceeded.
 if error !=nil, reconnectAttempts < maxReconnectAttempts {
 reconnectAttempts += 1
 attemptReconnect(stream: continuation)
 } else {
 // Close connection if attempts exceeded
 close(stream: continuation, urlSession: urlSession)
 }
 }

 private func handleSessionResponse(
Expand DownExpand Up@@ -321,6 +375,7 @@ public extension EventSource {

 private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
 readyState = .open
 reconnectAttempts = 0 // reset attempts when opening
 continuation.yield(.open)
 }

Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
feat: add resilient SSE session with reconnection/backoff by dogo · Pull Request #43 · Recouse/EventSource · GitHub
Skip to content
feat: add resilient SSE session with reconnection/backoff#43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

dogo
wants to merge
4
commits into

base:main
Choose a base branch

from

 Open


feat: add resilient SSE session with reconnection/backoff
#43
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
97e320b

feat: add resilient SSE session with reconnection/backoff


dogo Oct 1, 2025
38d9edb

Merge remote-tracking branch 'upstream/main' into feature/reconnect


dogo Nov 12, 2025
0a21302

Merge branch 'main' of github.com:Recouse/EventSource into feature/re…


dogo Feb 21, 2026
9b134e8

Use Task.sleep for reconnect delay and reorder methods after init


dogo Feb 23, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


133 changes: 94 additions & 39 deletions
Sources/EventSource/EventSource.swift

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -43,9 +43,12 @@ public struct EventSource: Sendable {
 private let eventParser: @Sendable () -> EventParser

 public var timeoutIntervalForRequest: TimeInterval

 public var timeoutIntervalForResource: TimeInterval

 public var maxReconnectAttempts: Int
 public var reconnectInitialDelay: TimeInterval
 public var reconnectBackoffFactor: Double

 public init(
 mode: Mode = .default,
 timeoutIntervalForRequest: TimeInterval = 60,
Expand All@@ -63,20 +66,29 @@ public struct EventSource: Sendable {
 mode: Mode = .default,
 eventParser: @autoclosure @escaping @Sendable () -> EventParser,
 timeoutIntervalForRequest: TimeInterval = 60,
 timeoutIntervalForResource: TimeInterval = 300
 timeoutIntervalForResource: TimeInterval = 300,
 maxReconnectAttempts: Int = 5,
 reconnectInitialDelay: TimeInterval = 1.0,
 reconnectBackoffFactor: Double = 2.0
 ) {
 self.mode = mode
 self.eventParser = eventParser
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 public func dataTask(for urlRequest: URLRequest) -> DataTask {
 DataTask(
 urlRequest: urlRequest,
 eventParser: eventParser(),
 timeoutIntervalForRequest: timeoutIntervalForRequest,
 timeoutIntervalForResource: timeoutIntervalForResource
 timeoutIntervalForResource: timeoutIntervalForResource,
 maxReconnectAttempts: maxReconnectAttempts,
 reconnectInitialDelay: reconnectInitialDelay,
 reconnectBackoffFactor: reconnectBackoffFactor
 )
 }
}
Expand All@@ -88,8 +100,24 @@ public extension EventSource {
 /// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
 /// it can be started by iterating event stream returned by ``DataTask/events()``.
 final class DataTask: Sendable {

 private let _readyState: Mutex<ReadyState> = Mutex(.none)

 // Reconnection properties
 private let maxReconnectAttempts: Int
 private let reconnectInitialDelay: TimeInterval
 private let reconnectBackoffFactor: Double

 private let _reconnectAttempts: Mutex<Int> = Mutex(0)
 private var reconnectAttempts: Int {
 get {
 _reconnectAttempts.withLock { $0 }
 }
 set {
 _reconnectAttempts.withLock { $0 = newValue }
 }
 }

 /// A value representing the state of the connection.
 public var readyState: ReadyState {
 get {
Expand DownExpand Up@@ -170,12 +198,18 @@ public extension EventSource {
 urlRequest: URLRequest,
 eventParser: EventParser,
 timeoutIntervalForRequest: TimeInterval,
 timeoutIntervalForResource: TimeInterval
 timeoutIntervalForResource: TimeInterval,
 maxReconnectAttempts: Int,
 reconnectInitialDelay: TimeInterval,
 reconnectBackoffFactor: Double
 ) {
 self.urlRequest = urlRequest
 self._eventParser = Mutex(eventParser)
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 /// Creates and returns event stream.
Expand All@@ -188,42 +222,56 @@ public extension EventSource {
 }

 return AsyncStream { continuation in
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)

 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }

 switch event {
 case let .didCompleteWithError(error):
 handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 startSession(stream: continuation)
 readyState = .connecting
 consumed = true
 }
 }

 /// Initializes or reinitializes the SSE session
 private func startSession(stream continuation: AsyncStream<EventType>.Continuation) {
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }
 switch event {
 case let .didCompleteWithError(error):
 self.handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 self.handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 self.parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 }
 }

continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}
 continuation.onTermination = { @Sendable [weak self] _ in
 sessionDelegateTask.cancel()
 Task { self?.close(stream: continuation, urlSession: urlSession) }
 }

 urlSessionDataTask.resume()
 readyState = .connecting
 consumed = true
 urlSessionDataTask.resume()
 }

 /// Helper method for reconnection
 private func attemptReconnect(stream continuation: AsyncStream<EventType>.Continuation) {
 let delay = reconnectInitialDelay * pow(reconnectBackoffFactor, Double(reconnectAttempts - 1))
 Task { [weak self] in
 try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
 self?.startSession(stream: continuation)
 self?.readyState = .connecting
 self?.consumed = true
 }
 }

Expand All@@ -241,9 +289,15 @@ public extension EventSource {
 if let error {
 sendErrorEvent(with: error, stream: continuation)
 }

 // Close connection
 close(stream: continuation, urlSession: urlSession)

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {

Comment on lines

+293
 to 
+294

@RecouseRecouseFeb 24, 2026

Copy link


Copy Markdown

Owner
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Suggested change

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {
 // If an error occurred, an error object will indicate how the request failed, otherwise nil.
 // Attempts to reconnect on error if the limit has not been exceeded.
 if error !=nil, reconnectAttempts < maxReconnectAttempts {
 reconnectAttempts += 1
 attemptReconnect(stream: continuation)
 } else {
 // Close connection if attempts exceeded
 close(stream: continuation, urlSession: urlSession)
 }
 }

 private func handleSessionResponse(
Expand DownExpand Up@@ -321,6 +375,7 @@ public extension EventSource {

 private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
 readyState = .open
 reconnectAttempts = 0 // reset attempts when opening
 continuation.yield(.open)
 }

Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
feat: add resilient SSE session with reconnection/backoff by dogo · Pull Request #43 · Recouse/EventSource · GitHub
Skip to content
feat: add resilient SSE session with reconnection/backoff#43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

dogo
wants to merge
4
commits into

base:main
Choose a base branch

from

 Open


feat: add resilient SSE session with reconnection/backoff
#43
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
97e320b

feat: add resilient SSE session with reconnection/backoff


dogo Oct 1, 2025
38d9edb

Merge remote-tracking branch 'upstream/main' into feature/reconnect


dogo Nov 12, 2025
0a21302

Merge branch 'main' of github.com:Recouse/EventSource into feature/re…


dogo Feb 21, 2026
9b134e8

Use Task.sleep for reconnect delay and reorder methods after init


dogo Feb 23, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


133 changes: 94 additions & 39 deletions
Sources/EventSource/EventSource.swift

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -43,9 +43,12 @@ public struct EventSource: Sendable {
 private let eventParser: @Sendable () -> EventParser

 public var timeoutIntervalForRequest: TimeInterval

 public var timeoutIntervalForResource: TimeInterval

 public var maxReconnectAttempts: Int
 public var reconnectInitialDelay: TimeInterval
 public var reconnectBackoffFactor: Double

 public init(
 mode: Mode = .default,
 timeoutIntervalForRequest: TimeInterval = 60,
Expand All@@ -63,20 +66,29 @@ public struct EventSource: Sendable {
 mode: Mode = .default,
 eventParser: @autoclosure @escaping @Sendable () -> EventParser,
 timeoutIntervalForRequest: TimeInterval = 60,
 timeoutIntervalForResource: TimeInterval = 300
 timeoutIntervalForResource: TimeInterval = 300,
 maxReconnectAttempts: Int = 5,
 reconnectInitialDelay: TimeInterval = 1.0,
 reconnectBackoffFactor: Double = 2.0
 ) {
 self.mode = mode
 self.eventParser = eventParser
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 public func dataTask(for urlRequest: URLRequest) -> DataTask {
 DataTask(
 urlRequest: urlRequest,
 eventParser: eventParser(),
 timeoutIntervalForRequest: timeoutIntervalForRequest,
 timeoutIntervalForResource: timeoutIntervalForResource
 timeoutIntervalForResource: timeoutIntervalForResource,
 maxReconnectAttempts: maxReconnectAttempts,
 reconnectInitialDelay: reconnectInitialDelay,
 reconnectBackoffFactor: reconnectBackoffFactor
 )
 }
}
Expand All@@ -88,8 +100,24 @@ public extension EventSource {
 /// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
 /// it can be started by iterating event stream returned by ``DataTask/events()``.
 final class DataTask: Sendable {

 private let _readyState: Mutex<ReadyState> = Mutex(.none)

 // Reconnection properties
 private let maxReconnectAttempts: Int
 private let reconnectInitialDelay: TimeInterval
 private let reconnectBackoffFactor: Double

 private let _reconnectAttempts: Mutex<Int> = Mutex(0)
 private var reconnectAttempts: Int {
 get {
 _reconnectAttempts.withLock { $0 }
 }
 set {
 _reconnectAttempts.withLock { $0 = newValue }
 }
 }

 /// A value representing the state of the connection.
 public var readyState: ReadyState {
 get {
Expand DownExpand Up@@ -170,12 +198,18 @@ public extension EventSource {
 urlRequest: URLRequest,
 eventParser: EventParser,
 timeoutIntervalForRequest: TimeInterval,
 timeoutIntervalForResource: TimeInterval
 timeoutIntervalForResource: TimeInterval,
 maxReconnectAttempts: Int,
 reconnectInitialDelay: TimeInterval,
 reconnectBackoffFactor: Double
 ) {
 self.urlRequest = urlRequest
 self._eventParser = Mutex(eventParser)
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 /// Creates and returns event stream.
Expand All@@ -188,42 +222,56 @@ public extension EventSource {
 }

 return AsyncStream { continuation in
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)

 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }

 switch event {
 case let .didCompleteWithError(error):
 handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 startSession(stream: continuation)
 readyState = .connecting
 consumed = true
 }
 }

 /// Initializes or reinitializes the SSE session
 private func startSession(stream continuation: AsyncStream<EventType>.Continuation) {
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }
 switch event {
 case let .didCompleteWithError(error):
 self.handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 self.handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 self.parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 }
 }

continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}
 continuation.onTermination = { @Sendable [weak self] _ in
 sessionDelegateTask.cancel()
 Task { self?.close(stream: continuation, urlSession: urlSession) }
 }

 urlSessionDataTask.resume()
 readyState = .connecting
 consumed = true
 urlSessionDataTask.resume()
 }

 /// Helper method for reconnection
 private func attemptReconnect(stream continuation: AsyncStream<EventType>.Continuation) {
 let delay = reconnectInitialDelay * pow(reconnectBackoffFactor, Double(reconnectAttempts - 1))
 Task { [weak self] in
 try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
 self?.startSession(stream: continuation)
 self?.readyState = .connecting
 self?.consumed = true
 }
 }

Expand All@@ -241,9 +289,15 @@ public extension EventSource {
 if let error {
 sendErrorEvent(with: error, stream: continuation)
 }

 // Close connection
 close(stream: continuation, urlSession: urlSession)

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {

Comment on lines

+293
 to 
+294

@RecouseRecouseFeb 24, 2026

Copy link


Copy Markdown

Owner
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Suggested change

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {
 // If an error occurred, an error object will indicate how the request failed, otherwise nil.
 // Attempts to reconnect on error if the limit has not been exceeded.
 if error !=nil, reconnectAttempts < maxReconnectAttempts {
 reconnectAttempts += 1
 attemptReconnect(stream: continuation)
 } else {
 // Close connection if attempts exceeded
 close(stream: continuation, urlSession: urlSession)
 }
 }

 private func handleSessionResponse(
Expand DownExpand Up@@ -321,6 +375,7 @@ public extension EventSource {

 private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
 readyState = .open
 reconnectAttempts = 0 // reset attempts when opening
 continuation.yield(.open)
 }

Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();
feat: add resilient SSE session with reconnection/backoff by dogo · Pull Request #43 · Recouse/EventSource · GitHub
Skip to content
feat: add resilient SSE session with reconnection/backoff#43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

dogo
wants to merge
4
commits into

base:main
Choose a base branch

from

 Open


feat: add resilient SSE session with reconnection/backoff
#43
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
97e320b

feat: add resilient SSE session with reconnection/backoff


dogo Oct 1, 2025
38d9edb

Merge remote-tracking branch 'upstream/main' into feature/reconnect


dogo Nov 12, 2025
0a21302

Merge branch 'main' of github.com:Recouse/EventSource into feature/re…


dogo Feb 21, 2026
9b134e8

Use Task.sleep for reconnect delay and reorder methods after init


dogo Feb 23, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


133 changes: 94 additions & 39 deletions
Sources/EventSource/EventSource.swift

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -43,9 +43,12 @@ public struct EventSource: Sendable {
 private let eventParser: @Sendable () -> EventParser

 public var timeoutIntervalForRequest: TimeInterval

 public var timeoutIntervalForResource: TimeInterval

 public var maxReconnectAttempts: Int
 public var reconnectInitialDelay: TimeInterval
 public var reconnectBackoffFactor: Double

 public init(
 mode: Mode = .default,
 timeoutIntervalForRequest: TimeInterval = 60,
Expand All@@ -63,20 +66,29 @@ public struct EventSource: Sendable {
 mode: Mode = .default,
 eventParser: @autoclosure @escaping @Sendable () -> EventParser,
 timeoutIntervalForRequest: TimeInterval = 60,
 timeoutIntervalForResource: TimeInterval = 300
 timeoutIntervalForResource: TimeInterval = 300,
 maxReconnectAttempts: Int = 5,
 reconnectInitialDelay: TimeInterval = 1.0,
 reconnectBackoffFactor: Double = 2.0
 ) {
 self.mode = mode
 self.eventParser = eventParser
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 public func dataTask(for urlRequest: URLRequest) -> DataTask {
 DataTask(
 urlRequest: urlRequest,
 eventParser: eventParser(),
 timeoutIntervalForRequest: timeoutIntervalForRequest,
 timeoutIntervalForResource: timeoutIntervalForResource
 timeoutIntervalForResource: timeoutIntervalForResource,
 maxReconnectAttempts: maxReconnectAttempts,
 reconnectInitialDelay: reconnectInitialDelay,
 reconnectBackoffFactor: reconnectBackoffFactor
 )
 }
}
Expand All@@ -88,8 +100,24 @@ public extension EventSource {
 /// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
 /// it can be started by iterating event stream returned by ``DataTask/events()``.
 final class DataTask: Sendable {

 private let _readyState: Mutex<ReadyState> = Mutex(.none)

 // Reconnection properties
 private let maxReconnectAttempts: Int
 private let reconnectInitialDelay: TimeInterval
 private let reconnectBackoffFactor: Double

 private let _reconnectAttempts: Mutex<Int> = Mutex(0)
 private var reconnectAttempts: Int {
 get {
 _reconnectAttempts.withLock { $0 }
 }
 set {
 _reconnectAttempts.withLock { $0 = newValue }
 }
 }

 /// A value representing the state of the connection.
 public var readyState: ReadyState {
 get {
Expand DownExpand Up@@ -170,12 +198,18 @@ public extension EventSource {
 urlRequest: URLRequest,
 eventParser: EventParser,
 timeoutIntervalForRequest: TimeInterval,
 timeoutIntervalForResource: TimeInterval
 timeoutIntervalForResource: TimeInterval,
 maxReconnectAttempts: Int,
 reconnectInitialDelay: TimeInterval,
 reconnectBackoffFactor: Double
 ) {
 self.urlRequest = urlRequest
 self._eventParser = Mutex(eventParser)
 self.timeoutIntervalForRequest = timeoutIntervalForRequest
 self.timeoutIntervalForResource = timeoutIntervalForResource
 self.maxReconnectAttempts = maxReconnectAttempts
 self.reconnectInitialDelay = reconnectInitialDelay
 self.reconnectBackoffFactor = reconnectBackoffFactor
 }

 /// Creates and returns event stream.
Expand All@@ -188,42 +222,56 @@ public extension EventSource {
 }

 return AsyncStream { continuation in
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)

 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }

 switch event {
 case let .didCompleteWithError(error):
 handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 startSession(stream: continuation)
 readyState = .connecting
 consumed = true
 }
 }

 /// Initializes or reinitializes the SSE session
 private func startSession(stream continuation: AsyncStream<EventType>.Continuation) {
 let sessionDelegate = SessionDelegate()
 let urlSession = URLSession(
 configuration: urlSessionConfiguration,
 delegate: sessionDelegate,
 delegateQueue: nil
 )
 let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
 let sessionDelegateTask = Task { [weak self] in
 for await event in sessionDelegate.eventStream {
 guard let self else { return }
 switch event {
 case let .didCompleteWithError(error):
 self.handleSessionError(error, stream: continuation, urlSession: urlSession)
 case let .didReceiveResponse(response, completionHandler):
 self.handleSessionResponse(
 response,
 stream: continuation,
 urlSession: urlSession,
 completionHandler: completionHandler
 )
 case let .didReceiveData(data):
 self.parseMessages(from: data, stream: continuation, urlSession: urlSession)
 }
 }
 }

continuation.onTermination = { @Sendable [weak self] _ in
sessionDelegateTask.cancel()
Task { self?.close(stream: continuation, urlSession: urlSession) }
}
 continuation.onTermination = { @Sendable [weak self] _ in
 sessionDelegateTask.cancel()
 Task { self?.close(stream: continuation, urlSession: urlSession) }
 }

 urlSessionDataTask.resume()
 readyState = .connecting
 consumed = true
 urlSessionDataTask.resume()
 }

 /// Helper method for reconnection
 private func attemptReconnect(stream continuation: AsyncStream<EventType>.Continuation) {
 let delay = reconnectInitialDelay * pow(reconnectBackoffFactor, Double(reconnectAttempts - 1))
 Task { [weak self] in
 try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
 self?.startSession(stream: continuation)
 self?.readyState = .connecting
 self?.consumed = true
 }
 }

Expand All@@ -241,9 +289,15 @@ public extension EventSource {
 if let error {
 sendErrorEvent(with: error, stream: continuation)
 }

 // Close connection
 close(stream: continuation, urlSession: urlSession)

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {

Comment on lines

+293
 to 
+294

@RecouseRecouseFeb 24, 2026

Copy link


Copy Markdown

Owner
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Suggested change

 // Attempts to reconnect if the limit has not been exceeded
 if reconnectAttempts < maxReconnectAttempts {
 // If an error occurred, an error object will indicate how the request failed, otherwise nil.
 // Attempts to reconnect on error if the limit has not been exceeded.
 if error !=nil, reconnectAttempts < maxReconnectAttempts {
 reconnectAttempts += 1
 attemptReconnect(stream: continuation)
 } else {
 // Close connection if attempts exceeded
 close(stream: continuation, urlSession: urlSession)
 }
 }

 private func handleSessionResponse(
Expand DownExpand Up@@ -321,6 +375,7 @@ public extension EventSource {

 private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
 readyState = .open
 reconnectAttempts = 0 // reset attempts when opening
 continuation.yield(.open)
 }

Expand Down