Fixes for strict-concurrency (Transferring workarounds)

Author: swhitty

FlyingFox/Tests/HTTPServerTests.swift

@@ -532,7 +532,7 @@ extension HTTPServer {
 
 
 #if canImport(Darwin)
-extension URLSessionWebSocketTask.Message: Equatable {
+extension URLSessionWebSocketTask.Message {
     public static func == (lhs: URLSessionWebSocketTask.Message, rhs: URLSessionWebSocketTask.Message) -> Bool {
         switch (lhs, rhs) {
         case (.string(let lval), .string(let rval)):
@@ -544,6 +544,11 @@ extension URLSessionWebSocketTask.Message: Equatable {
         }
     }
 }
+#if compiler(>=6)
+extension URLSessionWebSocketTask.Message: @retroactive Equatable { }
+#else
+extension URLSessionWebSocketTask.Message: Equatable { }
+#endif
 #endif
 
 extension Task where Success == Never, Failure == Never {

FlyingFox/Tests/Handlers/WebSocketHTTPHandlerTests.swift

@@ -205,9 +205,23 @@ private extension WebSocketHTTPHandler {
 
 private struct MockHandler: WSHandler {
     func makeFrames(for client: AsyncThrowingStream<WSFrame, any Error>) async throws -> AsyncStream<WSFrame> {
-        var iterator = client.makeAsyncIterator()
-        return AsyncStream<WSFrame> {
-            try? await iterator.next()
-        }
+        UnsafeFrames(source: client).makeStream()
+    }
+}
+
+private final class UnsafeFrames: @unchecked Sendable {
+
+    private var iterator: AsyncThrowingStream<WSFrame, any Error>.Iterator
+
+    init(source: AsyncThrowingStream<WSFrame, any Error>) {
+        self.iterator = source.makeAsyncIterator()
+    }
+
+    func makeStream() -> AsyncStream<WSFrame> {
+        AsyncStream<WSFrame> { await self.nextFrame() }
+    }
+
+    func nextFrame() async -> WSFrame? {
+        try? await iterator.next()
     }
 }

FlyingFox/Tests/WebSocket/WSFrameValidatorTests.swift

@@ -96,7 +96,24 @@ final class WSFrameValidatorTests: XCTestCase {
 private extension WSFrameValidator {
 
     static func validate(_ frames: [WSFrame]) async throws -> AsyncThrowingStream<WSFrame, any Swift.Error> {
-        var iterator = validateFrames(from: AsyncThrowingStream.make(frames)).makeAsyncIterator()
-        return AsyncThrowingStream { try await iterator.next() }
+        UnsafeFrames(frames: frames).makeStream()
     }
 }
+
+private final class UnsafeFrames: @unchecked Sendable {
+
+    var iterator: AsyncThrowingCompactMapSequence<AsyncThrowingStream<WSFrame, any Error>, WSFrame>.Iterator
+
+    init(frames: [WSFrame]) {
+        self.iterator = WSFrameValidator.validateFrames(from: AsyncThrowingStream.make(frames)).makeAsyncIterator()
+    }
+
+    func makeStream() -> AsyncThrowingStream<WSFrame, any Error> {
+        AsyncThrowingStream { try await self.nextFrame() }
+    }
+
+    func nextFrame() async throws -> WSFrame? {
+        try await iterator.next()
+    }
+}
+

FlyingSocks/Sources/AllocatedLock.swift

@@ -53,7 +53,7 @@ package struct AllocatedLock<State>: @unchecked Sendable {
     }
 
     @inlinable
-    package func withLock<R>(_ body: @Sendable (inout State) throws -> R) rethrows -> R where R: Sendable {
+    package func withLock<R>(_ body: @Sendable (inout State) throws -> R) rethrows -> R {
         storage.lock()
         defer { storage.unlock() }
         return try body(&storage.state)

FlyingSocks/Sources/AsyncBufferedCollection.swift

@@ -31,7 +31,7 @@
 
 import Foundation
 
-package struct AsyncBufferedCollection<C: Collection>: AsyncBufferedSequence {
+package struct AsyncBufferedCollection<C: Collection>: AsyncBufferedSequence where C: Sendable, C.Element: Sendable {
     package typealias Element = C.Element
 
     private let collection: C

FlyingSocks/Sources/AsyncBufferedEmptySequence.swift

@@ -29,7 +29,7 @@
 //  SOFTWARE.
 //
 
-package struct AsyncBufferedEmptySequence<Element>: Sendable, AsyncBufferedSequence {
+package struct AsyncBufferedEmptySequence<Element: Sendable>: Sendable, AsyncBufferedSequence {
 
     private let completeImmediately: Bool
 

FlyingSocks/Sources/AsyncBufferedSequence.swift

@@ -30,7 +30,7 @@
 //
 
 /// AsyncSequence that is buffered and can optionally receive contiguous elements in chunks, instead of just one-at-a-time.
-public protocol AsyncBufferedSequence<Element>: AsyncSequence where AsyncIterator: AsyncBufferedIteratorProtocol {
+public protocol AsyncBufferedSequence<Element>: AsyncSequence, Sendable where AsyncIterator: AsyncBufferedIteratorProtocol, Element: Sendable {
 
 }
 

FlyingSocks/Sources/AsyncSharedReplaySequence.swift

@@ -189,10 +189,9 @@ extension AsyncSharedReplaySequence {
 
         private func requestNextChunk(atMost count: Int) async throws -> (some Collection<Element>)? {
             do {
-                var iterator = try state.makeAsyncIterator()
-
-                let chunk = try await iterator.nextBuffer(suggested: count)
-                state = .iterating(iterator)
+                var iterator = try Transferring(state.makeAsyncIterator())
+                let chunk = try await iterator.nextBuffer(suggested: count)?.value
+                state = .iterating(iterator.value)
                 return chunk
             } catch {
                 state = .error(error)
@@ -202,6 +201,14 @@ extension AsyncSharedReplaySequence {
     }
 }
 
+private extension Transferring where Value: AsyncBufferedIteratorProtocol {
+
+    mutating func nextBuffer(suggested count: Int) async throws -> Transferring<Value.Buffer>? {
+        guard let buffer = try await value.nextBuffer(suggested: count) else { return nil }
+        return Transferring<Value.Buffer>(buffer)
+    }
+}
+
 extension AsyncSharedReplaySequence.SharedBuffer.State {
 
     mutating func makeAsyncIterator() throws -> Base.AsyncIterator {

FlyingSocks/Sources/ConsumingAsyncSequence.swift

@@ -92,8 +92,7 @@ extension ConsumingAsyncSequence {
             guard let element = try await iterator.next() else {
                 return nil
             }
-            let newState = State.iterating(iterator, index: index + 1)
-            state.withLock { $0 = newState }
+            setState(.iterating(iterator, index: index + 1))
             return element
         }
 
@@ -102,10 +101,14 @@ extension ConsumingAsyncSequence {
             guard let buffer = try await iterator.nextBuffer(suggested: count) else {
                 return nil
             }
-            let newState = State.iterating(iterator, index: index + buffer.count)
-            state.withLock { $0 = newState }
+            setState(.iterating(iterator, index: index + buffer.count))
             return buffer
         }
+
+        func setState(_ state: State) {
+            let t = Transferring(state)
+            self.state.withLock { $0 = t.value }
+        }
     }
 }
 

FlyingSocks/Sources/Transferring.swift

@@ -0,0 +1,38 @@
+//
+//  Transferring.swift
+//  FlyingFox
+//
+//  Created by Simon Whitty on 09/08/2024.
+//  Copyright © 2024 Simon Whitty. All rights reserved.
+//
+//  Distributed under the permissive MIT license
+//  Get the latest version from here:
+//
+//  https://github.com/swhitty/FlyingFox
+//
+//  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.
+//
+
+package struct Transferring<Value>: @unchecked Sendable {
+    package var value: Value
+
+    package init(_ value: Value) {
+        self.value = value
+    }
+}