Sendable Take 2 (#136)

* Add dev containers to gitignore * Revert "Revert "Add Sendable conformances to WebsocketKit (#131)" (#135)" This reverts commit 0b43ce5. * Add @preconcurrency annotations to work around unsafe users * More preconcurrency annotations * Add API breakage allowlist * Add missing preconcurrency annotations * Last missing annotations * Update allowlist --------- Co-authored-by: Gwynne Raskind <gwynne@vapor.codes>
vapor · May 25, 2023 · a8f3ec8 · a8f3ec8
1 parent 0b43ce5
commit a8f3ec8
Show file tree

Hide file tree

Showing 13 changed files with 233 additions and 175 deletions.
diff --git a/.api-breakage/allowlist-branch-sendable-take-2.txt b/.api-breakage/allowlist-branch-sendable-take-2.txt
@@ -0,0 +1,14 @@
+API breakage: func WebSocket.onText(_:) is now with @preconcurrency
+API breakage: func WebSocket.onBinary(_:) is now with @preconcurrency
+API breakage: func WebSocket.onPong(_:) is now with @preconcurrency
+API breakage: func WebSocket.onPing(_:) is now with @preconcurrency
+API breakage: func WebSocket.connect(to:headers:configuration:on:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocket.connect(scheme:host:port:path:query:headers:configuration:on:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocket.connect(scheme:host:port:path:query:headers:proxy:proxyPort:proxyHeaders:proxyConnectDeadline:configuration:on:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocket.connect(to:headers:proxy:proxyPort:proxyHeaders:proxyConnectDeadline:configuration:on:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocket.client(on:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocket.client(on:config:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocket.server(on:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocket.server(on:config:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocketClient.connect(scheme:host:port:path:query:headers:onUpgrade:) is now with @preconcurrency
+API breakage: func WebSocketClient.connect(scheme:host:port:path:query:headers:proxy:proxyPort:proxyHeaders:proxyConnectDeadline:onUpgrade:) is now with @preconcurrency
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
@@ -5,9 +5,7 @@ concurrency:
 on:
   pull_request: { types: [opened, reopened, synchronize, ready_for_review] }
   push: { branches: [ main ] }
-
 jobs:
-
   vapor-integration:
     if: ${{ !(github.event.pull_request.draft || false) }}
     runs-on: ubuntu-latest

diff --git a/.gitignore b/.gitignore
@@ -5,3 +5,4 @@
 DerivedData
 .swiftpm
 Package.resolved
+.devcontainer/
diff --git a/Package.swift b/Package.swift
@@ -16,8 +16,8 @@ let package = Package(
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.53.0"),
         .package(url: "https://github.com/apple/swift-nio-extras.git", from: "1.16.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.24.0"),
-        .package(url: "https://github.com/apple/swift-nio-transport-services.git", from: "1.11.4"),
-        .package(url: "https://github.com/apple/swift-atomics.git", from: "1.0.2"),
+        .package(url: "https://github.com/apple/swift-nio-transport-services.git", from: "1.16.0"),
+        .package(url: "https://github.com/apple/swift-atomics.git", from: "1.1.0"),
     ],
     targets: [
         .target(name: "WebSocketKit", dependencies: [

diff --git a/Sources/WebSocketKit/Concurrency/WebSocket+Concurrency.swift b/Sources/WebSocketKit/Concurrency/WebSocket+Concurrency.swift
@@ -40,44 +40,52 @@ extension WebSocket {
         try await close(code: code).get()
     }
 
-    public func onText(_ callback: @escaping (WebSocket, String) async -> ()) {
-        onText { socket, text in
-            Task {
-                await callback(socket, text)
+    @preconcurrency public func onText(_ callback: @Sendable @escaping (WebSocket, String) async -> ()) {
+        self.eventLoop.execute {
+            self.onText { socket, text in
+                Task {
+                    await callback(socket, text)
+                }
             }
         }
     }
 
-    public func onBinary(_ callback: @escaping (WebSocket, ByteBuffer) async -> ()) {
-        onBinary { socket, binary in
-            Task {
-                await callback(socket, binary)
+    @preconcurrency public func onBinary(_ callback: @Sendable @escaping (WebSocket, ByteBuffer) async -> ()) {
+        self.eventLoop.execute {
+            self.onBinary { socket, binary in
+                Task {
+                    await callback(socket, binary)
+                }
             }
         }
     }
 
-    public func onPong(_ callback: @escaping (WebSocket) async -> ()) {
-        onPong { socket in
-            Task {
-                await callback(socket)
+    @preconcurrency public func onPong(_ callback: @Sendable @escaping (WebSocket) async -> ()) {
+        self.eventLoop.execute {
+            self.onPong { socket in
+                Task {
+                    await callback(socket)
+                }
             }
         }
     }
 
-    public func onPing(_ callback: @escaping (WebSocket) async -> ()) {
-        onPing { socket in
-            Task {
-                await callback(socket)
+    @preconcurrency public func onPing(_ callback: @Sendable @escaping (WebSocket) async -> ()) {
+        self.eventLoop.execute {
+            self.onPing { socket in
+                Task {
+                    await callback(socket)
+                }
             }
         }
     }
 
-    public static func connect(
+    @preconcurrency public static func connect(
         to url: String,
         headers: HTTPHeaders = [:],
         configuration: WebSocketClient.Configuration = .init(),
         on eventLoopGroup: EventLoopGroup,
-        onUpgrade: @escaping (WebSocket) async -> ()
+        onUpgrade: @Sendable @escaping (WebSocket) async -> ()
     ) async throws {
         return try await self.connect(
             to: url,
@@ -92,12 +100,12 @@ extension WebSocket {
         ).get()
     }
 
-    public static func connect(
+    @preconcurrency public static func connect(
         to url: URL,
         headers: HTTPHeaders = [:],
         configuration: WebSocketClient.Configuration = .init(),
         on eventLoopGroup: EventLoopGroup,
-        onUpgrade: @escaping (WebSocket) async -> ()
+        onUpgrade: @Sendable @escaping (WebSocket) async -> ()
     ) async throws {
         return try await self.connect(
             to: url,
@@ -112,7 +120,7 @@ extension WebSocket {
         ).get()
     }
 
-    public static func connect(
+    @preconcurrency public static func connect(
         scheme: String = "ws",
         host: String,
         port: Int = 80,
@@ -121,7 +129,7 @@ extension WebSocket {
         headers: HTTPHeaders = [:],
         configuration: WebSocketClient.Configuration = .init(),
         on eventLoopGroup: EventLoopGroup,
-        onUpgrade: @escaping (WebSocket) async -> ()
+        onUpgrade: @Sendable @escaping (WebSocket) async -> ()
     ) async throws {
         return try await self.connect(
             scheme: scheme,

diff --git a/Sources/WebSocketKit/Exports.swift b/Sources/WebSocketKit/Exports.swift
@@ -6,9 +6,7 @@
 @_documentation(visibility: internal) @_exported import protocol NIOCore.EventLoopGroup
 @_documentation(visibility: internal) @_exported import struct NIOCore.EventLoopPromise
 @_documentation(visibility: internal) @_exported import class NIOCore.EventLoopFuture
-
 @_documentation(visibility: internal) @_exported import struct NIOHTTP1.HTTPHeaders
-
 @_documentation(visibility: internal) @_exported import struct Foundation.URL
 
 #else
@@ -19,9 +17,7 @@
 @_exported import protocol NIOCore.EventLoopGroup
 @_exported import struct NIOCore.EventLoopPromise
 @_exported import class NIOCore.EventLoopFuture
-
 @_exported import struct NIOHTTP1.HTTPHeaders
-
 @_exported import struct Foundation.URL
 
 #endif
diff --git a/Sources/WebSocketKit/WebSocket+Connect.swift b/Sources/WebSocketKit/WebSocket+Connect.swift
@@ -12,12 +12,13 @@ extension WebSocket {
     ///   - eventLoopGroup: Event loop group to be used by the WebSocket client.
     ///   - onUpgrade: An escaping closure to be executed after the upgrade is completed by `NIOWebSocketClientUpgrader`.
     /// - Returns: An future which completes when the connection to the WebSocket server is established.
+    @preconcurrency
     public static func connect(
         to url: String,
         headers: HTTPHeaders = [:],
         configuration: WebSocketClient.Configuration = .init(),
         on eventLoopGroup: EventLoopGroup,
-        onUpgrade: @escaping (WebSocket) -> ()
+        onUpgrade: @Sendable @escaping (WebSocket) -> ()
     ) -> EventLoopFuture<Void> {
         guard let url = URL(string: url) else {
             return eventLoopGroup.any().makeFailedFuture(WebSocketClient.Error.invalidURL)
@@ -40,12 +41,13 @@ extension WebSocket {
     ///   - eventLoopGroup: Event loop group to be used by the WebSocket client.
     ///   - onUpgrade: An escaping closure to be executed after the upgrade is completed by `NIOWebSocketClientUpgrader`.
     /// - Returns: An future which completes when the connection to the WebSocket server is established.
+    @preconcurrency
     public static func connect(
         to url: URL,
         headers: HTTPHeaders = [:],
         configuration: WebSocketClient.Configuration = .init(),
         on eventLoopGroup: EventLoopGroup,
-        onUpgrade: @escaping (WebSocket) -> ()
+        onUpgrade: @Sendable @escaping (WebSocket) -> ()
     ) -> EventLoopFuture<Void> {
         let scheme = url.scheme ?? "ws"
         return self.connect(
@@ -74,6 +76,7 @@ extension WebSocket {
     ///   - eventLoopGroup: Event loop group to be used by the WebSocket client.
     ///   - onUpgrade: An escaping closure to be executed after the upgrade is completed by `NIOWebSocketClientUpgrader`.
     /// - Returns: An future which completes when the connection to the WebSocket server is established.
+    @preconcurrency
     public static func connect(
         scheme: String = "ws",
         host: String,
@@ -83,7 +86,7 @@ extension WebSocket {
         headers: HTTPHeaders = [:],
         configuration: WebSocketClient.Configuration = .init(),
         on eventLoopGroup: EventLoopGroup,
-        onUpgrade: @escaping (WebSocket) -> ()
+        onUpgrade: @Sendable @escaping (WebSocket) -> ()
     ) -> EventLoopFuture<Void> {
         return WebSocketClient(
             eventLoopGroupProvider: .shared(eventLoopGroup),
@@ -116,6 +119,7 @@ extension WebSocket {
     ///   - eventLoopGroup: Event loop group to be used by the WebSocket client.
     ///   - onUpgrade: An escaping closure to be executed after the upgrade is completed by `NIOWebSocketClientUpgrader`.
     /// - Returns: An future which completes when the connection to the origin server is established.
+    @preconcurrency
     public static func connect(
         scheme: String = "ws",
         host: String,
@@ -129,7 +133,7 @@ extension WebSocket {
         proxyConnectDeadline: NIODeadline = NIODeadline.distantFuture,
         configuration: WebSocketClient.Configuration = .init(),
         on eventLoopGroup: EventLoopGroup,
-        onUpgrade: @escaping (WebSocket) -> ()
+        onUpgrade: @Sendable @escaping (WebSocket) -> ()
     ) -> EventLoopFuture<Void> {
         return WebSocketClient(
             eventLoopGroupProvider: .shared(eventLoopGroup),
@@ -162,6 +166,7 @@ extension WebSocket {
     ///   - eventLoopGroup: Event loop group to be used by the WebSocket client.
     ///   - onUpgrade: An escaping closure to be executed after the upgrade is completed by `NIOWebSocketClientUpgrader`.
     /// - Returns: An future which completes when the connection to the origin server is established.
+    @preconcurrency
     public static func connect(
         to url: String,
         headers: HTTPHeaders = [:],
@@ -171,7 +176,7 @@ extension WebSocket {
         proxyConnectDeadline: NIODeadline = NIODeadline.distantFuture,
         configuration: WebSocketClient.Configuration = .init(),
         on eventLoopGroup: EventLoopGroup,
-        onUpgrade: @escaping (WebSocket) -> ()
+        onUpgrade: @Sendable @escaping (WebSocket) -> ()
     ) -> EventLoopFuture<Void> {
         guard let url = URL(string: url) else {
             return eventLoopGroup.any().makeFailedFuture(WebSocketClient.Error.invalidURL)