diff --git a/.swiftpm/xcode/xcshareddata/xcschemes/AsyncServerSentEvents.xcscheme b/.swiftpm/xcode/xcshareddata/xcschemes/AsyncServerSentEvents.xcscheme
new file mode 100644
index 0000000..e93a893
--- /dev/null
+++ b/.swiftpm/xcode/xcshareddata/xcschemes/AsyncServerSentEvents.xcscheme
@@ -0,0 +1,80 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..74632fc
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Zac White
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/PLAN.md b/PLAN.md
new file mode 100644
index 0000000..5278d5b
--- /dev/null
+++ b/PLAN.md
@@ -0,0 +1,35 @@
+# Improve the SSE Parser for 0.2.0
+
+## Summary
+
+Extract the parsing logic into `SSEParser`, a synchronous, value-semantic incremental parser, and rebuild the async APIs as thin wrappers over it. Everything that exists in 0.1.0 keeps compiling and behaving identically; the new surface is small and non-throwing.
+
+## Public API Additions
+
+- `SSEParser`, a `Sendable` **struct** (value semantics, no locking):
+ - `mutating func consume(_ byte: UInt8) -> ServerSentEvent?`
+ - `mutating func consume(_ bytes: some Sequence) -> [ServerSentEvent]`
+ - `mutating func finish()` — end of stream: applies an unterminated trailing field line (so a trailing `retry` still counts, matching 0.1.0), discards the incomplete block per spec, and resets for a new stream (BOM stripped again) while retaining reconnect state.
+ - `private(set) var lastEventId: String?` and `retryInterval: Int?` — the committed reconnect state.
+ - `static func parse(_ bytes: some Sequence) -> [ServerSentEvent]` for fully buffered input.
+ - Nothing throws: per spec, invalid UTF-8 decodes with replacement characters and malformed lines are ignored, so parsing is total.
+- Generic `AsyncSequence where Element == UInt8` extension providing `.sse()`, replacing (source-compatibly) the concrete `SSEByteStream` and Darwin-only `URLSession.AsyncBytes` extensions.
+
+## Deliberately cut from the other plan
+
+- `ServerSentEventBlock` — its `data == nil` sentinel duplicated `ServerSentEvent` awkwardly; block-level state consumers just read `lastEventId`/`retryInterval` off the parser.
+- `SSEParser.Limits` and throwing `consume` — speculative; can be added additively in a later release if ever needed.
+- `.sseBlocks()` adapters — no consumer.
+
+## Implementation Changes
+
+- New `SSEParser.swift`: internal `SSELineSplitter` (byte → line state machine, CR/LF/CRLF, leading BOM) + `SSEParser` (field parsing directly from bytes, decoding only recognized values; no `String.split`).
+- `AsyncServerSentEvents.AsyncIterator` becomes a loop feeding bytes to an `SSEParser`, mirroring committed state into the existing public `SSEState` actor only when it changes.
+- `EventSource` reads reconnect state synchronously from its iterator's parser instead of awaiting the `SSEState` actor.
+- Behavioral compatibility pinned: id commits only at blank line; retry applies per line (including an unterminated final line at EOF); state-only blocks observable; comment-only blocks silent; per-stream last-event-ID buffer resets across reconnects while the committed value persists.
+
+## Test Plan
+
+- All 63 existing tests pass unchanged (line-splitting tests port to the sync splitter, same assertions).
+- New `SSEParserTests`: every existing fixture parsed sync vs async with identical events and final state; chunk-boundary invariance (sizes 1/2/3/7/16/1024); emission timing; state-only blocks; `finish()` reuse incl. BOM re-strip; trailing unterminated `retry` line.
+- Clean `swift build` and `swift test`.
diff --git a/Package.swift b/Package.swift
index f04ff22..d2d1107 100644
--- a/Package.swift
+++ b/Package.swift
@@ -6,11 +6,11 @@ import PackageDescription
let package = Package(
name: "AsyncServerSentEvents",
platforms: [
- .iOS(.v17),
- .tvOS(.v17),
- .macOS(.v14),
- .watchOS(.v9),
- .macCatalyst(.v17),
+ .iOS(.v13),
+ .tvOS(.v13),
+ .macOS(.v10_15),
+ .watchOS(.v6),
+ .macCatalyst(.v13),
.visionOS(.v1)
],
products: [
diff --git a/README.md b/README.md
index 968457d..2dcd814 100644
--- a/README.md
+++ b/README.md
@@ -53,14 +53,36 @@ for try await event in bytes.sse() {
### Parsing any byte stream
-The parser is generic over `AsyncSequence` with `UInt8` elements, so it isn't tied to `URLSession`:
+The parser is generic over `AsyncSequence` with `UInt8` elements, so it isn't tied to `URLSession`. Any byte sequence can be wrapped directly or via the `.sse()` convenience:
```swift
let events = AsyncServerSentEvents(bytes: someByteSequence)
+// or equivalently:
+for try await event in someByteSequence.sse() { ... }
```
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.
+### Incremental and buffered parsing (no async required)
+
+`SSEParser` is the synchronous core the async APIs are built on. Use it directly when bytes arrive through something other than an `AsyncSequence` — a delegate callback, a WebSocket frame — or when you already have the whole payload:
+
+```swift
+// Incremental: feed chunks as they arrive.
+var parser = SSEParser()
+for event in parser.consume(chunk) {
+ print(event.type, event.data)
+}
+// At end of stream: discards an incomplete trailing block (per spec) and
+// resets for a new stream, keeping lastEventId/retryInterval for reconnection.
+parser.finish()
+
+// Buffered: parse a complete payload in one call.
+let events = SSEParser.parse(data)
+```
+
+`SSEParser` is a value type, never throws (malformed input is handled by the spec's own rules), and exposes `lastEventId` and `retryInterval` for implementing reconnection.
+
## Events
Each `ServerSentEvent` carries:
@@ -72,6 +94,7 @@ Each `ServerSentEvent` carries:
## Supported Features
+- [x] Synchronous incremental parsing (`SSEParser`) usable without async sequences
- [x] WHATWG-compliant line parsing (`CR`, `LF`, `CRLF`)
- [x] UTF-8 decoding with replacement characters and leading BOM removal
- [x] Strict field parsing (`data`, `id`, `event`, `retry`) with single-space value trim
diff --git a/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift b/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift
index 44e7f8f..4a6da6a 100644
--- a/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift
+++ b/Sources/AsyncServerSentEvents/AsyncServerSentEvents.swift
@@ -37,10 +37,6 @@ public struct ServerSentEvent: Hashable, Sendable {
self.data = data
self.lastEventId = lastEventId
}
-
- mutating func appending(dataLine: String) {
- data.append(dataLine + "\n")
- }
}
/// Stream-level state observed while parsing, shared between the parser and the consumer.
@@ -64,7 +60,9 @@ public actor SSEState {
///
/// 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.
+/// and errors from the byte stream are rethrown to the consumer. The parsing
+/// itself is performed by ``SSEParser``, which is also available directly for
+/// non-async byte sources.
///
/// The sequence is single-pass: iterate it once.
public struct AsyncServerSentEvents: AsyncSequence where Base.Element == UInt8 {
@@ -82,138 +80,62 @@ public struct AsyncServerSentEvents: AsyncSequence where Ba
}
public func makeAsyncIterator() -> AsyncIterator {
- AsyncIterator(lines: SSELineIterator(base: base.makeAsyncIterator()), state: state)
+ AsyncIterator(base: base.makeAsyncIterator(), state: state)
}
public struct AsyncIterator: AsyncIteratorProtocol {
- var lines: SSELineIterator
+ var base: Base.AsyncIterator
+ var parser = SSEParser()
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?
+ private var mirroredRetryInterval: Int?
+ private var mirroredLastEventId: String?
+ private var finished = false
- 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)
- }
-
- guard !event.data.isEmpty else {
- event = ServerSentEvent(data: "")
- 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("")
- }
+ init(base: Base.AsyncIterator, state: SSEState) {
+ self.base = base
+ self.state = state
+ }
- let field = String(elements[0])
- var value = String(elements[1])
- if value.first == " " {
- value.removeFirst()
- }
+ public mutating func next() async throws -> ServerSentEvent? {
+ guard !finished else { return nil }
- 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
+ while let byte = try await base.next() {
+ let event = parser.consume(byte)
+ await mirrorStateIfChanged()
+ if let event {
+ return event
}
}
- // End of stream: an incomplete final block is discarded per spec,
- // including any id it carried.
+ // End of stream: the incomplete final block is discarded per spec,
+ // but a trailing retry field still takes effect.
+ finished = true
+ parser.finish()
+ await mirrorStateIfChanged()
return nil
}
- }
-}
-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()
+ /// Keeps the public ``SSEState`` actor in sync with the parser,
+ /// hopping to the actor only when a value actually changed.
+ private mutating func mirrorStateIfChanged() async {
+ if parser.retryInterval != mirroredRetryInterval, let interval = parser.retryInterval {
+ mirroredRetryInterval = interval
+ await state.updateRetryInterval(interval)
}
-
- 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)
+ if parser.lastEventId != mirroredLastEventId, let id = parser.lastEventId {
+ mirroredLastEventId = id
+ await state.updateLastEventId(id)
}
}
}
+}
- 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)
+extension AsyncServerSentEvents: Sendable where Base: Sendable {}
+
+public extension AsyncSequence where Element == UInt8 {
+ /// Parses these bytes as a server-sent event stream.
+ func sse() -> AsyncServerSentEvents {
+ AsyncServerSentEvents(bytes: self)
}
}
diff --git a/Sources/AsyncServerSentEvents/EventSource.swift b/Sources/AsyncServerSentEvents/EventSource.swift
index 36317cd..edea91b 100644
--- a/Sources/AsyncServerSentEvents/EventSource.swift
+++ b/Sources/AsyncServerSentEvents/EventSource.swift
@@ -75,7 +75,6 @@ public struct EventSource: AsyncSequence, Sendable {
var lastEventId: String?
var current: AsyncServerSentEvents.AsyncIterator?
- var currentState: SSEState?
var hasAttemptedConnection = false
var finished = false
@@ -112,13 +111,13 @@ public struct EventSource: AsyncSequence, Sendable {
return event
}
// The server closed the stream cleanly: reconnect.
- await prepareForReconnect()
+ prepareForReconnect()
} catch is CancellationError {
finished = true
throw CancellationError()
} catch {
// A network error mid-stream: reconnect.
- await prepareForReconnect()
+ prepareForReconnect()
}
}
}
@@ -133,24 +132,22 @@ public struct EventSource: AsyncSequence, Sendable {
bytes.task.cancel()
throw error
}
- let sse = bytes.sse()
- current = sse.makeAsyncIterator()
- currentState = sse.state
+ current = bytes.sse().makeAsyncIterator()
}
- private mutating func prepareForReconnect() async {
+ private mutating func prepareForReconnect() {
// 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 {
+ // event (retry-only or id-only blocks), read directly off the
+ // iterator's parser.
+ if let parser = current?.parser {
+ if let interval = parser.retryInterval {
retryInterval = interval
}
- if let id = await state.lastEventId {
+ if let id = parser.lastEventId {
lastEventId = id
}
}
current = nil
- currentState = nil
}
}
}
diff --git a/Sources/AsyncServerSentEvents/SSEByteStream.swift b/Sources/AsyncServerSentEvents/SSEByteStream.swift
index 7320879..35572a3 100644
--- a/Sources/AsyncServerSentEvents/SSEByteStream.swift
+++ b/Sources/AsyncServerSentEvents/SSEByteStream.swift
@@ -36,11 +36,6 @@ public struct SSEByteStream: AsyncSequence, @unchecked Sendable {
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
diff --git a/Sources/AsyncServerSentEvents/SSEParser.swift b/Sources/AsyncServerSentEvents/SSEParser.swift
new file mode 100644
index 0000000..c24cc87
--- /dev/null
+++ b/Sources/AsyncServerSentEvents/SSEParser.swift
@@ -0,0 +1,205 @@
+import Foundation
+
+/// An incremental, synchronous parser for the server-sent events stream format.
+///
+/// `SSEParser` is a value type: feed it bytes as they arrive and collect the
+/// events it emits. It is the core that ``AsyncServerSentEvents`` is built on,
+/// and can be used directly when bytes arrive through something other than an
+/// `AsyncSequence` — a delegate callback, a WebSocket frame, or a fully
+/// buffered response body:
+///
+/// ```swift
+/// var parser = SSEParser()
+/// for chunk in chunks {
+/// for event in parser.consume(chunk) {
+/// print(event.type, event.data)
+/// }
+/// }
+/// parser.finish()
+/// ```
+///
+/// Parsing never fails: per the
+/// [WHATWG specification](https://html.spec.whatwg.org/multipage/server-sent-events.html),
+/// invalid UTF-8 decodes with replacement characters and malformed field lines
+/// are ignored.
+public struct SSEParser: Sendable {
+
+ /// The last event ID committed by a completed block.
+ ///
+ /// This is the value to resume from (via the `Last-Event-ID` header) when
+ /// reconnecting, so ``finish()`` retains it. It is reconnect state only:
+ /// per the specification the ID buffer is per-stream, so events parsed
+ /// after ``finish()`` do not inherit this value into their
+ /// ``ServerSentEvent/lastEventId`` until the new stream sends an `id`.
+ public private(set) var lastEventId: String?
+
+ /// The reconnection time in milliseconds from the most recent valid
+ /// `retry` field. Retained by ``finish()``.
+ public private(set) var retryInterval: Int?
+
+ private var lines = SSELineSplitter()
+
+ /// Accumulates each data value plus a newline; the trailing newline is
+ /// removed at dispatch, so a lone `data:` line still dispatches an event
+ /// with empty data while a block with no data field dispatches nothing.
+ private var dataBuffer = ""
+ private var eventName: String?
+ private var eventId: String?
+
+ /// The last event ID buffer: set immediately by `id` fields, but only
+ /// committed to ``lastEventId`` when a blank line completes a block.
+ private var idBuffer: String?
+
+ public init() {}
+
+ /// Consumes one byte, returning an event if this byte completed a block.
+ public mutating func consume(_ byte: UInt8) -> ServerSentEvent? {
+ guard let line = lines.consume(byte) else { return nil }
+ return process(line: line)
+ }
+
+ /// Consumes a chunk of bytes, returning the events completed within it.
+ public mutating func consume(_ bytes: some Sequence) -> [ServerSentEvent] {
+ var events: [ServerSentEvent] = []
+ for byte in bytes {
+ if let event = consume(byte) {
+ events.append(event)
+ }
+ }
+ return events
+ }
+
+ /// Signals the end of the stream.
+ ///
+ /// An unterminated final line is still processed (so a trailing `retry`
+ /// field takes effect), the incomplete block is discarded per the
+ /// specification, and the parser resets for a new stream — stripping a
+ /// leading BOM again — while ``lastEventId`` and ``retryInterval`` are
+ /// retained for reconnection. The ID buffer is per-stream, so events on
+ /// the next stream carry a `nil` ``ServerSentEvent/lastEventId`` until
+ /// that stream sends an `id` field.
+ public mutating func finish() {
+ if let line = lines.finish() {
+ // An unterminated line is never blank, so this cannot dispatch.
+ _ = process(line: line)
+ }
+ lines = SSELineSplitter()
+ dataBuffer = ""
+ eventName = nil
+ eventId = nil
+ idBuffer = nil
+ }
+
+ /// Parses a complete stream, returning every dispatched event.
+ ///
+ /// A trailing block not terminated by a blank line is discarded, per the
+ /// specification's end-of-file rule.
+ public static func parse(_ bytes: some Sequence) -> [ServerSentEvent] {
+ var parser = SSEParser()
+ return parser.consume(bytes)
+ }
+
+ private mutating func process(line: [UInt8]) -> ServerSentEvent? {
+ guard !line.isEmpty else { return dispatchBlock() }
+
+ // Split at the first colon; a line without one is a field name with an
+ // empty value. Only one optional space after the colon is removed.
+ let colonIndex = line.firstIndex(of: UInt8(ascii: ":"))
+ let field = line[..<(colonIndex ?? line.endIndex)]
+ var value = line[(colonIndex.map { $0 + 1 } ?? line.endIndex)...]
+ if value.first == UInt8(ascii: " ") {
+ value = value.dropFirst()
+ }
+
+ if field.elementsEqual("data".utf8) {
+ dataBuffer.append(String(decoding: value, as: UTF8.self))
+ dataBuffer.append("\n")
+ } else if field.elementsEqual("event".utf8) {
+ eventName = String(decoding: value, as: UTF8.self)
+ } else if field.elementsEqual("id".utf8) {
+ if !value.contains(0) {
+ let id = String(decoding: value, as: UTF8.self)
+ eventId = id
+ idBuffer = id
+ }
+ } else if field.elementsEqual("retry".utf8) {
+ if !value.isEmpty,
+ value.allSatisfy({ (UInt8(ascii: "0")...UInt8(ascii: "9")).contains($0) }),
+ let milliseconds = Int(String(decoding: value, as: UTF8.self)) {
+ retryInterval = milliseconds
+ }
+ }
+ // Comment lines (empty field name) and unknown fields are ignored.
+ return nil
+ }
+
+ /// Handles a blank line: commits the last event ID — even when no event is
+ /// dispatched (e.g. an id-only block) — and dispatches the block's event
+ /// if it accumulated any data.
+ private mutating func dispatchBlock() -> ServerSentEvent? {
+ if let id = idBuffer {
+ lastEventId = id
+ }
+ defer {
+ dataBuffer = ""
+ eventName = nil
+ eventId = nil
+ }
+ guard !dataBuffer.isEmpty else { return nil }
+ var data = dataBuffer
+ data.removeLast() // the newline appended by the final data field
+ return ServerSentEvent(id: eventId, name: eventName, data: data, lastEventId: idBuffer)
+ }
+}
+
+/// Splits a byte stream into lines terminated by LF, CR, or CRLF, stripping a
+/// single leading BOM from the stream, 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 SSELineSplitter: Sendable {
+ private var buffer: [UInt8] = []
+ private var sawCarriageReturn = false
+ private var isFirstLine = true
+
+ /// Consumes one byte, returning a completed line if this byte ended one.
+ mutating func consume(_ byte: UInt8) -> [UInt8]? {
+ if sawCarriageReturn {
+ sawCarriageReturn = false
+ if byte == 0x0A { // LF completing a CRLF pair
+ return nil
+ }
+ }
+
+ switch byte {
+ case 0x0A:
+ return takeLine()
+ case 0x0D:
+ sawCarriageReturn = true
+ return takeLine()
+ default:
+ buffer.append(byte)
+ return nil
+ }
+ }
+
+ /// Flushes an unterminated final line at end of input, if any.
+ mutating func finish() -> [UInt8]? {
+ sawCarriageReturn = false
+ guard !buffer.isEmpty else { return nil }
+ return takeLine()
+ }
+
+ private mutating func takeLine() -> [UInt8] {
+ defer { buffer.removeAll(keepingCapacity: true) }
+ if isFirstLine {
+ isFirstLine = false
+ if buffer.starts(with: [0xEF, 0xBB, 0xBF]) {
+ buffer.removeFirst(3)
+ }
+ }
+ return buffer
+ }
+}
diff --git a/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift b/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift
index 164e257..c557180 100644
--- a/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift
+++ b/Sources/AsyncServerSentEvents/URLSession+AsyncServerSentEvents.swift
@@ -65,15 +65,6 @@ enum SSERequest {
}
}
-#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.
diff --git a/Tests/AsyncServerSentEventsTests/SSELineSplittingTests.swift b/Tests/AsyncServerSentEventsTests/SSELineSplittingTests.swift
index c54073b..8e04b56 100644
--- a/Tests/AsyncServerSentEventsTests/SSELineSplittingTests.swift
+++ b/Tests/AsyncServerSentEventsTests/SSELineSplittingTests.swift
@@ -10,68 +10,73 @@ import FoundationNetworking
@Suite("Line Splitting")
struct SSELineSplittingTests {
- private func splitLines(_ bytes: [UInt8]) async throws -> [String] {
- var iterator = SSELineIterator(base: bytes.byteStream.makeAsyncIterator())
+ private func splitLines(_ bytes: [UInt8]) -> [String] {
+ var splitter = SSELineSplitter()
var lines: [String] = []
- while let line = try await iterator.next() {
- lines.append(line)
+ for byte in bytes {
+ if let line = splitter.consume(byte) {
+ lines.append(String(decoding: line, as: UTF8.self))
+ }
+ }
+ if let line = splitter.finish() {
+ lines.append(String(decoding: line, as: UTF8.self))
}
return lines
}
- private func splitLines(_ text: String) async throws -> [String] {
- try await splitLines(Array(text.utf8))
+ private func splitLines(_ text: String) -> [String] {
+ 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")
+ func mixedTerminators() {
+ let lines = 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 {
+ func emptyLines() {
// a CR | CR LF (empty line) | LF (empty line) | b (unterminated)
- let lines = try await splitLines("a\r\r\n\nb")
+ let lines = splitLines("a\r\r\n\nb")
#expect(lines == ["a", "", "", "b"])
- let crlfOnly = try await splitLines("\r\n\r\n")
+ let crlfOnly = 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")
+ func crlfIsSingleBoundary() {
+ let lines = 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")
+ func consecutiveCRs() {
+ let lines = 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"])
+ func trailingTerminators() {
+ #expect(splitLines("a\r") == ["a"])
+ #expect(splitLines("a\n") == ["a"])
+ #expect(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"])
+ func edgeInputs() {
+ #expect(splitLines("") == [])
+ #expect(splitLines("\n") == [""])
+ #expect(splitLines("\r") == [""])
+ #expect(splitLines("\r\n") == [""])
+ #expect(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")
+ let lines = splitLines(text + "\n")
#expect(lines == [text])
// And they must survive through the full parser as data content.
diff --git a/Tests/AsyncServerSentEventsTests/SSEParserTests.swift b/Tests/AsyncServerSentEventsTests/SSEParserTests.swift
new file mode 100644
index 0000000..79fb892
--- /dev/null
+++ b/Tests/AsyncServerSentEventsTests/SSEParserTests.swift
@@ -0,0 +1,148 @@
+import Testing
+import Foundation
+@testable import AsyncServerSentEvents
+
+@Suite("SSEParser")
+struct SSEParserTests {
+
+ /// Every fixture must produce identical events and final state whether it
+ /// is parsed synchronously in one buffer or through the async sequence.
+ @Test("Synchronous parsing should match the async sequence", arguments: SSETestData.allValidTests + SSETestData.allResilienceTests)
+ func matchesAsyncParsing(fixture: String) async throws {
+ // Normalize the fixture the same way the async test helper does.
+ var input = fixture.unindented
+ if input.hasSuffix("\n") && !input.hasSuffix("\n\n") {
+ input.append("\n")
+ }
+ let bytes = Array(input.utf8)
+
+ let sse = AsyncServerSentEvents(bytes: bytes.byteStream)
+ let asyncEvents = try await sse.collect()
+
+ var parser = SSEParser()
+ let syncEvents = parser.consume(bytes)
+ parser.finish()
+ let asyncLastEventId = await sse.state.lastEventId
+ let asyncRetryInterval = await sse.state.retryInterval
+
+ #expect(syncEvents == asyncEvents)
+ #expect(SSEParser.parse(bytes) == asyncEvents)
+ #expect(parser.lastEventId == asyncLastEventId)
+ #expect(parser.retryInterval == asyncRetryInterval)
+ }
+
+ @Test("Chunk boundaries should not affect parsing", arguments: [1, 2, 3, 7, 16, 1024])
+ func chunkBoundaryInvariance(chunkSize: Int) {
+ let input = "\u{FEFF}id: 1\nevent: tick\ndata: first\ndata: sec:ond\n\nretry: 250\n\nid: 2\r\ndata: é🎉\r\n\r\ndata: dangling"
+ let bytes = Array(input.utf8)
+ let expected = SSEParser.parse(bytes)
+
+ var parser = SSEParser()
+ var events: [ServerSentEvent] = []
+ var start = 0
+ while start < bytes.count {
+ let end = min(start + chunkSize, bytes.count)
+ events += parser.consume(bytes[start..