From d1b24cd8a0fd3c3a062fad431e2add0b4bede289 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 14:13:54 +0000 Subject: [PATCH 1/6] Overhaul parser for spec compliance; add reconnecting EventSource client Rework the parser to be iterator-driven and generic over any AsyncSequence of bytes: - Errors from the byte stream are rethrown to the consumer instead of being swallowed (previously a network failure left the stream open and the consumer awaiting forever) - Cancellation propagates to the transport; no more detached parse task buffering events without bound - UTF-8 decodes with replacement characters, so invalid bytes no longer fabricate blank lines that split events - A single leading BOM is stripped per spec - Fixed the line splitter emitting a phantom empty line for CR CR - The last event ID persists across events, is stamped on every dispatched event, and commits at dispatch time so an incomplete final block cannot poison the resume ID - ServerSentEvent.type applies the spec's "message" default - Removed the never-populated comment field and unused split helpers URLSession helpers now send Accept: text/event-stream and Cache-Control: no-cache, and validate the response status code and content type, throwing SSEError otherwise. New EventSource client mirrors the spec's reconnection model: automatic reconnect on close or network error, retry-interval handling, Last-Event-ID resumption, and HTTP 204 to stop. The parser now builds on Linux (URLSession pieces are Darwin-gated); CI gains a Linux job. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NkByVdGaxGYiRiGnaQJ8Uc --- .github/workflows/tests.yml | 11 + README.md | 72 +++- .../AsyncSequence+Split.swift | 87 ----- .../AsyncServerSentEvents.swift | 340 +++++++++--------- .../AsyncServerSentEvents/EventSource.swift | 155 ++++++++ .../URLSession+AsyncServerSentEvents.swift | 87 ++++- .../AsyncServerSentEventsTests.swift | 68 ++-- .../SSEParserResilienceTests.swift | 16 +- .../SSESpecComplianceTests.swift | 172 +++++++++ .../SplitFunctionTests.swift | 105 ------ .../URLSessionExtensionTests.swift | 78 +++- 11 files changed, 766 insertions(+), 425 deletions(-) delete mode 100644 Sources/AsyncServerSentEvents/AsyncSequence+Split.swift create mode 100644 Sources/AsyncServerSentEvents/EventSource.swift create mode 100644 Tests/AsyncServerSentEventsTests/SSESpecComplianceTests.swift delete mode 100644 Tests/AsyncServerSentEventsTests/SplitFunctionTests.swift diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index fffc080..64478f3 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -23,3 +23,14 @@ jobs: run: swift test --parallel - name: Test coverage run: swift test --parallel --enable-code-coverage + + linux: + name: Swift 6.0 on Linux + runs-on: ubuntu-24.04 + container: swift:6.0-noble + steps: + - uses: actions/checkout@v4 + - name: Build + run: swift build --build-tests + - name: Run tests + run: swift test --parallel diff --git a/README.md b/README.md index 0a0a98c..cdb7103 100644 --- a/README.md +++ b/README.md @@ -1,54 +1,88 @@ # AsyncServerSentEvents -An implementation of Server-Sent Events (SSE) for Swift using async sequences. Events are parsed from `URLSession.AsyncBytes` and exposed as an `AsyncSequence`. +An implementation of [Server-Sent Events (SSE)](https://html.spec.whatwg.org/multipage/server-sent-events.html) for Swift using async sequences. Events are parsed from any `AsyncSequence` of bytes and exposed as an `AsyncSequence` of `ServerSentEvent` values, with an optional `EventSource`-style reconnecting client built on `URLSession`. ## Usage +### EventSource (reconnecting client) + +`EventSource` mirrors the specification's `EventSource` interface: it sends the `Accept: text/event-stream` header, automatically reconnects when the server closes the stream or the network drops, honors the server's `retry` interval, and resumes with the `Last-Event-ID` header. + ```swift let url = URL(string: "https://example.com/sse")! -let sse = try await URLSession.shared - .asyncBytes(from: url) - .sse() -for try await event in sse { - print(event) +for try await event in EventSource(url: url) { + print(event.type, event.data) } +``` -let lastEventId = await sse.state.lastEventId -let retryInterval = await sse.state.retryInterval +Iteration only ends when the server responds with HTTP 204 (the spec's "stop reconnecting" signal), an unrecoverable response is received (thrown as `SSEError`), or the consuming task is cancelled. To resume a previous session, pass the last event ID you stored: + +```swift +let source = EventSource( + request: URLRequest(url: url), + configuration: .init(retryInterval: 3000, lastEventId: storedId) +) ``` -or +### Single connection (no reconnection) + +The `URLSession` helpers open one connection, send the required headers, and validate that the response is a `200` with a `text/event-stream` content type (throwing `SSEError` otherwise): ```swift let (sse, response) = try await URLSession.shared.serverSentEvents(from: url) for try await event in sse { - print(event) + print(event.data) } +// For manual reconnection: let lastEventId = await sse.state.lastEventId let retryInterval = await sse.state.retryInterval ``` -or +You can also parse an existing byte stream directly: ```swift -let request = URLRequest(url: url) -let (sse, response) = try await URLSession.shared.serverSentEvents(for: request) +let (bytes, response) = try await URLSession.shared.bytes(from: url) -for try await event in sse { +for try await event in bytes.sse() { print(event) } +``` -let lastEventId = await sse.state.lastEventId -let retryInterval = await sse.state.retryInterval +### Parsing any byte stream + +The parser is generic over `AsyncSequence` with `UInt8` elements, so it isn't tied to `URLSession`: + +```swift +let events = AsyncServerSentEvents(bytes: someByteSequence) ``` +Parsing is lazy and driven by iteration: bytes are only consumed as you request events, errors from the byte stream are rethrown to the consumer, and cancelling the consuming task stops the parse. + +## Events + +Each `ServerSentEvent` carries: + +- `data` — the event payload, with multiple `data` fields joined by newlines per the spec +- `name` — the raw `event` field, if present; `type` applies the spec's `"message"` default +- `id` — the `id` field explicitly present on this event's block, if any +- `lastEventId` — the last event ID in effect when the event was dispatched (persists across events, per the spec); this is the value to resume from + ## Supported Features - [x] WHATWG-compliant line parsing (`CR`, `LF`, `CRLF`) -- [x] Strict field parsing (`data`, `id`, `event`) with single-space value trim +- [x] UTF-8 decoding with replacement characters and leading BOM removal +- [x] Strict field parsing (`data`, `id`, `event`, `retry`) with single-space value trim - [x] Comment lines ignored per spec -- [ ] Retry field handling -- [ ] Last-Event-ID persistence across events/reconnects +- [x] Last event ID persistence across events, committed at dispatch time +- [x] `retry` field handling +- [x] Errors from the transport are rethrown to the consumer +- [x] `Accept: text/event-stream` request header and response validation (status and content type) +- [x] Automatic reconnection with `retry` interval and `Last-Event-ID` header (`EventSource`) +- [x] HTTP 204 handled as "stop reconnecting" + +## Platforms + +The parser (`AsyncServerSentEvents`, `ServerSentEvent`) builds anywhere Swift and Foundation are available, including Linux. The `URLSession` helpers and `EventSource` require Apple platforms, where `URLSession.AsyncBytes` is available. diff --git a/Sources/AsyncServerSentEvents/AsyncSequence+Split.swift b/Sources/AsyncServerSentEvents/AsyncSequence+Split.swift deleted file mode 100644 index d982052..0000000 --- a/Sources/AsyncServerSentEvents/AsyncSequence+Split.swift +++ /dev/null @@ -1,87 +0,0 @@ -import Foundation - -extension AsyncSequence { - - /// Returns an async sequence that emits the longest possible subsequences of the sequence, in order, - /// around elements that match the provided test. The values returned in the closure are the separators. - /// - /// - Parameters: - /// - test: A closure that takes a subsequence and returns a subsequence to split on, or nil to indicate that the subsequence should not be split. - /// - window: The number of elements to consider in the lookahead. - /// - maxSplits: The maximum number of splits to perform. - /// - omittingEmptySubsequences: Whether to omit empty subsequences from the result. - /// - Returns: An async stream of subsequences. - func split(_ test: ([Element]) -> [Element]?, window: Int = 1, maxSplits: Int = .max, omittingEmptySubsequences: Bool = true) async throws -> AsyncStream<[Element]> where Element: Sendable, Element: Equatable { - let (stream, continuation) = AsyncStream<[Element]>.makeStream() - var accumulator: [Element] = [] - var lookahead: [Element] = [] - var splitCount = 0 - - for try await element in self { - lookahead.append(element) - if lookahead.count > window { - lookahead.removeFirst() - } - - accumulator.append(element) - - if let match = test(lookahead), splitCount < maxSplits { // lookahead is a match - let elements = lookahead.split(separator: match, maxSplits: 1, omittingEmptySubsequences: false) - var result: [Self.Element] = accumulator.dropLast(elements: lookahead) - result.append(contentsOf: elements.first ?? []) - if !result.isEmpty || !omittingEmptySubsequences { - continuation.yield(result) - } - - result.removeAll() - splitCount += 1 - - accumulator = Array(elements.last ?? []) - } - } - - if !accumulator.isEmpty || !omittingEmptySubsequences { - continuation.yield(accumulator) - } - - continuation.finish() - return stream - } - - /// Returns an async sequence that emits the longest possible subsequences of the sequence, in order, - /// around elements that match the provided separator. - /// - /// - Parameters: - /// - separator: The separator to split on. - /// - maxSplits: The maximum number of splits to perform. - /// - omittingEmptySubsequences: Whether to omit empty subsequences from the result. - /// - Returns: An async stream of subsequences. - func split(separator: Element, maxSplits: Int = .max, omittingEmptySubsequences: Bool = true) async throws -> AsyncStream<[Element]> where Element: Equatable, Element: Sendable { - try await split({ $0.first == separator ? $0 : nil }, window: 1, maxSplits: maxSplits, omittingEmptySubsequences: omittingEmptySubsequences) - } - - /// Returns a sequence that emits the longest possible subsequences of the sequence, in order, - /// around elements that match the provided separator array. - /// - /// - Parameters: - /// - separator: The separator array to split on. - /// - maxSplits: The maximum number of splits to perform. - /// - omittingEmptySubsequences: Whether to omit empty subsequences from the result. - /// - Returns: An async stream of subsequences. - func split(separator: [Element], maxSplits: Int = .max, omittingEmptySubsequences: Bool = true) async throws -> AsyncStream<[Element]> where Element: Equatable, Element: Sendable { - try await split({ $0 == separator ? $0 : nil }, window: separator.count, maxSplits: maxSplits, omittingEmptySubsequences: omittingEmptySubsequences) - } -} - -extension Array where Element: Equatable { - - /// Drop the last elements from the array if they match the provided elements. - /// - Parameter elements: The elements to drop. - /// - Returns: The array with the last elements dropped if they matched. - func dropLast(elements: [Element]) -> [Element] { - guard self.suffix(elements.count) == elements else { - return self - } - return Array(self.dropLast(elements.count)) - } -} diff --git a/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift b/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift index 6281414..a8e516a 100644 --- a/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift +++ b/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift @@ -1,196 +1,214 @@ -// The Swift Programming Language -// https://docs.swift.org/swift-book import Foundation -public struct AsyncServerSentEvents: AsyncSequence { - public typealias Element = Event - typealias Continuation = AsyncStream.Continuation - - public actor State { - public private(set) var retryInterval: Int? - public private(set) var lastEventId: String? - - func updateRetryInterval(_ milliseconds: Int) { - retryInterval = milliseconds +/// A single event parsed from a server-sent event stream. +/// +/// See [the WHATWG specification](https://html.spec.whatwg.org/multipage/server-sent-events.html) +/// for details on the wire format. +public struct ServerSentEvent: Hashable, Sendable { + /// The value of the `id` field explicitly present in this event's block, if any. + public var id: String? + + /// The raw value of the `event` field, if present. See ``type`` for the + /// dispatch type with the spec's `"message"` default applied. + public var name: String? + + /// The event's data. Multiple `data` fields are joined with newlines per the spec. + public var data: String + + /// The last event ID string in effect when this event was dispatched. + /// + /// Per the specification, the last event ID persists across events: an event + /// without an `id` field inherits the most recently seen ID. This is the value + /// to resume from (via the `Last-Event-ID` header) when reconnecting. + public var lastEventId: String? + + /// The event type used for dispatch: ``name`` when non-empty, otherwise + /// `"message"`, matching `EventSource` semantics in the specification. + public var type: String { + if let name, !name.isEmpty { + return name } + return "message" + } - func updateLastEventId(_ value: String) { - lastEventId = value - } + public init(id: String? = nil, name: String? = nil, data: String = "", lastEventId: String? = nil) { + self.id = id + self.name = name + self.data = data + self.lastEventId = lastEventId } - private struct LineSplitter: AsyncSequence { - typealias Element = String + mutating func appending(dataLine: String) { + data.append(dataLine + "\n") + } +} - let bytes: URLSession.AsyncBytes +/// Stream-level state observed while parsing, shared between the parser and the consumer. +public actor SSEState { + /// The reconnection time in milliseconds from the most recent valid `retry` field. + public private(set) var retryInterval: Int? - struct AsyncIterator: AsyncIteratorProtocol { - var iterator: URLSession.AsyncBytes.AsyncIterator - var buffer: [UInt8] = [] - var pendingLine: String? - var sawCarriageReturn = false + /// The last event ID committed by a dispatched event block. + public private(set) var lastEventId: String? - mutating func next() async throws -> String? { - if let line = pendingLine { - pendingLine = nil - return line - } + func updateRetryInterval(_ milliseconds: Int) { + retryInterval = milliseconds + } - while let byte = try await iterator.next() { - if sawCarriageReturn { - sawCarriageReturn = false - if byte == 10 { - let line = String(bytes: buffer, encoding: .utf8) ?? "" - buffer.removeAll(keepingCapacity: true) - return line - } - - let line = String(bytes: buffer, encoding: .utf8) ?? "" - buffer.removeAll(keepingCapacity: true) - - if byte == 13 { - sawCarriageReturn = true - pendingLine = "" - return line - } - - if byte == 10 { - return line - } - - buffer.append(byte) - return line - } + func updateLastEventId(_ value: String) { + lastEventId = value + } +} - if byte == 10 { - let line = String(bytes: buffer, encoding: .utf8) ?? "" - buffer.removeAll(keepingCapacity: true) - return line - } +/// An `AsyncSequence` of ``ServerSentEvent`` values parsed from a stream of UTF-8 bytes. +/// +/// Parsing is driven lazily by iteration: bytes are only read from the underlying +/// sequence as events are requested, cancellation propagates to the byte stream, +/// and errors from the byte stream are rethrown to the consumer. +/// +/// The sequence is single-pass: iterate it once. +public struct AsyncServerSentEvents: AsyncSequence where Base.Element == UInt8 { + public typealias Element = ServerSentEvent + public typealias Event = ServerSentEvent + + /// Stream-level state (`retry` interval and committed last event ID). + public let state: SSEState + + private let base: Base + + public init(bytes: Base) { + self.base = bytes + self.state = SSEState() + } - if byte == 13 { - sawCarriageReturn = true - let line = String(bytes: buffer, encoding: .utf8) ?? "" - buffer.removeAll(keepingCapacity: true) - return line - } + public func makeAsyncIterator() -> AsyncIterator { + AsyncIterator(lines: LineIterator(base: base.makeAsyncIterator()), state: state) + } - buffer.append(byte) + /// Splits a byte stream into lines terminated by LF, CR, or CRLF, decoding + /// UTF-8 with replacement characters and stripping a single leading BOM, + /// as required by the specification. + struct LineIterator where BaseIterator.Element == UInt8 { + var base: BaseIterator + var buffer: [UInt8] = [] + var sawCarriageReturn = false + var isFirstLine = true + var atEnd = false + + mutating func next() async throws -> String? { + guard !atEnd else { return nil } + + while true { + guard let byte = try await base.next() else { + atEnd = true + if buffer.isEmpty { + return nil + } + return makeLine() } if sawCarriageReturn { sawCarriageReturn = false - let line = String(bytes: buffer, encoding: .utf8) ?? "" - buffer.removeAll(keepingCapacity: true) - return line + if byte == 0x0A { // LF completing a CRLF pair + continue + } } - if buffer.isEmpty { - return nil + switch byte { + case 0x0A: + return makeLine() + case 0x0D: + sawCarriageReturn = true + return makeLine() + default: + buffer.append(byte) } - - let line = String(bytes: buffer, encoding: .utf8) ?? "" - buffer.removeAll(keepingCapacity: true) - return line } } - func makeAsyncIterator() -> AsyncIterator { - AsyncIterator(iterator: bytes.makeAsyncIterator()) - } - } - - public struct Event: Hashable, Sendable, Equatable { - public var id: String? - public var name: String? - public var comment: String? - public var data: String - - mutating func appending(commentLine: String) { - if comment != nil { - comment?.append(commentLine + "\n") - } else if !commentLine.isEmpty { - comment = commentLine + "\n" + private mutating func makeLine() -> String { + defer { buffer.removeAll(keepingCapacity: true) } + if isFirstLine { + isFirstLine = false + if buffer.starts(with: [0xEF, 0xBB, 0xBF]) { + buffer.removeFirst(3) + } } + return String(decoding: buffer, as: UTF8.self) } + } - mutating func appending(dataLine: String) { - data.append(dataLine + "\n") - } + public struct AsyncIterator: AsyncIteratorProtocol { + var lines: LineIterator + let state: SSEState + + /// The last event ID buffer; per spec this persists across events and is + /// only committed to ``SSEState`` when a block is dispatched. + var lastEventIdBuffer: String? + var committedLastEventId: String? + + public mutating func next() async throws -> ServerSentEvent? { + var event = ServerSentEvent(data: "") + + while let line = try await lines.next() { + guard !line.isEmpty else { + // Blank line: dispatch the block. The last event ID commits + // even when no event is emitted (e.g. an id-only block). + if let id = lastEventIdBuffer, id != committedLastEventId { + committedLastEventId = id + await state.updateLastEventId(id) + } - mutating func trim() { - data = data.trimmingCharacters(in: .whitespacesAndNewlines) - comment = comment?.trimmingCharacters(in: .whitespacesAndNewlines) - } - } + guard !event.data.isEmpty else { + event = ServerSentEvent(data: "") + continue + } - private let stream: AsyncStream - private let continuation: Continuation - public let state: State - - public init(bytes: URLSession.AsyncBytes) { - state = State() - (stream, continuation) = AsyncStream.makeStream() - - Task { [continuation, state] in - do { - let lines = LineSplitter(bytes: bytes) - - let emptyEvent = Event(data: "") - var event = emptyEvent - for try await line in lines { - if line.isEmpty { - if !event.data.isEmpty { - if event.data.hasSuffix("\n") { - event.data.removeLast() - } - continuation.yield(event) - } - event = emptyEvent - } else { - var elements = line.split(separator: ":", maxSplits: 1, omittingEmptySubsequences: false) - if elements.count == 1 { - elements.append("") - } - - let field = String(elements[0]) - var value = String(elements[1]) - if value.first == " " { - value.removeFirst() - } - - switch field { - case "id": - if !value.contains("\0") { - event.id = value - await state.updateLastEventId(value) - } - case "event": - event.name = value - case "": - continue - case "data": - event.appending(dataLine: value) - case "retry": - if !value.isEmpty, - value.unicodeScalars.allSatisfy({ $0.value >= 48 && $0.value <= 57 }), - let milliseconds = Int(value) { - await state.updateRetryInterval(milliseconds) - } - default: - continue - } + if event.data.hasSuffix("\n") { + event.data.removeLast() } + event.lastEventId = lastEventIdBuffer + return event + } + + var elements = line.split(separator: ":", maxSplits: 1, omittingEmptySubsequences: false) + if elements.count == 1 { + elements.append("") + } + + let field = String(elements[0]) + var value = String(elements[1]) + if value.first == " " { + value.removeFirst() } - continuation.finish() - } catch { - print("ERROR: \(error)") + switch field { + case "data": + event.appending(dataLine: value) + case "event": + event.name = value + case "id": + if !value.contains("\0") { + event.id = value + lastEventIdBuffer = value + } + case "retry": + if !value.isEmpty, + value.unicodeScalars.allSatisfy({ $0.value >= 48 && $0.value <= 57 }), + let milliseconds = Int(value) { + await state.updateRetryInterval(milliseconds) + } + default: + // Comment lines (empty field name) and unknown fields are ignored. + continue + } } - } - } - public func makeAsyncIterator() -> AsyncStream.AsyncIterator { - stream.makeAsyncIterator() + // End of stream: an incomplete final block is discarded per spec, + // including any id it carried. + return nil + } } } + +extension AsyncServerSentEvents: Sendable where Base: Sendable {} diff --git a/Sources/AsyncServerSentEvents/EventSource.swift b/Sources/AsyncServerSentEvents/EventSource.swift new file mode 100644 index 0000000..559dfe4 --- /dev/null +++ b/Sources/AsyncServerSentEvents/EventSource.swift @@ -0,0 +1,155 @@ +#if canImport(Darwin) +import Foundation + +/// A reconnecting server-sent events client modeled on the specification's +/// `EventSource` interface, exposed as an `AsyncSequence` of ``ServerSentEvent``. +/// +/// `EventSource` connects with the required `Accept: text/event-stream` header +/// and, like the browser API, automatically reestablishes the connection when +/// the stream ends or a network error occurs: +/// +/// - It waits the current reconnection time before each attempt (the server can +/// adjust it with the `retry` field; ``Configuration/retryInterval`` is the default). +/// - It sends the `Last-Event-ID` header with the most recently committed event ID. +/// - An HTTP 204 response ends the sequence normally — the specification's way +/// for a server to tell a client to stop reconnecting. +/// - Any other non-200 status, or a non-`text/event-stream` content type, fails +/// the connection by throwing ``SSEError``. +/// +/// Because the stream reconnects on server close, iteration only ends via 204, +/// a thrown error, or cancellation of the consuming task. +/// +/// ```swift +/// for try await event in EventSource(url: url) { +/// print(event.type, event.data) +/// } +/// ``` +public struct EventSource: AsyncSequence, Sendable { + public typealias Element = ServerSentEvent + + public struct Configuration: Sendable { + /// The reconnection delay in milliseconds used until the server provides + /// one via the `retry` field. + public var retryInterval: Int + + /// An event ID to resume from on the first connection, sent as the + /// `Last-Event-ID` header. + public var lastEventId: String? + + public init(retryInterval: Int = 3000, lastEventId: String? = nil) { + self.retryInterval = retryInterval + self.lastEventId = lastEventId + } + } + + private let request: URLRequest + private let session: URLSession + private let configuration: Configuration + + public init(request: URLRequest, session: URLSession = .shared, configuration: Configuration = Configuration()) { + self.request = request + self.session = session + self.configuration = configuration + } + + public init(url: URL, session: URLSession = .shared, configuration: Configuration = Configuration()) { + self.init(request: URLRequest(url: url), session: session, configuration: configuration) + } + + public func makeAsyncIterator() -> AsyncIterator { + AsyncIterator( + request: request, + session: session, + retryInterval: configuration.retryInterval, + lastEventId: configuration.lastEventId + ) + } + + public struct AsyncIterator: AsyncIteratorProtocol { + let request: URLRequest + let session: URLSession + + var retryInterval: Int + var lastEventId: String? + + var current: AsyncServerSentEvents.AsyncIterator? + var currentState: SSEState? + var hasAttemptedConnection = false + var finished = false + + public mutating func next() async throws -> ServerSentEvent? { + guard !finished else { return nil } + + while true { + if current == nil { + if hasAttemptedConnection { + try await Task.sleep(nanoseconds: UInt64(max(0, retryInterval)) * 1_000_000) + } + do { + try await connect() + } catch let error as SSEError { + finished = true + if case .unacceptableStatusCode(204) = error { + return nil + } + throw error + } catch is CancellationError { + finished = true + throw CancellationError() + } catch { + // A network error establishing the connection: retry. + continue + } + } + + do { + if let event = try await current?.next() { + if let id = event.lastEventId { + lastEventId = id + } + return event + } + // The server closed the stream cleanly: reconnect. + await prepareForReconnect() + } catch is CancellationError { + finished = true + throw CancellationError() + } catch { + // A network error mid-stream: reconnect. + await prepareForReconnect() + } + } + } + + private mutating func connect() async throws { + hasAttemptedConnection = true + let preparedRequest = SSERequest.prepared(request, lastEventId: lastEventId) + let (bytes, response) = try await session.bytes(for: preparedRequest) + do { + try SSERequest.validate(response) + } catch { + bytes.task.cancel() + throw error + } + let sse = bytes.sse() + current = sse.makeAsyncIterator() + currentState = sse.state + } + + private mutating func prepareForReconnect() async { + // Pick up retry and id changes from blocks that never dispatched an + // event (retry-only or id-only blocks). + if let state = currentState { + if let interval = await state.retryInterval { + retryInterval = interval + } + if let id = await state.lastEventId { + lastEventId = id + } + } + current = nil + currentState = nil + } + } +} +#endif diff --git a/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift b/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift index 2935348..1399287 100644 --- a/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift +++ b/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift @@ -1,29 +1,92 @@ +#if canImport(Darwin) import Foundation +/// Errors thrown when a server-sent event connection cannot be established. +public enum SSEError: Error, Hashable, Sendable { + /// The server responded with a status code other than 200. + case unacceptableStatusCode(Int) + + /// The server responded with a `Content-Type` other than `text/event-stream`. + case unacceptableContentType(String?) +} + +enum SSERequest { + /// Returns the request with the headers the specification requires an SSE + /// client to send, without overriding values the caller set explicitly. + static func prepared(_ request: URLRequest, lastEventId: String? = nil) -> URLRequest { + var request = request + if request.value(forHTTPHeaderField: "Accept") == nil { + request.setValue("text/event-stream", forHTTPHeaderField: "Accept") + } + if request.value(forHTTPHeaderField: "Cache-Control") == nil { + request.setValue("no-cache", forHTTPHeaderField: "Cache-Control") + } + if let lastEventId, !lastEventId.isEmpty { + request.setValue(lastEventId, forHTTPHeaderField: "Last-Event-ID") + } + return request + } + + /// Validates an HTTP response per the specification: status must be 200 and + /// the content type must be `text/event-stream`. Non-HTTP responses (for + /// example `file:` URLs) are not validated. + static func validate(_ response: URLResponse) throws { + guard let http = response as? HTTPURLResponse else { return } + guard http.statusCode == 200 else { + throw SSEError.unacceptableStatusCode(http.statusCode) + } + let contentType = http.value(forHTTPHeaderField: "Content-Type") + guard let contentType, contentType.lowercased().hasPrefix("text/event-stream") else { + throw SSEError.unacceptableContentType(contentType) + } + } +} + public extension URLSession.AsyncBytes { - func sse() -> AsyncServerSentEvents { + /// Parses these bytes as a server-sent event stream. + func sse() -> AsyncServerSentEvents { AsyncServerSentEvents(bytes: self) } } public extension URLSession { - func serverSentEvents(from url: URL) async throws -> (AsyncServerSentEvents, URLResponse) { - let (bytes, response) = try await bytes(from: url) - return (bytes.sse(), response) + /// Opens a server-sent event stream, sending the `Accept: text/event-stream` + /// header and validating the response status code and content type. + /// + /// - Throws: ``SSEError`` if the response is not a 200 `text/event-stream` response. + func serverSentEvents(from url: URL) async throws -> (AsyncServerSentEvents, URLResponse) { + try await serverSentEvents(for: URLRequest(url: url)) } - func serverSentEvents(for request: URLRequest) async throws -> (AsyncServerSentEvents, URLResponse) { - let (bytes, response) = try await bytes(for: request) - return (bytes.sse(), response) + /// Opens a server-sent event stream, sending the `Accept: text/event-stream` + /// header and validating the response status code and content type. + /// + /// - Throws: ``SSEError`` if the response is not a 200 `text/event-stream` response. + func serverSentEvents(for request: URLRequest) async throws -> (AsyncServerSentEvents, URLResponse) { + try await serverSentEvents(for: request, delegate: nil) } - func serverSentEvents(from url: URL, delegate: URLSessionTaskDelegate) async throws -> (AsyncServerSentEvents, URLResponse) { - let (bytes, response) = try await bytes(from: url, delegate: delegate) - return (bytes.sse(), response) + /// Opens a server-sent event stream, sending the `Accept: text/event-stream` + /// header and validating the response status code and content type. + /// + /// - Throws: ``SSEError`` if the response is not a 200 `text/event-stream` response. + func serverSentEvents(from url: URL, delegate: URLSessionTaskDelegate?) async throws -> (AsyncServerSentEvents, URLResponse) { + try await serverSentEvents(for: URLRequest(url: url), delegate: delegate) } - func serverSentEvents(for request: URLRequest, delegate: URLSessionTaskDelegate) async throws -> (AsyncServerSentEvents, URLResponse) { - let (bytes, response) = try await bytes(for: request, delegate: delegate) + /// Opens a server-sent event stream, sending the `Accept: text/event-stream` + /// header and validating the response status code and content type. + /// + /// - Throws: ``SSEError`` if the response is not a 200 `text/event-stream` response. + func serverSentEvents(for request: URLRequest, delegate: URLSessionTaskDelegate?) async throws -> (AsyncServerSentEvents, URLResponse) { + let (bytes, response) = try await bytes(for: SSERequest.prepared(request), delegate: delegate) + do { + try SSERequest.validate(response) + } catch { + bytes.task.cancel() + throw error + } return (bytes.sse(), response) } } +#endif diff --git a/Tests/AsyncServerSentEventsTests/AsyncServerSentEventsTests.swift b/Tests/AsyncServerSentEventsTests/AsyncServerSentEventsTests.swift index 867132d..b93006a 100644 --- a/Tests/AsyncServerSentEventsTests/AsyncServerSentEventsTests.swift +++ b/Tests/AsyncServerSentEventsTests/AsyncServerSentEventsTests.swift @@ -14,16 +14,22 @@ extension String { return stripped.joined(separator: "\n") } - var asyncBytes: URLSession.AsyncBytes { - get async throws { - let tempFile = FileManager.default.temporaryDirectory - .appendingPathComponent(UUID().uuidString) - var normalized = unindented - if normalized.hasSuffix("\n") && !normalized.hasSuffix("\n\n") { - normalized.append("\n") + var byteStream: AsyncStream { + var normalized = unindented + if normalized.hasSuffix("\n") && !normalized.hasSuffix("\n\n") { + normalized.append("\n") + } + return Array(normalized.utf8).byteStream + } +} + +extension [UInt8] { + var byteStream: AsyncStream { + AsyncStream { continuation in + for byte in self { + continuation.yield(byte) } - try normalized.data(using: .utf8)!.write(to: tempFile) - return try await URLSession.shared.bytes(from: tempFile).0 + continuation.finish() } } } @@ -33,7 +39,7 @@ struct SSEParsingTests { @Test("Basic events should parse into Event structs") func basicEvents() async throws { - let bytes = try await SSETestData.basicEvents.asyncBytes + let bytes = SSETestData.basicEvents.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -48,13 +54,12 @@ struct SSEParsingTests { // Single event with concatenated data fields #expect(events[0].id == nil) #expect(events[0].name == nil) - #expect(events[0].comment == nil) #expect(events[0].data == expectedData) } @Test("Events with IDs should parse ID field") func eventIds() async throws { - let bytes = try await SSETestData.eventIds.asyncBytes + let bytes = SSETestData.eventIds.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -79,7 +84,7 @@ struct SSEParsingTests { @Test("Named events should parse event field") func namedEvents() async throws { - let bytes = try await SSETestData.namedEvents.asyncBytes + let bytes = SSETestData.namedEvents.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -100,7 +105,7 @@ struct SSEParsingTests { @Test("Comments should be ignored") func comments() async throws { - let bytes = try await SSETestData.comments.asyncBytes + let bytes = SSETestData.comments.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -109,7 +114,7 @@ struct SSEParsingTests { @Test("Comment-only blocks should not emit events") func commentOnlyEvent() async throws { - let bytes = try await SSETestData.commentOnlyEvent.asyncBytes + let bytes = SSETestData.commentOnlyEvent.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -118,7 +123,7 @@ struct SSEParsingTests { @Test("Multiple data fields should concatenate with newlines") func multipleDataFields() async throws { - let bytes = try await SSETestData.multipleDataFields.asyncBytes + let bytes = SSETestData.multipleDataFields.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -135,7 +140,7 @@ struct SSEParsingTests { @Test("Mixed fields should parse all fields correctly") func mixedFields() async throws { - let bytes = try await SSETestData.mixedFields.asyncBytes + let bytes = SSETestData.mixedFields.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -143,7 +148,6 @@ struct SSEParsingTests { #expect(events[0].id == "42") // First and only ID field #expect(events[0].name == "update") // First and only event field - #expect(events[0].comment == nil) #expect(events[0].data == """ mixed field event more data @@ -152,7 +156,7 @@ struct SSEParsingTests { @Test("Data fields with leading spaces should preserve spacing") func dataLeadingSpaces() async throws { - let bytes = try await SSETestData.dataLeadingSpaces.asyncBytes + let bytes = SSETestData.dataLeadingSpaces.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -162,7 +166,7 @@ struct SSEParsingTests { @Test("Special characters should be preserved in all fields") func specialCharacters() async throws { - let bytes = try await SSETestData.specialCharacters.asyncBytes + let bytes = SSETestData.specialCharacters.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -177,7 +181,7 @@ struct SSEParsingTests { @Test("Complete event should parse all fields") func completeEvent() async throws { - let bytes = try await SSETestData.completeEvent.asyncBytes + let bytes = SSETestData.completeEvent.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -194,7 +198,7 @@ struct SSEParsingTests { @Test("Empty data fields should create empty events") func emptyDataFields() async throws { - let bytes = try await SSETestData.emptyDataFields.asyncBytes + let bytes = SSETestData.emptyDataFields.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -204,7 +208,7 @@ struct SSEParsingTests { @Test("Parser should handle [DONE] data") func doneAtEnd() async throws { - let bytes = try await SSETestData.doneAtEnd.asyncBytes + let bytes = SSETestData.doneAtEnd.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -218,7 +222,7 @@ struct SSEParsingTests { @Test("Retry field should update state") func retryFieldState() async throws { - let bytes = try await SSETestData.retryIntervals.asyncBytes + let bytes = SSETestData.retryIntervals.byteStream let sse = AsyncServerSentEvents(bytes: bytes) _ = try await sse.collect() @@ -228,7 +232,7 @@ struct SSEParsingTests { @Test("Last event ID should update state") func lastEventIdState() async throws { - let bytes = try await SSETestData.lastEventIdUpdates.asyncBytes + let bytes = SSETestData.lastEventIdUpdates.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -241,7 +245,7 @@ struct SSEParsingTests { @Test("Line endings should handle CR, CRLF, and LF") func lineEndingVariants() async throws { - let bytes = try await SSETestData.lineEndingVariants.asyncBytes + let bytes = SSETestData.lineEndingVariants.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -255,7 +259,7 @@ struct SSEParsingTests { @Test("Missing trailing blank line should discard final event") func noTrailingBlankLine() async throws { - let bytes = try await SSETestData.noTrailingBlankLine.asyncBytes + let bytes = SSETestData.noTrailingBlankLine.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -264,11 +268,11 @@ struct SSEParsingTests { @Test("Events should be Hashable") func eventHashable() async throws { - let event1 = AsyncServerSentEvents.Event(id: "1", name: "test", comment: "comment", data: "data") - let event2 = AsyncServerSentEvents.Event(id: "1", name: "test", comment: "comment", data: "data") - let event3 = AsyncServerSentEvents.Event(id: "2", name: "test", comment: "comment", data: "data") + let event1 = ServerSentEvent(id: "1", name: "test", data: "data") + let event2 = ServerSentEvent(id: "1", name: "test", data: "data") + let event3 = ServerSentEvent(id: "2", name: "test", data: "data") - var eventSet = Set() + var eventSet = Set() eventSet.insert(event1) eventSet.insert(event2) eventSet.insert(event3) @@ -280,7 +284,7 @@ struct SSEParsingTests { @Test("Whitespace-only lines should be ignored") func whitespaceOnlyLines() async throws { - let bytes = try await SSETestData.whitespaceOnlyLines.asyncBytes + let bytes = SSETestData.whitespaceOnlyLines.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() diff --git a/Tests/AsyncServerSentEventsTests/SSEParserResilienceTests.swift b/Tests/AsyncServerSentEventsTests/SSEParserResilienceTests.swift index 469bbce..e280116 100644 --- a/Tests/AsyncServerSentEventsTests/SSEParserResilienceTests.swift +++ b/Tests/AsyncServerSentEventsTests/SSEParserResilienceTests.swift @@ -14,7 +14,7 @@ struct SSEParserResilienceTests { @Test("Parser should ignore invalid field names") func invalidFields() async throws { - let bytes = try await SSETestData.invalidFields.asyncBytes + let bytes = SSETestData.invalidFields.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -24,7 +24,7 @@ struct SSEParserResilienceTests { @Test("Parser should handle unusual whitespace") func unusualWhitespace() async throws { - let bytes = try await SSETestData.unusualWhitespace.asyncBytes + let bytes = SSETestData.unusualWhitespace.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -37,7 +37,7 @@ struct SSEParserResilienceTests { @Test("Parser should normalize mixed line endings") func mixedLineEndings() async throws { - let bytes = try await SSETestData.mixedLineEndings.asyncBytes + let bytes = SSETestData.mixedLineEndings.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -48,7 +48,7 @@ struct SSEParserResilienceTests { @Test("Parser should ignore almost-valid fields") func almostValidFields() async throws { - let bytes = try await SSETestData.almostValidFields.asyncBytes + let bytes = SSETestData.almostValidFields.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -58,7 +58,7 @@ struct SSEParserResilienceTests { @Test("Parser should handle Unicode whitespace") func unicodeWhitespace() async throws { - let bytes = try await SSETestData.unicodeWhitespace.asyncBytes + let bytes = SSETestData.unicodeWhitespace.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -71,7 +71,7 @@ struct SSEParserResilienceTests { @Test("Parser should ignore id fields with null characters") func idWithNull() async throws { - let bytes = try await SSETestData.idWithNull.asyncBytes + let bytes = SSETestData.idWithNull.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() @@ -85,7 +85,7 @@ struct SSEParserResilienceTests { @Test("Parser should process all resilience tests without crashing") func allResilienceTests() async throws { for testCase in SSETestData.allResilienceTests { - let bytes = try await testCase.asyncBytes + let bytes = testCase.byteStream let sse = AsyncServerSentEvents(bytes: bytes) // Should not throw when collecting events @@ -106,7 +106,7 @@ struct SSEParserResilienceTests { dAtA:mixed case """ - let bytes = try await testData.asyncBytes + let bytes = testData.byteStream let sse = AsyncServerSentEvents(bytes: bytes) let events = try await sse.collect() diff --git a/Tests/AsyncServerSentEventsTests/SSESpecComplianceTests.swift b/Tests/AsyncServerSentEventsTests/SSESpecComplianceTests.swift new file mode 100644 index 0000000..ba24565 --- /dev/null +++ b/Tests/AsyncServerSentEventsTests/SSESpecComplianceTests.swift @@ -0,0 +1,172 @@ +import Testing +import Foundation +@testable import AsyncServerSentEvents + +@Suite("Server-Sent Events Spec Compliance") +struct SSESpecComplianceTests { + + @Test("A single leading BOM should be stripped") + func leadingBOM() async throws { + let bytes = ([0xEF, 0xBB, 0xBF] + Array("data: hello\n\n".utf8)).byteStream + let sse = AsyncServerSentEvents(bytes: bytes) + let events = try await sse.collect() + + try #require(events.count == 1) + #expect(events[0].data == "hello") + } + + @Test("A BOM after the stream start should not be stripped") + func nonLeadingBOM() async throws { + let bytes = (Array(":comment\n".utf8) + [0xEF, 0xBB, 0xBF] + Array("data: hello\n\ndata: second\n\n".utf8)).byteStream + let sse = AsyncServerSentEvents(bytes: bytes) + let events = try await sse.collect() + + // "\u{FEFF}data" is not a valid field name, so the first data line is ignored. + try #require(events.count == 1) + #expect(events[0].data == "second") + } + + @Test("Invalid UTF-8 should decode with replacement characters, not fabricate blank lines") + func invalidUTF8() async throws { + var input = Array("data: before".utf8) + input.append(0xFF) // invalid UTF-8 byte + input.append(contentsOf: Array("after\ndata: second line\n\n".utf8)) + + let sse = AsyncServerSentEvents(bytes: input.byteStream) + let events = try await sse.collect() + + // One event: the invalid byte must not split the block in two. + try #require(events.count == 1) + #expect(events[0].data == "before\u{FFFD}after\nsecond line") + } + + @Test("Errors from the byte stream should be rethrown to the consumer") + func streamErrorsPropagate() async throws { + struct TestError: Error {} + + let bytes = AsyncThrowingStream { continuation in + for byte in Array("data: first\n\ndata: second".utf8) { + continuation.yield(byte) + } + continuation.finish(throwing: TestError()) + } + + let sse = AsyncServerSentEvents(bytes: bytes) + var received: [ServerSentEvent] = [] + + await #expect(throws: TestError.self) { + for try await event in sse { + received.append(event) + } + } + + // Events dispatched before the failure are still delivered. + #expect(received.count == 1) + #expect(received.first?.data == "first") + } + + @Test("Last event ID should persist onto subsequent events") + func lastEventIdPersists() async throws { + let input = "id: 1\ndata: first\n\ndata: second\n\nid: 2\ndata: third\n\n" + let sse = AsyncServerSentEvents(bytes: Array(input.utf8).byteStream) + let events = try await sse.collect() + + try #require(events.count == 3) + + #expect(events[0].id == "1") + #expect(events[0].lastEventId == "1") + + // No explicit id, but the last event ID is inherited. + #expect(events[1].id == nil) + #expect(events[1].lastEventId == "1") + + #expect(events[2].id == "2") + #expect(events[2].lastEventId == "2") + } + + @Test("An id-only block should commit the last event ID without dispatching") + func idOnlyBlockCommits() async throws { + let input = "id: 7\n\ndata: payload\n\n" + let sse = AsyncServerSentEvents(bytes: Array(input.utf8).byteStream) + let events = try await sse.collect() + + try #require(events.count == 1) + #expect(events[0].id == nil) + #expect(events[0].lastEventId == "7") + + let lastEventId = await sse.state.lastEventId + #expect(lastEventId == "7") + } + + @Test("An incomplete final block should not commit its id") + func incompleteFinalBlockDiscarded() async throws { + let input = "id: 1\ndata: complete\n\nid: 99\ndata: incomplete" + let sse = AsyncServerSentEvents(bytes: Array(input.utf8).byteStream) + let events = try await sse.collect() + + try #require(events.count == 1) + #expect(events[0].data == "complete") + + let lastEventId = await sse.state.lastEventId + #expect(lastEventId == "1") + } + + @Test("Event type should default to message") + func typeDefaultsToMessage() async throws { + let input = "data: unnamed\n\nevent:\ndata: empty name\n\nevent: custom\ndata: named\n\n" + let sse = AsyncServerSentEvents(bytes: Array(input.utf8).byteStream) + let events = try await sse.collect() + + try #require(events.count == 3) + #expect(events[0].name == nil) + #expect(events[0].type == "message") + #expect(events[1].name == "") + #expect(events[1].type == "message") + #expect(events[2].name == "custom") + #expect(events[2].type == "custom") + } + + @Test("Consecutive carriage returns should split into distinct lines") + func consecutiveCarriageReturns() async throws { + let input = "data: first\r\rdata: second\r\r" + let sse = AsyncServerSentEvents(bytes: Array(input.utf8).byteStream) + let events = try await sse.collect() + + try #require(events.count == 2) + #expect(events[0].data == "first") + #expect(events[1].data == "second") + } + + @Test("Cancellation should stop iteration") + func cancellationStopsIteration() async throws { + // An endless byte stream that keeps producing events. + let bytes = AsyncStream { continuation in + let task = Task { + let block = Array("data: tick\n\n".utf8) + while !Task.isCancelled { + for byte in block { + continuation.yield(byte) + } + try? await Task.sleep(nanoseconds: 1_000_000) + } + continuation.finish() + } + continuation.onTermination = { _ in task.cancel() } + } + + let sse = AsyncServerSentEvents(bytes: bytes) + let consumer = Task { + var count = 0 + for try await _ in sse { + count += 1 + } + return count + } + + try? await Task.sleep(nanoseconds: 50_000_000) + consumer.cancel() + + // The consumer must finish rather than hang once cancelled. + _ = try? await consumer.value + } +} diff --git a/Tests/AsyncServerSentEventsTests/SplitFunctionTests.swift b/Tests/AsyncServerSentEventsTests/SplitFunctionTests.swift deleted file mode 100644 index 4fb6a96..0000000 --- a/Tests/AsyncServerSentEventsTests/SplitFunctionTests.swift +++ /dev/null @@ -1,105 +0,0 @@ -import Testing -import XCTest -@testable import AsyncServerSentEvents - -@Suite("Split Function Tests") -struct SplitFunctionTests { - - @Test("Split by single element") - func splitBySingleElement() async throws { - let sequence = [1, 2, 3, 4, 5, 6, 7, 8, 9].async - let result = try await sequence.split(separator: 5).collect() - #expect(result == [[1, 2, 3, 4], [6, 7, 8, 9]]) - } - - @Test("Split by array of elements") - func splitByArrayOfElements() async throws { - let sequence = [1, 2, 3, 4, 3, 5, 6, 7, 8, 9].async - let result = try await sequence.split(separator: [3, 4]).collect() - #expect(result == [[1, 2], [3, 5, 6, 7, 8, 9]]) - } - - @Test("Split with custom condition") - func splitWithCustomCondition() async throws { - let sequence = [1, 2, 3, 4, 5, 6, 7, 8, 9].async - let result = try await sequence.split({ $0.first?.isMultiple(of: 3) == true ? [$0.first!] : nil }, omittingEmptySubsequences: false).collect() - #expect(result == [[1, 2], [4, 5], [7, 8], []]) - } - - @Test("Split empty sequence") - func splitEmptySequence() async throws { - let emptySequence = [Int]().async - let result = try await emptySequence.split(separator: 1).collect() - #expect(result.isEmpty) - } - - @Test("Split with no matches") - func splitWithNoMatches() async throws { - let sequence = [1, 2, 3, 4, 5].async - let result = try await sequence.split(separator: 6).collect() - #expect(result == [[1, 2, 3, 4, 5]]) - } - - @Test("Split at the beginning") - func splitAtBeginning() async throws { - let sequence = [1, 2, 3, 4, 5].async - let result = try await sequence.split(separator: 1, omittingEmptySubsequences: false).collect() - #expect(result == [[], [2, 3, 4, 5]]) - } - - @Test("Split at the end") - func splitAtEnd() async throws { - let sequence = [1, 2, 3, 4, 5].async - let result = try await sequence.split(separator: 5, omittingEmptySubsequences: false).collect() - #expect(result == [[1, 2, 3, 4], []]) - } - - @Test("Split with consecutive separators") - func splitWithConsecutiveSeparators() async throws { - let sequence = [1, 2, 2, 3, 4, 5].async - let result = try await sequence.split(separator: 2, omittingEmptySubsequences: false).collect() - #expect(result == [[1], [], [3, 4, 5]]) - } - - @Test("Split with larger window") - func splitWithLargerWindow() async throws { - let sequence = [1, 2, 3, 4, 5, 6, 7, 8, 9].async - let result = try await sequence.split({ $0 == [3, 4, 5] ? $0 : nil }, window: 3).collect() - #expect(result == [[1, 2], [6, 7, 8, 9]]) - } - - @Test("Split with maxSplits") - func splitWithMaxSplits() async throws { - let sequence = [1, 2, 3, 4, 5, 6, 3, 8, 9].async - let result = try await sequence.split(separator: 3, maxSplits: 1).collect() - #expect(result == [[1, 2], [4, 5, 6, 3, 8, 9]]) - } - - @Test("Split string on newlines") - func splitStringOnNewlines() async throws { - let sequence = try await "Hello\nWorld\nSwift\nAsync".asyncBytes - let result = try await sequence.split({ elements in - if elements == [10, 13] { - return [10, 13] - } else if elements.first == 10 { - return [10] - } else if elements.first == 13 { - return [13] - } - return nil - }, window: 2, omittingEmptySubsequences: false) - let values = try await result.collect() - #expect(values.compactMap { String(bytes: $0, encoding: .utf8) } == ["Hello", "World", "Swift", "Async"]) - } -} - -extension Array where Element: Sendable { - var async: AsyncStream { - AsyncStream { continuation in - for element in self { - continuation.yield(element) - } - continuation.finish() - } - } -} diff --git a/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift b/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift index 59e166d..e0915a7 100644 --- a/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift +++ b/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift @@ -1,5 +1,7 @@ +#if canImport(Darwin) import Testing import Foundation +@testable import AsyncServerSentEvents @Suite("URLSession Extension Tests") struct URLSessionExtensionTests { @@ -36,5 +38,79 @@ struct URLSessionExtensionTests { let events = try await sse.collect() #expect(events.count == 1) } -} + @Test("Prepared requests should send SSE headers") + func preparedRequestHeaders() { + let request = URLRequest(url: URL(string: "https://example.com/sse")!) + let prepared = SSERequest.prepared(request, lastEventId: "42") + + #expect(prepared.value(forHTTPHeaderField: "Accept") == "text/event-stream") + #expect(prepared.value(forHTTPHeaderField: "Cache-Control") == "no-cache") + #expect(prepared.value(forHTTPHeaderField: "Last-Event-ID") == "42") + } + + @Test("Prepared requests should not override caller-set headers") + func preparedRequestPreservesHeaders() { + var request = URLRequest(url: URL(string: "https://example.com/sse")!) + request.setValue("text/event-stream, application/json", forHTTPHeaderField: "Accept") + let prepared = SSERequest.prepared(request) + + #expect(prepared.value(forHTTPHeaderField: "Accept") == "text/event-stream, application/json") + } + + @Test("Empty last event ID should not send a Last-Event-ID header") + func preparedRequestEmptyLastEventId() { + let request = URLRequest(url: URL(string: "https://example.com/sse")!) + let prepared = SSERequest.prepared(request, lastEventId: "") + + #expect(prepared.value(forHTTPHeaderField: "Last-Event-ID") == nil) + } + + @Test("Response validation should reject non-200 status codes") + func validationRejectsBadStatus() throws { + let url = URL(string: "https://example.com/sse")! + let response = HTTPURLResponse( + url: url, statusCode: 404, httpVersion: "HTTP/1.1", + headerFields: ["Content-Type": "text/event-stream"] + )! + + #expect(throws: SSEError.unacceptableStatusCode(404)) { + try SSERequest.validate(response) + } + } + + @Test("Response validation should reject wrong content types") + func validationRejectsBadContentType() throws { + let url = URL(string: "https://example.com/sse")! + let response = HTTPURLResponse( + url: url, statusCode: 200, httpVersion: "HTTP/1.1", + headerFields: ["Content-Type": "text/html"] + )! + + #expect(throws: SSEError.unacceptableContentType("text/html")) { + try SSERequest.validate(response) + } + } + + @Test("Response validation should accept event streams with parameters") + func validationAcceptsContentTypeParameters() throws { + let url = URL(string: "https://example.com/sse")! + let response = HTTPURLResponse( + url: url, statusCode: 200, httpVersion: "HTTP/1.1", + headerFields: ["Content-Type": "text/event-stream; charset=utf-8"] + )! + + try SSERequest.validate(response) + } + + @Test("Response validation should skip non-HTTP responses") + func validationSkipsNonHTTP() throws { + let response = URLResponse( + url: localFileUrl, mimeType: "text/plain", + expectedContentLength: 0, textEncodingName: nil + ) + + try SSERequest.validate(response) + } +} +#endif From 8971b446b125ac515131282e8887b6c1fc5a32ad Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 14:15:38 +0000 Subject: [PATCH 2/6] Qualify Swift.max to avoid instance method shadowing Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NkByVdGaxGYiRiGnaQJ8Uc --- Sources/AsyncServerSentEvents/EventSource.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Sources/AsyncServerSentEvents/EventSource.swift b/Sources/AsyncServerSentEvents/EventSource.swift index 559dfe4..1491b9c 100644 --- a/Sources/AsyncServerSentEvents/EventSource.swift +++ b/Sources/AsyncServerSentEvents/EventSource.swift @@ -83,7 +83,7 @@ public struct EventSource: AsyncSequence, Sendable { while true { if current == nil { if hasAttemptedConnection { - try await Task.sleep(nanoseconds: UInt64(max(0, retryInterval)) * 1_000_000) + try await Task.sleep(nanoseconds: UInt64(Swift.max(0, retryInterval)) * 1_000_000) } do { try await connect() From 4f28c63bdf3a95d55f76e9cd04b7eef05aabc8ea Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 15:20:45 +0000 Subject: [PATCH 3/6] Address review: strict content-type match, clear stale Last-Event-ID The content-type check now requires the media type to be exactly text/event-stream (parameters allowed), rejecting lookalikes such as text/event-stream+json. An explicitly cleared last event ID (empty id: field) now removes a caller-set Last-Event-ID header on reconnect instead of resuming from the stale value. --- .../URLSession+AsyncServerSentEvents.swift | 16 ++++++++-- .../URLSessionExtensionTests.swift | 29 +++++++++++++++++++ 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift b/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift index 1399287..a04431c 100644 --- a/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift +++ b/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift @@ -21,8 +21,11 @@ enum SSERequest { if request.value(forHTTPHeaderField: "Cache-Control") == nil { request.setValue("no-cache", forHTTPHeaderField: "Cache-Control") } - if let lastEventId, !lastEventId.isEmpty { - request.setValue(lastEventId, forHTTPHeaderField: "Last-Event-ID") + if let lastEventId { + // Per spec, the header is only sent when the last event ID string is + // non-empty. An explicitly empty ID (the stream cleared it) must also + // remove any Last-Event-ID header the caller set on the request. + request.setValue(lastEventId.isEmpty ? nil : lastEventId, forHTTPHeaderField: "Last-Event-ID") } return request } @@ -36,7 +39,14 @@ enum SSERequest { throw SSEError.unacceptableStatusCode(http.statusCode) } let contentType = http.value(forHTTPHeaderField: "Content-Type") - guard let contentType, contentType.lowercased().hasPrefix("text/event-stream") else { + // The media type must be exactly text/event-stream; only parameters may + // follow (e.g. "text/event-stream; charset=utf-8"). A bare prefix match + // would wrongly accept types like text/event-stream+json. + let mediaType = contentType? + .split(separator: ";", maxSplits: 1, omittingEmptySubsequences: false)[0] + .trimmingCharacters(in: .whitespaces) + .lowercased() + guard mediaType == "text/event-stream" else { throw SSEError.unacceptableContentType(contentType) } } diff --git a/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift b/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift index e0915a7..feba37e 100644 --- a/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift +++ b/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift @@ -66,6 +66,20 @@ struct URLSessionExtensionTests { #expect(prepared.value(forHTTPHeaderField: "Last-Event-ID") == nil) } + @Test("An explicitly cleared last event ID should remove a caller-set header") + func preparedRequestClearsStaleLastEventId() { + var request = URLRequest(url: URL(string: "https://example.com/sse")!) + request.setValue("stale", forHTTPHeaderField: "Last-Event-ID") + + // The stream cleared the ID with an empty id: field. + let cleared = SSERequest.prepared(request, lastEventId: "") + #expect(cleared.value(forHTTPHeaderField: "Last-Event-ID") == nil) + + // With no ID tracked at all, the caller's header is left untouched. + let untouched = SSERequest.prepared(request, lastEventId: nil) + #expect(untouched.value(forHTTPHeaderField: "Last-Event-ID") == "stale") + } + @Test("Response validation should reject non-200 status codes") func validationRejectsBadStatus() throws { let url = URL(string: "https://example.com/sse")! @@ -92,6 +106,21 @@ struct URLSessionExtensionTests { } } + @Test("Response validation should reject types that merely share the SSE prefix") + func validationRejectsPrefixLookalikes() throws { + let url = URL(string: "https://example.com/sse")! + for contentType in ["text/event-stream+json", "text/event-streaming"] { + let response = HTTPURLResponse( + url: url, statusCode: 200, httpVersion: "HTTP/1.1", + headerFields: ["Content-Type": contentType] + )! + + #expect(throws: SSEError.unacceptableContentType(contentType)) { + try SSERequest.validate(response) + } + } + } + @Test("Response validation should accept event streams with parameters") func validationAcceptsContentTypeParameters() throws { let url = URL(string: "https://example.com/sse")! From b8caf15c5d51b43e4dd703a0349505348769e1d1 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 17:20:22 +0000 Subject: [PATCH 4/6] Replace URLSession.AsyncBytes with a delegate-based byte stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit URLSession.AsyncBytes is unavailable on Linux, which kept the URLSession helpers and EventSource Apple-only. SSEByteStream streams a data task's body through URLSessionDataDelegate — available on both Darwin and corelibs Foundation — so the entire library now works on Linux. Connections run on a session derived from the caller's session configuration (a session-level delegate is required for streaming); dropping the stream cancels the task and invalidates the session. The delegate-taking serverSentEvents variants are removed since per-task delegates don't compose with the streaming delegate. Header lookup goes through allHeaderFields for consistent behavior on corelibs Foundation, and new tests cover chunk-boundary parsing and error propagation through the byte stream. --- README.md | 2 +- .../AsyncServerSentEvents/EventSource.swift | 9 +- .../AsyncServerSentEvents/SSEByteStream.swift | 114 ++++++++++++++++++ .../URLSession+AsyncServerSentEvents.swift | 49 ++++---- .../URLSessionExtensionTests.swift | 46 ++++++- 5 files changed, 192 insertions(+), 28 deletions(-) create mode 100644 Sources/AsyncServerSentEvents/SSEByteStream.swift diff --git a/README.md b/README.md index cdb7103..968457d 100644 --- a/README.md +++ b/README.md @@ -85,4 +85,4 @@ Each `ServerSentEvent` carries: ## Platforms -The parser (`AsyncServerSentEvents`, `ServerSentEvent`) builds anywhere Swift and Foundation are available, including Linux. The `URLSession` helpers and `EventSource` require Apple platforms, where `URLSession.AsyncBytes` is available. +The library works on Apple platforms and Linux. Instead of `URLSession.AsyncBytes` (which is unavailable on Linux), networking is built on `SSEByteStream`, a small delegate-based byte stream over `URLSessionDataTask`, so the parser, the `URLSession` helpers, and `EventSource` all behave identically everywhere. On Apple platforms, `URLSession.AsyncBytes.sse()` remains available as a convenience. diff --git a/Sources/AsyncServerSentEvents/EventSource.swift b/Sources/AsyncServerSentEvents/EventSource.swift index 1491b9c..36317cd 100644 --- a/Sources/AsyncServerSentEvents/EventSource.swift +++ b/Sources/AsyncServerSentEvents/EventSource.swift @@ -1,5 +1,7 @@ -#if canImport(Darwin) import Foundation +#if canImport(FoundationNetworking) +import FoundationNetworking +#endif /// A reconnecting server-sent events client modeled on the specification's /// `EventSource` interface, exposed as an `AsyncSequence` of ``ServerSentEvent``. @@ -72,7 +74,7 @@ public struct EventSource: AsyncSequence, Sendable { var retryInterval: Int var lastEventId: String? - var current: AsyncServerSentEvents.AsyncIterator? + var current: AsyncServerSentEvents.AsyncIterator? var currentState: SSEState? var hasAttemptedConnection = false var finished = false @@ -124,7 +126,7 @@ public struct EventSource: AsyncSequence, Sendable { private mutating func connect() async throws { hasAttemptedConnection = true let preparedRequest = SSERequest.prepared(request, lastEventId: lastEventId) - let (bytes, response) = try await session.bytes(for: preparedRequest) + let (bytes, response) = try await SSEConnection.open(request: preparedRequest, configuration: session.configuration) do { try SSERequest.validate(response) } catch { @@ -152,4 +154,3 @@ public struct EventSource: AsyncSequence, Sendable { } } } -#endif diff --git a/Sources/AsyncServerSentEvents/SSEByteStream.swift b/Sources/AsyncServerSentEvents/SSEByteStream.swift new file mode 100644 index 0000000..7320879 --- /dev/null +++ b/Sources/AsyncServerSentEvents/SSEByteStream.swift @@ -0,0 +1,114 @@ +import Foundation +#if canImport(FoundationNetworking) +import FoundationNetworking +#endif + +/// An `AsyncSequence` of bytes streamed from a `URLSession` data task. +/// +/// This is the library's replacement for `URLSession.AsyncBytes`: it is built on +/// `URLSessionDataDelegate`, which is available on Linux as well as Apple +/// platforms. Dropping the sequence (or cancelling the consuming task) cancels +/// the underlying data task. +public struct SSEByteStream: AsyncSequence, @unchecked Sendable { + public typealias Element = UInt8 + + /// The underlying data task, exposed for cancellation. + public let task: URLSessionDataTask + + let chunks: AsyncThrowingStream + + public func makeAsyncIterator() -> AsyncIterator { + AsyncIterator(chunks: chunks.makeAsyncIterator()) + } + + public struct AsyncIterator: AsyncIteratorProtocol { + var chunks: AsyncThrowingStream.AsyncIterator + var current: Data = Data() + var offset = 0 + + public mutating func next() async throws -> UInt8? { + while offset >= current.count { + guard let chunk = try await chunks.next() else { return nil } + current = chunk + offset = 0 + } + defer { offset += 1 } + return current[current.startIndex + offset] + } + } + + /// Parses these bytes as a server-sent event stream. + public func sse() -> AsyncServerSentEvents { + AsyncServerSentEvents(bytes: self) + } +} + +/// Opens streaming connections through a session-level delegate so that byte +/// streaming works identically on Darwin and Linux. +enum SSEConnection { + + /// Mirrors `URLSession.bytes(for:)`: performs the request on a session + /// created from the given configuration and returns once the response + /// headers arrive, exposing the body as a byte stream. The session is + /// invalidated when the task completes or the stream is dropped. + static func open(request: URLRequest, configuration: URLSessionConfiguration) async throws -> (SSEByteStream, URLResponse) { + let delegate = StreamingDelegate() + let (chunks, continuation) = AsyncThrowingStream.makeStream() + delegate.dataContinuation = continuation + + let session = URLSession(configuration: configuration, delegate: delegate, delegateQueue: nil) + let task = session.dataTask(with: request) + + continuation.onTermination = { @Sendable _ in + task.cancel() + session.finishTasksAndInvalidate() + } + + let response = try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { (responseContinuation: CheckedContinuation) in + delegate.responseContinuation = responseContinuation + task.resume() + } + } onCancel: { + task.cancel() + } + + return (SSEByteStream(task: task, chunks: chunks), response) + } + + private final class StreamingDelegate: NSObject, URLSessionDataDelegate { + // Only mutated from the session's serial delegate queue, except for the + // initial assignments which happen before the task is resumed. + var responseContinuation: CheckedContinuation? + var dataContinuation: AsyncThrowingStream.Continuation? + + func urlSession( + _ session: URLSession, + dataTask: URLSessionDataTask, + didReceive response: URLResponse, + completionHandler: @escaping (URLSession.ResponseDisposition) -> Void + ) { + responseContinuation?.resume(returning: response) + responseContinuation = nil + completionHandler(.allow) + } + + func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) { + dataContinuation?.yield(data) + } + + func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) { + if let responseContinuation { + responseContinuation.resume(throwing: error ?? URLError(.badServerResponse)) + self.responseContinuation = nil + } + if let error { + dataContinuation?.finish(throwing: error) + } else { + dataContinuation?.finish() + } + dataContinuation = nil + session.finishTasksAndInvalidate() + } + } +} diff --git a/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift b/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift index a04431c..164e257 100644 --- a/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift +++ b/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift @@ -1,5 +1,7 @@ -#if canImport(Darwin) import Foundation +#if canImport(FoundationNetworking) +import FoundationNetworking +#endif /// Errors thrown when a server-sent event connection cannot be established. public enum SSEError: Error, Hashable, Sendable { @@ -38,7 +40,7 @@ enum SSERequest { guard http.statusCode == 200 else { throw SSEError.unacceptableStatusCode(http.statusCode) } - let contentType = http.value(forHTTPHeaderField: "Content-Type") + let contentType = headerValue(in: http, field: "Content-Type") // The media type must be exactly text/event-stream; only parameters may // follow (e.g. "text/event-stream; charset=utf-8"). A bare prefix match // would wrongly accept types like text/event-stream+json. @@ -50,46 +52,52 @@ enum SSERequest { throw SSEError.unacceptableContentType(contentType) } } + + /// Case-insensitive header lookup via `allHeaderFields`, which behaves + /// consistently across Darwin and corelibs Foundation. + static func headerValue(in response: HTTPURLResponse, field: String) -> String? { + for (key, value) in response.allHeaderFields { + if let key = key as? String, key.caseInsensitiveCompare(field) == .orderedSame { + return value as? String + } + } + return nil + } } +#if canImport(Darwin) public extension URLSession.AsyncBytes { /// Parses these bytes as a server-sent event stream. func sse() -> AsyncServerSentEvents { AsyncServerSentEvents(bytes: self) } } +#endif public extension URLSession { /// Opens a server-sent event stream, sending the `Accept: text/event-stream` /// header and validating the response status code and content type. /// - /// - Throws: ``SSEError`` if the response is not a 200 `text/event-stream` response. - func serverSentEvents(from url: URL) async throws -> (AsyncServerSentEvents, URLResponse) { - try await serverSentEvents(for: URLRequest(url: url)) - } - - /// Opens a server-sent event stream, sending the `Accept: text/event-stream` - /// header and validating the response status code and content type. + /// The connection runs on a session derived from this session's + /// `configuration`, so streaming works on Linux as well as Apple platforms. /// /// - Throws: ``SSEError`` if the response is not a 200 `text/event-stream` response. - func serverSentEvents(for request: URLRequest) async throws -> (AsyncServerSentEvents, URLResponse) { - try await serverSentEvents(for: request, delegate: nil) + func serverSentEvents(from url: URL) async throws -> (AsyncServerSentEvents, URLResponse) { + try await serverSentEvents(for: URLRequest(url: url)) } /// Opens a server-sent event stream, sending the `Accept: text/event-stream` /// header and validating the response status code and content type. /// - /// - Throws: ``SSEError`` if the response is not a 200 `text/event-stream` response. - func serverSentEvents(from url: URL, delegate: URLSessionTaskDelegate?) async throws -> (AsyncServerSentEvents, URLResponse) { - try await serverSentEvents(for: URLRequest(url: url), delegate: delegate) - } - - /// Opens a server-sent event stream, sending the `Accept: text/event-stream` - /// header and validating the response status code and content type. + /// The connection runs on a session derived from this session's + /// `configuration`, so streaming works on Linux as well as Apple platforms. /// /// - Throws: ``SSEError`` if the response is not a 200 `text/event-stream` response. - func serverSentEvents(for request: URLRequest, delegate: URLSessionTaskDelegate?) async throws -> (AsyncServerSentEvents, URLResponse) { - let (bytes, response) = try await bytes(for: SSERequest.prepared(request), delegate: delegate) + func serverSentEvents(for request: URLRequest) async throws -> (AsyncServerSentEvents, URLResponse) { + let (bytes, response) = try await SSEConnection.open( + request: SSERequest.prepared(request), + configuration: configuration + ) do { try SSERequest.validate(response) } catch { @@ -99,4 +107,3 @@ public extension URLSession { return (bytes.sse(), response) } } -#endif diff --git a/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift b/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift index feba37e..987f389 100644 --- a/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift +++ b/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift @@ -1,6 +1,8 @@ -#if canImport(Darwin) import Testing import Foundation +#if canImport(FoundationNetworking) +import FoundationNetworking +#endif @testable import AsyncServerSentEvents @Suite("URLSession Extension Tests") @@ -141,5 +143,45 @@ struct URLSessionExtensionTests { try SSERequest.validate(response) } + + @Test("Byte stream should parse events across arbitrary chunk boundaries") + func byteStreamChunkBoundaries() async throws { + let (chunks, continuation) = AsyncThrowingStream.makeStream() + let task = URLSession.shared.dataTask(with: URLRequest(url: localFileUrl)) + let bytes = SSEByteStream(task: task, chunks: chunks) + + continuation.yield(Data("data: hel".utf8)) + continuation.yield(Data()) // empty chunk + continuation.yield(Data("lo\nid: 4".utf8)) + continuation.yield(Data("2\n\ndata: again\n\n".utf8)) + continuation.finish() + + let events = try await bytes.sse().collect() + + try #require(events.count == 2) + #expect(events[0].data == "hello") + #expect(events[0].id == "42") + #expect(events[1].data == "again") + #expect(events[1].lastEventId == "42") + } + + @Test("Byte stream should rethrow chunk errors") + func byteStreamErrorPropagation() async throws { + struct TestError: Error {} + + let (chunks, continuation) = AsyncThrowingStream.makeStream() + let task = URLSession.shared.dataTask(with: URLRequest(url: localFileUrl)) + let bytes = SSEByteStream(task: task, chunks: chunks) + + continuation.yield(Data("data: first\n\n".utf8)) + continuation.finish(throwing: TestError()) + + var received: [ServerSentEvent] = [] + await #expect(throws: TestError.self) { + for try await event in bytes.sse() { + received.append(event) + } + } + #expect(received.count == 1) + } } -#endif From 74b6b6ded5ff7abe0e3531ffb586669c15175b33 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 17:27:11 +0000 Subject: [PATCH 5/6] Serve streaming tests over local HTTP; exhaustive line-splitting tests corelibs URLSession does not support file: URLs for data tasks, so the streaming tests now run against a minimal scripted HTTP server that works on both Darwin and Linux. This also enables true end-to-end coverage: request headers are asserted on the wire, and a new EventSource test drives the full reconnect cycle (Last-Event-ID on reconnect, retry interval pickup, 204 termination). The byte-to-line state machine moves to a top-level SSELineIterator so tests can drive it directly, with a dedicated suite covering CR/LF/CRLF equivalence, empty lines, consecutive CRs, trailing terminators at end of stream, CRLF split across chunk boundaries, multi-byte characters split across chunks, and that NEL/U+2028/U+2029 are content, not boundaries. --- .../AsyncServerSentEvents.swift | 115 ++++++++------- .../EventSourceTests.swift | 55 +++++++ .../SSELineSplittingTests.swift | 138 ++++++++++++++++++ .../TestHTTPServer.swift | 134 +++++++++++++++++ .../URLSessionExtensionTests.swift | 59 +++++--- 5 files changed, 422 insertions(+), 79 deletions(-) create mode 100644 Tests/AsyncServerSentEventsTests/EventSourceTests.swift create mode 100644 Tests/AsyncServerSentEventsTests/SSELineSplittingTests.swift create mode 100644 Tests/AsyncServerSentEventsTests/TestHTTPServer.swift diff --git a/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift b/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift index a8e516a..44e7f8f 100644 --- a/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift +++ b/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift @@ -82,64 +82,11 @@ public struct AsyncServerSentEvents: AsyncSequence where Ba } public func makeAsyncIterator() -> AsyncIterator { - AsyncIterator(lines: LineIterator(base: base.makeAsyncIterator()), state: state) - } - - /// Splits a byte stream into lines terminated by LF, CR, or CRLF, decoding - /// UTF-8 with replacement characters and stripping a single leading BOM, - /// as required by the specification. - struct LineIterator where BaseIterator.Element == UInt8 { - var base: BaseIterator - var buffer: [UInt8] = [] - var sawCarriageReturn = false - var isFirstLine = true - var atEnd = false - - mutating func next() async throws -> String? { - guard !atEnd else { return nil } - - while true { - guard let byte = try await base.next() else { - atEnd = true - if buffer.isEmpty { - return nil - } - return makeLine() - } - - if sawCarriageReturn { - sawCarriageReturn = false - if byte == 0x0A { // LF completing a CRLF pair - continue - } - } - - switch byte { - case 0x0A: - return makeLine() - case 0x0D: - sawCarriageReturn = true - return makeLine() - default: - buffer.append(byte) - } - } - } - - private mutating func makeLine() -> String { - defer { buffer.removeAll(keepingCapacity: true) } - if isFirstLine { - isFirstLine = false - if buffer.starts(with: [0xEF, 0xBB, 0xBF]) { - buffer.removeFirst(3) - } - } - return String(decoding: buffer, as: UTF8.self) - } + AsyncIterator(lines: SSELineIterator(base: base.makeAsyncIterator()), state: state) } public struct AsyncIterator: AsyncIteratorProtocol { - var lines: LineIterator + var lines: SSELineIterator let state: SSEState /// The last event ID buffer; per spec this persists across events and is @@ -212,3 +159,61 @@ public struct AsyncServerSentEvents: AsyncSequence where Ba } extension AsyncServerSentEvents: Sendable where Base: Sendable {} + +/// Splits a byte stream into lines terminated by LF, CR, or CRLF, decoding +/// UTF-8 with replacement characters and stripping a single leading BOM, +/// as required by the specification. +/// +/// The state machine treats CR as an immediate line terminator and swallows an +/// LF that directly follows it, so `CR LF` yields one line boundary while +/// `CR CR` yields two. Only CR, LF, and CRLF are boundaries — other Unicode +/// line separators (NEL, U+2028, U+2029) are ordinary content bytes per spec. +struct SSELineIterator where BaseIterator.Element == UInt8 { + var base: BaseIterator + var buffer: [UInt8] = [] + var sawCarriageReturn = false + var isFirstLine = true + var atEnd = false + + mutating func next() async throws -> String? { + guard !atEnd else { return nil } + + while true { + guard let byte = try await base.next() else { + atEnd = true + if buffer.isEmpty { + return nil + } + return makeLine() + } + + if sawCarriageReturn { + sawCarriageReturn = false + if byte == 0x0A { // LF completing a CRLF pair + continue + } + } + + switch byte { + case 0x0A: + return makeLine() + case 0x0D: + sawCarriageReturn = true + return makeLine() + default: + buffer.append(byte) + } + } + } + + private mutating func makeLine() -> String { + defer { buffer.removeAll(keepingCapacity: true) } + if isFirstLine { + isFirstLine = false + if buffer.starts(with: [0xEF, 0xBB, 0xBF]) { + buffer.removeFirst(3) + } + } + return String(decoding: buffer, as: UTF8.self) + } +} diff --git a/Tests/AsyncServerSentEventsTests/EventSourceTests.swift b/Tests/AsyncServerSentEventsTests/EventSourceTests.swift new file mode 100644 index 0000000..d6f9f39 --- /dev/null +++ b/Tests/AsyncServerSentEventsTests/EventSourceTests.swift @@ -0,0 +1,55 @@ +import Testing +import Foundation +#if canImport(FoundationNetworking) +import FoundationNetworking +#endif +@testable import AsyncServerSentEvents + +@Suite("EventSource") +struct EventSourceTests { + + @Test("EventSource should reconnect with Last-Event-ID and stop on 204") + func reconnectionAndStop() async throws { + let server = try TestHTTPServer(responses: [ + .init(body: "id: 1\ndata: first\n\n"), + .init(body: "retry: 25\nid: 2\ndata: second\n\n"), + // The script is exhausted after this, so the next connection + // receives 204 No Content and the sequence ends. + ]) + defer { server.stop() } + + let url = URL(string: "http://127.0.0.1:\(server.port)/sse")! + let source = EventSource(url: url, configuration: .init(retryInterval: 25)) + + var events: [ServerSentEvent] = [] + for try await event in source { + events.append(event) + } + + try #require(events.count == 2) + #expect(events[0].data == "first") + #expect(events[0].lastEventId == "1") + #expect(events[1].data == "second") + #expect(events[1].lastEventId == "2") + + let requests = server.requests + try #require(requests.count == 3) + #expect(requests[0].lowercased().contains("accept: text/event-stream")) + #expect(!requests[0].lowercased().contains("last-event-id")) + #expect(requests[1].lowercased().contains("last-event-id: 1")) + #expect(requests[2].lowercased().contains("last-event-id: 2")) + } + + @Test("EventSource should fail the connection for non-SSE responses") + func failsOnWrongContentType() async throws { + let server = try TestHTTPServer(responses: [ + .init(contentType: "application/json", body: "{}"), + ]) + defer { server.stop() } + + let url = URL(string: "http://127.0.0.1:\(server.port)/sse")! + await #expect(throws: SSEError.unacceptableContentType("application/json")) { + for try await _ in EventSource(url: url) {} + } + } +} diff --git a/Tests/AsyncServerSentEventsTests/SSELineSplittingTests.swift b/Tests/AsyncServerSentEventsTests/SSELineSplittingTests.swift new file mode 100644 index 0000000..c54073b --- /dev/null +++ b/Tests/AsyncServerSentEventsTests/SSELineSplittingTests.swift @@ -0,0 +1,138 @@ +import Testing +import Foundation +#if canImport(FoundationNetworking) +import FoundationNetworking +#endif +@testable import AsyncServerSentEvents + +/// Exhaustive coverage of the byte-to-line state machine, which must treat +/// LF, CR, and CRLF as equivalent line terminators per the specification. +@Suite("Line Splitting") +struct SSELineSplittingTests { + + private func splitLines(_ bytes: [UInt8]) async throws -> [String] { + var iterator = SSELineIterator(base: bytes.byteStream.makeAsyncIterator()) + var lines: [String] = [] + while let line = try await iterator.next() { + lines.append(line) + } + return lines + } + + private func splitLines(_ text: String) async throws -> [String] { + try await splitLines(Array(text.utf8)) + } + + @Test("All three terminators should be equivalent") + func mixedTerminators() async throws { + let lines = try await splitLines("a\rb\nc\r\nd") + #expect(lines == ["a", "b", "c", "d"]) + } + + @Test("Empty lines should be preserved for every terminator style") + func emptyLines() async throws { + // a CR | CR LF (empty line) | LF (empty line) | b (unterminated) + let lines = try await splitLines("a\r\r\n\nb") + #expect(lines == ["a", "", "", "b"]) + + let crlfOnly = try await splitLines("\r\n\r\n") + #expect(crlfOnly == ["", ""]) + } + + @Test("A CRLF pair should produce exactly one boundary") + func crlfIsSingleBoundary() async throws { + let lines = try await splitLines("a\r\nb") + #expect(lines == ["a", "b"]) + } + + @Test("Consecutive CRs should each terminate a line") + func consecutiveCRs() async throws { + let lines = try await splitLines("a\r\rb") + #expect(lines == ["a", "", "b"]) + } + + @Test("Trailing terminators at end of stream should not add lines") + func trailingTerminators() async throws { + #expect(try await splitLines("a\r") == ["a"]) + #expect(try await splitLines("a\n") == ["a"]) + #expect(try await splitLines("a\r\n") == ["a"]) + } + + @Test("Edge inputs") + func edgeInputs() async throws { + #expect(try await splitLines("") == []) + #expect(try await splitLines("\n") == [""]) + #expect(try await splitLines("\r") == [""]) + #expect(try await splitLines("\r\n") == [""]) + #expect(try await splitLines("a") == ["a"]) + } + + @Test("Other Unicode line separators should not split lines") + func unicodeSeparatorsAreContent() async throws { + // NEL (U+0085), LINE SEPARATOR (U+2028), PARAGRAPH SEPARATOR (U+2029) + let text = "a\u{0085}b\u{2028}c\u{2029}d" + let lines = try await splitLines(text + "\n") + #expect(lines == [text]) + + // And they must survive through the full parser as data content. + let sse = AsyncServerSentEvents(bytes: Array("data: \(text)\n\n".utf8).byteStream) + let events = try await sse.collect() + try #require(events.count == 1) + #expect(events[0].data == text) + } + + @Test("A trailing CR should act as a blank-line dispatch boundary") + func trailingCRDispatches() async throws { + let sse = AsyncServerSentEvents(bytes: Array("data: hi\n\r".utf8).byteStream) + let events = try await sse.collect() + + try #require(events.count == 1) + #expect(events[0].data == "hi") + } + + @Test("A leading BOM should be stripped even with CRLF terminators") + func bomWithCRLF() async throws { + let bytes: [UInt8] = [0xEF, 0xBB, 0xBF] + Array("data: x\r\n\r\n".utf8) + let events = try await AsyncServerSentEvents(bytes: bytes.byteStream).collect() + + try #require(events.count == 1) + #expect(events[0].data == "x") + } + + @Test("CRLF split across chunk boundaries should stay one boundary") + func crlfAcrossChunks() async throws { + let (chunks, continuation) = AsyncThrowingStream.makeStream() + let task = URLSession.shared.dataTask(with: URLRequest(url: URL(string: "https://example.com/sse")!)) + let bytes = SSEByteStream(task: task, chunks: chunks) + + continuation.yield(Data("data: a\r".utf8)) + continuation.yield(Data("\n\r".utf8)) + continuation.yield(Data("\ndata: b\n\n".utf8)) + continuation.finish() + + let events = try await bytes.sse().collect() + + try #require(events.count == 2) + #expect(events[0].data == "a") + #expect(events[1].data == "b") + } + + @Test("Multi-byte characters split across chunk boundaries should decode intact") + func multiByteAcrossChunks() async throws { + let (chunks, continuation) = AsyncThrowingStream.makeStream() + let task = URLSession.shared.dataTask(with: URLRequest(url: URL(string: "https://example.com/sse")!)) + let bytes = SSEByteStream(task: task, chunks: chunks) + + let payload = Array("data: é🎉\n\n".utf8) + // Split in the middle of the emoji's four-byte sequence. + let splitIndex = payload.count - 5 + continuation.yield(Data(payload[..= 0, "socket() failed") + + var reuse: Int32 = 1 + setsockopt(listenSocket, SOL_SOCKET, SO_REUSEADDR, &reuse, socklen_t(MemoryLayout.size)) + + var address = sockaddr_in() + address.sin_family = sa_family_t(AF_INET) + address.sin_addr = in_addr(s_addr: UInt32(0x7F000001).bigEndian) // 127.0.0.1 + address.sin_port = 0 // any free port + + let bound = withUnsafePointer(to: &address) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { + bind(listenSocket, $0, socklen_t(MemoryLayout.size)) + } + } + precondition(bound == 0, "bind() failed") + precondition(listen(listenSocket, 8) == 0, "listen() failed") + + var assigned = sockaddr_in() + var length = socklen_t(MemoryLayout.size) + _ = withUnsafeMutablePointer(to: &assigned) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { + getsockname(listenSocket, $0, &length) + } + } + port = UInt16(bigEndian: assigned.sin_port) + + let socketDescriptor = listenSocket + var remaining = responses + let thread = Thread { [weak self] in + while true { + let client = accept(socketDescriptor, nil, nil) + guard client >= 0 else { return } + self?.record(Self.readRequestHead(from: client)) + let response = remaining.isEmpty + ? Response(status: "204 No Content", contentType: nil, body: "") + : remaining.removeFirst() + Self.send(response: response, to: client) + close(client) + } + } + thread.start() + } + + func stop() { + close(listenSocket) + } + + private func record(_ request: String) { + lock.lock() + receivedRequests.append(request) + lock.unlock() + } + + private static func readRequestHead(from fd: Int32) -> String { + var data = Data() + var buffer = [UInt8](repeating: 0, count: 4096) + let terminator = Data("\r\n\r\n".utf8) + while data.range(of: terminator) == nil && data.count < 16384 { + let count = recv(fd, &buffer, buffer.count, 0) + guard count > 0 else { break } + data.append(contentsOf: buffer[0.. Int in + #if canImport(Darwin) + Darwin.send(fd, pointer.baseAddress! + sent, bytes.count - sent, 0) + #else + Glibc.send(fd, pointer.baseAddress! + sent, bytes.count - sent, Int32(MSG_NOSIGNAL)) + #endif + } + guard result > 0 else { break } + sent += result + } + } +} diff --git a/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift b/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift index 987f389..ed5fe09 100644 --- a/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift +++ b/Tests/AsyncServerSentEventsTests/URLSessionExtensionTests.swift @@ -8,37 +8,48 @@ import FoundationNetworking @Suite("URLSession Extension Tests") struct URLSessionExtensionTests { - let localFileUrl: URL + static let eventBody = "data: hello\nid: 123\nevent: message\n\n" - init() throws { - let tempFile = FileManager.default.temporaryDirectory - .appendingPathComponent(UUID().uuidString) - let body = "data: hello\nid: 123\nevent: message\n\n" - try body.data(using: .utf8)!.write(to: tempFile) - - localFileUrl = tempFile - } + @Test("serverSentEvents(from:) should stream events over HTTP") + func testServerSentEventsFrom() async throws { + let server = try TestHTTPServer(responses: [.init(body: Self.eventBody)]) + defer { server.stop() } - @Test("sse()") - func testSSE() async throws { - let (sse, _) = try await URLSession.shared.serverSentEvents(from: localFileUrl) + let url = URL(string: "http://127.0.0.1:\(server.port)/sse")! + let (sse, response) = try await URLSession.shared.serverSentEvents(from: url) let events = try await sse.collect() - #expect(events.count == 1) - } - @Test("serverSentEvents(from:)") - func testServerSentEventsFrom() async throws { - let (sse, _) = try await URLSession.shared.serverSentEvents(from: localFileUrl) - let events = try await sse.collect() - #expect(events.count == 1) + try #require(events.count == 1) + #expect(events[0].data == "hello") + #expect(events[0].id == "123") + #expect((response as? HTTPURLResponse)?.statusCode == 200) } - @Test("serverSentEvents(for:)") + @Test("serverSentEvents(for:) should send the SSE request headers") func testServerSentEventsFor() async throws { - let request = URLRequest(url: localFileUrl) + let server = try TestHTTPServer(responses: [.init(body: Self.eventBody)]) + defer { server.stop() } + + let request = URLRequest(url: URL(string: "http://127.0.0.1:\(server.port)/sse")!) let (sse, _) = try await URLSession.shared.serverSentEvents(for: request) let events = try await sse.collect() + #expect(events.count == 1) + + let requests = server.requests + try #require(requests.count == 1) + #expect(requests[0].lowercased().contains("accept: text/event-stream")) + } + + @Test("serverSentEvents should reject non-SSE responses over HTTP") + func testRejectsWrongContentTypeOverHTTP() async throws { + let server = try TestHTTPServer(responses: [.init(contentType: "text/html", body: "")]) + defer { server.stop() } + + let url = URL(string: "http://127.0.0.1:\(server.port)/sse")! + await #expect(throws: SSEError.unacceptableContentType("text/html")) { + _ = try await URLSession.shared.serverSentEvents(from: url) + } } @Test("Prepared requests should send SSE headers") @@ -137,7 +148,7 @@ struct URLSessionExtensionTests { @Test("Response validation should skip non-HTTP responses") func validationSkipsNonHTTP() throws { let response = URLResponse( - url: localFileUrl, mimeType: "text/plain", + url: URL(string: "file:///tmp/stream")!, mimeType: "text/plain", expectedContentLength: 0, textEncodingName: nil ) @@ -147,7 +158,7 @@ struct URLSessionExtensionTests { @Test("Byte stream should parse events across arbitrary chunk boundaries") func byteStreamChunkBoundaries() async throws { let (chunks, continuation) = AsyncThrowingStream.makeStream() - let task = URLSession.shared.dataTask(with: URLRequest(url: localFileUrl)) + let task = URLSession.shared.dataTask(with: URLRequest(url: URL(string: "https://example.com/sse")!)) let bytes = SSEByteStream(task: task, chunks: chunks) continuation.yield(Data("data: hel".utf8)) @@ -170,7 +181,7 @@ struct URLSessionExtensionTests { struct TestError: Error {} let (chunks, continuation) = AsyncThrowingStream.makeStream() - let task = URLSession.shared.dataTask(with: URLRequest(url: localFileUrl)) + let task = URLSession.shared.dataTask(with: URLRequest(url: URL(string: "https://example.com/sse")!)) let bytes = SSEByteStream(task: task, chunks: chunks) continuation.yield(Data("data: first\n\n".utf8)) From 22d20b38b683674e45d61efbacf424781fce94f6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 17:29:19 +0000 Subject: [PATCH 6/6] Initialize all stored properties before closures capture self in TestHTTPServer --- .../TestHTTPServer.swift | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/Tests/AsyncServerSentEventsTests/TestHTTPServer.swift b/Tests/AsyncServerSentEventsTests/TestHTTPServer.swift index b5fdba6..158030f 100644 --- a/Tests/AsyncServerSentEventsTests/TestHTTPServer.swift +++ b/Tests/AsyncServerSentEventsTests/TestHTTPServer.swift @@ -42,11 +42,11 @@ final class TestHTTPServer: @unchecked Sendable { #else let socketType = Int32(SOCK_STREAM.rawValue) #endif - listenSocket = socket(AF_INET, socketType, 0) - precondition(listenSocket >= 0, "socket() failed") + let socketDescriptor = socket(AF_INET, socketType, 0) + precondition(socketDescriptor >= 0, "socket() failed") var reuse: Int32 = 1 - setsockopt(listenSocket, SOL_SOCKET, SO_REUSEADDR, &reuse, socklen_t(MemoryLayout.size)) + setsockopt(socketDescriptor, SOL_SOCKET, SO_REUSEADDR, &reuse, socklen_t(MemoryLayout.size)) var address = sockaddr_in() address.sin_family = sa_family_t(AF_INET) @@ -55,22 +55,22 @@ final class TestHTTPServer: @unchecked Sendable { let bound = withUnsafePointer(to: &address) { $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { - bind(listenSocket, $0, socklen_t(MemoryLayout.size)) + bind(socketDescriptor, $0, socklen_t(MemoryLayout.size)) } } precondition(bound == 0, "bind() failed") - precondition(listen(listenSocket, 8) == 0, "listen() failed") + precondition(listen(socketDescriptor, 8) == 0, "listen() failed") var assigned = sockaddr_in() var length = socklen_t(MemoryLayout.size) _ = withUnsafeMutablePointer(to: &assigned) { $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { - getsockname(listenSocket, $0, &length) + getsockname(socketDescriptor, $0, &length) } } - port = UInt16(bigEndian: assigned.sin_port) - let socketDescriptor = listenSocket + listenSocket = socketDescriptor + port = UInt16(bigEndian: assigned.sin_port) var remaining = responses let thread = Thread { [weak self] in while true {