Merge pull request #31 from Recouse/concurrency-improvements

Concurrency improvements

Author: Recouse

README.md

@@ -16,7 +16,6 @@ It also leverages Swift concurrency features to provide a more expressive and in
 - [x] Simple Swift API for SSE
 - [x] Supports data-only mode
 - [x] Data race safety with Swift 6
-- [ ] Broadcast event stream to multiple consumers (WIP)
 
 ## Installation
 
@@ -57,9 +56,9 @@ Using EventSource is easy. Simply create a new data task from an instance of Eve
 import EventSource
 
 let eventSource = EventSource()
-let dataTask = await eventSource.dataTask(for: urlRequest)
+let dataTask = eventSource.dataTask(for: urlRequest)
 
-for await event in await dataTask.events() {
+for await event in dataTask.events() {
     switch event {
     case .open:
         print("Connection was opened.")
@@ -96,11 +95,11 @@ urlRequest.httpBody = """
 """.data(using: .utf8)!
 
 let eventSource = EventSource(mode: .dataOnly)
-let dataTask = await eventSource.dataTask(for: urlRequest)
+let dataTask = eventSource.dataTask(for: urlRequest)
 
 var response: String = ""
 
-for await event in await dataTask.events() {
+for await event in dataTask.events() {
     switch event {
     case .event(let event):
         if let data = eventDevent.data?.data(using: .utf8) {
@@ -132,6 +131,10 @@ No dependencies.
 
 Contributions to are always welcomed! For more details see [CONTRIBUTING.md](CONTRIBUTING.md).
 
+## Credits
+
+* Mutex backport from [swift-sharing](https://github.com/pointfreeco/swift-sharing)
+
 ## License
 
 EventSource is released under the MIT License. See [LICENSE](LICENSE) for more information.

Sources/EventSource/EventSource.swift

@@ -11,11 +11,6 @@ import Foundation
     import FoundationNetworking
 #endif
 
-/// The global actor used for isolating ``EventSource/EventSource/DataTask``.
-@globalActor public actor EventSourceActor: GlobalActor {
-    public static let shared = EventSourceActor()
-}
-
 ///
 /// An `EventSource` instance opens a persistent connection to an HTTP server,
 /// which sends events in `text/event-stream` format.
@@ -63,7 +58,6 @@ public struct EventSource: Sendable {
         self.timeoutInterval = timeoutInterval
     }
 
-    @EventSourceActor
     public func dataTask(for urlRequest: URLRequest) -> DataTask {
         DataTask(
             urlRequest: urlRequest,
@@ -79,29 +73,70 @@ public extension EventSource {
     /// Creation of a task is exclusively handled by ``EventSource``. A new task can be created by calling
     /// ``EventSource/EventSource/dataTask(for:)`` method on the EventSource instance. After creating a task,
     /// it can be started by iterating event stream returned by ``DataTask/events()``.
-    @EventSourceActor final class DataTask {
+    final class DataTask: Sendable {
+        private let _readyState: Mutex<ReadyState> = Mutex(.none)
+
         /// A value representing the state of the connection.
-        public private(set) var readyState: ReadyState = .none
+        public var readyState: ReadyState {
+            get {
+                _readyState.withLock { $0 }
+            }
+            set {
+                _readyState.withLock { $0 = newValue }
+            }
+        }
+
+        private let _lastMessageId: Mutex<String> = Mutex("")
 
         /// Last event's ID string value.
         ///
         /// Sent in a HTTP request header and used when a user is to reestablish the connection.
-        public private(set) var lastMessageId: String = ""
+        public var lastMessageId: String {
+            get {
+                _lastMessageId.withLock { $0 }
+            }
+            set {
+                _lastMessageId.withLock { $0 = newValue }
+            }
+        }
 
         /// A URLRequest of the events source.
         public let urlRequest: URLRequest
 
-        private var eventParser: EventParser
+        private let _eventParser: Mutex<EventParser>
+
+        private var eventParser: EventParser {
+            get {
+                _eventParser.withLock { $0 }
+            }
+            set {
+                _eventParser.withLock { $0 = newValue }
+            }
+        }
 
         private let timeoutInterval: TimeInterval
 
-        private var continuation: AsyncStream<EventType>.Continuation?
+        private let _httpResponseErrorStatusCode: Mutex<Int?> = Mutex(nil)
 
-        private var urlSession: URLSession?
+        private var httpResponseErrorStatusCode: Int? {
+            get {
+                _httpResponseErrorStatusCode.withLock { $0 }
+            }
+            set {
+                _httpResponseErrorStatusCode.withLock { $0 = newValue }
+            }
+        }
 
-        private var urlSessionDataTask: URLSessionDataTask?
+        private let _consumed: Mutex<Bool> = Mutex(false)
 
-        private var httpResponseErrorStatusCode: Int?
+        private var consumed: Bool {
+            get {
+                _consumed.withLock { $0 }
+            }
+            set {
+                _consumed.withLock { $0 = newValue }
+            }
+        }
 
         private var urlSessionConfiguration: URLSessionConfiguration {
             let configuration = URLSessionConfiguration.default
@@ -121,13 +156,13 @@ public extension EventSource {
             timeoutInterval: TimeInterval
         ) {
             self.urlRequest = urlRequest
-            self.eventParser = eventParser
+            self._eventParser = Mutex(eventParser)
             self.timeoutInterval = timeoutInterval
         }
 
         /// Creates and returns event stream.
         public func events() -> AsyncStream<EventType> {
-            if urlSessionDataTask != nil {
+            if consumed {
                 return AsyncStream { continuation in
                     continuation.yield(.error(EventSourceError.alreadyConsumed))
                     continuation.finish()
@@ -136,67 +171,76 @@ public extension EventSource {
 
             return AsyncStream { continuation in
                 let sessionDelegate = SessionDelegate()
+                let urlSession = URLSession(
+                    configuration: urlSessionConfiguration,
+                    delegate: sessionDelegate,
+                    delegateQueue: nil
+                )
+                let urlSessionDataTask = urlSession.dataTask(with: urlRequest)
+
                 let sessionDelegateTask = Task { [weak self] in
                     for await event in sessionDelegate.eventStream {
                         guard let self else { return }
 
                         switch event {
                         case let .didCompleteWithError(error):
-                            handleSessionError(error)
+                            handleSessionError(error, stream: continuation, urlSession: urlSession)
                         case let .didReceiveResponse(response, completionHandler):
-                            handleSessionResponse(response, completionHandler: completionHandler)
+                            handleSessionResponse(
+                                response,
+                                stream: continuation,
+                                urlSession: urlSession,
+                                completionHandler: completionHandler
+                            )
                         case let .didReceiveData(data):
-                            parseMessages(from: data)
+                            parseMessages(from: data, stream: continuation, urlSession: urlSession)
                         }
                     }
                 }
 
                 #if compiler(>=6.0)
                 continuation.onTermination = { @Sendable [weak self] _ in
                     sessionDelegateTask.cancel()
-                    Task { await self?.close() }
+                    Task { self?.close(stream: continuation, urlSession: urlSession) }
                 }
                 #else
                 continuation.onTermination = { @Sendable _ in
                     sessionDelegateTask.cancel()
                     Task { [weak self] in
-                        await self?.close()
+                        await self?.close(stream: continuation, urlSession: urlSession)
                     }
                 }
                 #endif
 
-                self.continuation = continuation
-
-
-                urlSession = URLSession(
-                    configuration: urlSessionConfiguration,
-                    delegate: sessionDelegate,
-                    delegateQueue: nil
-                )
-
-                urlSessionDataTask = urlSession!.dataTask(with: urlRequest)
-                urlSessionDataTask!.resume()
+                urlSessionDataTask.resume()
                 readyState = .connecting
+                consumed = true
             }
         }
 
-        private func handleSessionError(_ error: Error?) {
+        private func handleSessionError(
+            _ error: Error?,
+            stream continuation: AsyncStream<EventType>.Continuation,
+            urlSession: URLSession
+        ) {
             guard readyState != .closed else {
-                close()
+                close(stream: continuation, urlSession: urlSession)
                 return
             }
 
             // Send error event
             if let error {
-                sendErrorEvent(with: error)
+                sendErrorEvent(with: error, stream: continuation)
             }
 
             // Close connection
-            close()
+            close(stream: continuation, urlSession: urlSession)
         }
 
         private func handleSessionResponse(
             _ response: URLResponse,
+            stream continuation: AsyncStream<EventType>.Continuation,
+            urlSession: URLSession,
             completionHandler: @escaping (URLSession.ResponseDisposition) -> Void
         ) {
             guard readyState != .closed else {
@@ -212,13 +256,13 @@ public extension EventSource {
             // Stop connection when 204 response code, otherwise keep open
             guard httpResponse.statusCode != 204 else {
                 completionHandler(.cancel)
-                close()
+                close(stream: continuation, urlSession: urlSession)
                 return
             }
 
             if 200...299 ~= httpResponse.statusCode {
                 if readyState != .open {
-                    setOpen()
+                    setOpen(stream: continuation)
                 }
             } else {
                 httpResponseErrorStatusCode = httpResponse.statusCode
@@ -230,20 +274,26 @@ public extension EventSource {
         /// Closes the connection, if one was made,
         /// and sets the `readyState` property to `.closed`.
         /// - Returns: State before closing.
-        private func close() {
+        private func close(stream continuation: AsyncStream<EventType>.Continuation, urlSession: URLSession) {
             let previousState = self.readyState
             if previousState != .closed {
-                continuation?.yield(.closed)
-                continuation?.finish()
+                continuation.yield(.closed)
+                continuation.finish()
             }
-            cancel()
+            cancel(urlSession: urlSession)
         }
 
-        private func parseMessages(from data: Data) {
+        private func parseMessages(
+            from data: Data,
+            stream continuation: AsyncStream<EventType>.Continuation,
+            urlSession: URLSession
+        ) {
             if let httpResponseErrorStatusCode {
                 self.httpResponseErrorStatusCode = nil
                 handleSessionError(
-                    EventSourceError.connectionError(statusCode: httpResponseErrorStatusCode, response: data)
+                    EventSourceError.connectionError(statusCode: httpResponseErrorStatusCode, response: data),
+                    stream: continuation,
+                    urlSession: urlSession
                 )
                 return
             }
@@ -256,17 +306,17 @@ public extension EventSource {
             }
 
             events.forEach {
-                continuation?.yield(.event($0))
+                continuation.yield(.event($0))
             }
         }
 
-        private func setOpen() {
+        private func setOpen(stream continuation: AsyncStream<EventType>.Continuation) {
             readyState = .open
-            continuation?.yield(.open)
+            continuation.yield(.open)
         }
 
-        private func sendErrorEvent(with error: Error) {
-            continuation?.yield(.error(error))
+        private func sendErrorEvent(with error: Error, stream continuation: AsyncStream<EventType>.Continuation) {
+            continuation.yield(.error(error))
         }
 
         /// Cancels the task.
@@ -275,11 +325,10 @@ public extension EventSource {
         /// The event stream supports cooperative task cancellation. However, it should be noted that
         /// canceling the parent Task only cancels the underlying `URLSessionDataTask` of
         /// ``EventSource/EventSource/DataTask``; this does not actually stop the ongoing request.
-        public func cancel() {
+        public func cancel(urlSession: URLSession) {
             readyState = .closed
             lastMessageId = ""
-            urlSessionDataTask?.cancel()
-            urlSession?.invalidateAndCancel()
+            urlSession.invalidateAndCancel()
         }
     }
 }