Concurrency

Package.swift

@@ -1,19 +1,26 @@
-// swift-tools-version:5.0
+// swift-tools-version:5.10
 // The swift-tools-version declares the minimum version of Swift required to build this package.
 
 import PackageDescription
 
 let package = Package(
     name: "DataLoader",
+    platforms: [.macOS(.v12), .iOS(.v15), .tvOS(.v15), .watchOS(.v8)],
     products: [
         .library(name: "DataLoader", targets: ["DataLoader"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/apple/swift-nio.git", from: "2.0.0"),
+        .package(url: "https://github.com/apple/swift-algorithms.git", from: "1.0.0"),
+        .package(url: "https://github.com/adam-fowler/async-collections", from: "0.0.1"),
     ],
     targets: [
-        .target(name: "DataLoader", dependencies: ["NIO", "NIOConcurrencyHelpers"]),
+        .target(
+            name: "DataLoader",
+            dependencies: [
+                .product(name: "Algorithms", package: "swift-algorithms"),
+                .product(name: "AsyncCollections", package: "async-collections"),
+            ]
+        ),
         .testTarget(name: "DataLoaderTests", dependencies: ["DataLoader"]),
-    ],
-    swiftLanguageVersions: [.v5]
+    ]
 )

Sources/DataLoader/Channel/Channel.swift

@@ -0,0 +1,55 @@
+public actor Channel<Success: Sendable, Failure: Error>: Sendable {
+    private var state = State<Success, Failure>()
+}
+
+public extension Channel {
+    @discardableResult
+    func fulfill(_ value: Success) async -> Bool {
+        if await state.result == nil {
+            await state.setResult(result: value)
+
+            for waiters in await state.waiters {
+                waiters.resume(returning: value)
+            }
+
+            await state.removeAllWaiters()
+
+            return false
+        }
+
+        return true
+    }
+
+    @discardableResult
+    func fail(_ failure: Failure) async -> Bool {
+        if await state.failure == nil {
+            await state.setFailure(failure: failure)
+
+            for waiters in await state.waiters {
+                waiters.resume(throwing: failure)
+            }
+
+            await state.removeAllWaiters()
+
+            return false
+        }
+
+        return true
+    }
+
+    var value: Success {
+        get async throws {
+            try await withCheckedThrowingContinuation { continuation in
+                Task {
+                    if let result = await state.result {
+                        continuation.resume(returning: result)
+                    } else if let failure = await self.state.failure {
+                        continuation.resume(throwing: failure)
+                    } else {
+                        await state.appendWaiters(waiters: continuation)
+                    }
+                }
+            }
+        }
+    }
+}

Sources/DataLoader/Channel/State.swift

@@ -0,0 +1,25 @@
+typealias Waiter<Success, Failure> = CheckedContinuation<Success, Error>
+
+actor State<Success, Failure> {
+    var waiters = [Waiter<Success, Failure>]()
+    var result: Success?
+    var failure: Failure?
+}
+
+extension State {
+    func setResult(result: Success) {
+        self.result = result
+    }
+
+    func setFailure(failure: Failure) {
+        self.failure = failure
+    }
+
+    func appendWaiters(waiters: Waiter<Success, Failure>...) {
+        self.waiters.append(contentsOf: waiters)
+    }
+
+    func removeAllWaiters() {
+        waiters.removeAll()
+    }
+}