revised to support AsyncChannel and AsyncThrowingChannel as the primary types

Author: Brennan Stehling

Sources/AsyncChannelKit/AsyncChannel.swift

@@ -1,71 +1,63 @@
 import Foundation
 
-public actor AsyncChannel<Element: Sendable> {
-    public enum Failure: Error {
+public actor AsyncChannel<Element: Sendable>: AsyncSequence {
+    public struct Iterator: AsyncIteratorProtocol, Sendable {
+        private let channel: AsyncChannel<Element>
+
+        public init(_ channel: AsyncChannel<Element>) {
+            self.channel = channel
+        }
+
+        public mutating func next() async -> Element? {
+            await channel.next()
+        }
+    }
+
+    public enum InternalFailure: Error {
         case cannotSendAfterTerminated
     }
-    public typealias ChannelContinuation = CheckedContinuation<Element?, Error>
+    public typealias ChannelContinuation = CheckedContinuation<Element?, Never>
 
     private var continuations: [ChannelContinuation] = []
     private var elements: [Element] = []
     private var terminated: Bool = false
-    private var error: Error? = nil
 
     private var hasNext: Bool {
         !continuations.isEmpty && !elements.isEmpty
     }
 
-    private var canFail: Bool {
-        error != nil && !continuations.isEmpty
-    }
-
     private var canTerminate: Bool {
         terminated && elements.isEmpty && !continuations.isEmpty
     }
 
     public init() {
     }
 
-    public func next() async throws -> Element? {
-        try await withCheckedThrowingContinuation { (continuation: ChannelContinuation) in
+    public nonisolated func makeAsyncIterator() -> Iterator {
+        Iterator(self)
+    }
+
+    public func next() async -> Element? {
+        await withCheckedContinuation { (continuation: ChannelContinuation) in
             continuations.append(continuation)
             processNext()
         }
     }
 
-    public func send(element: Element) throws {
+    public func send(_ element: Element) throws {
         guard !terminated else {
-            throw Failure.cannotSendAfterTerminated
+            throw InternalFailure.cannotSendAfterTerminated
         }
         elements.append(element)
         processNext()
     }
 
-    public func send(error: Error) throws {
-        guard !terminated else {
-            throw Failure.cannotSendAfterTerminated
-        }
-        self.error = error
-        processNext()
-    }
-
-    public func terminate() {
+    public func finish() {
         terminated = true
         processNext()
     }
 
     private func processNext() {
-        if canFail {
-            let contination = continuations.removeFirst()
-            assert(continuations.isEmpty)
-            assert(elements.isEmpty)
-            assert(error != nil)
-            if let error = error {
-                contination.resume(throwing: error)
-                return
-            }
-        }
-
         if canTerminate {
             let contination = continuations.removeFirst()
             assert(continuations.isEmpty)

Sources/AsyncChannelKit/AsyncChannelIterator.swift

@@ -0,0 +1,103 @@
+import Foundation
+
+public actor AsyncThrowingChannel<Element: Sendable, Failure: Error>: AsyncSequence {
+    public struct Iterator: AsyncIteratorProtocol, Sendable {
+        private let channel: AsyncThrowingChannel<Element, Failure>
+
+        public init(_ channel: AsyncThrowingChannel<Element, Failure>) {
+            self.channel = channel
+        }
+
+        public mutating func next() async throws -> Element? {
+            try await channel.next()
+        }
+    }
+
+    public enum InternalFailure: Error {
+        case cannotSendAfterTerminated
+    }
+    public typealias ChannelContinuation = CheckedContinuation<Element?, Error>
+
+    private var continuations: [ChannelContinuation] = []
+    private var elements: [Element] = []
+    private var terminated: Bool = false
+    private var error: Error? = nil
+
+    private var hasNext: Bool {
+        !continuations.isEmpty && !elements.isEmpty
+    }
+
+    private var canFail: Bool {
+        error != nil && !continuations.isEmpty
+    }
+
+    private var canTerminate: Bool {
+        terminated && elements.isEmpty && !continuations.isEmpty
+    }
+
+    public init() {
+    }
+
+    public nonisolated func makeAsyncIterator() -> Iterator {
+        Iterator(self)
+    }
+
+    public func next() async throws -> Element? {
+        try await withCheckedThrowingContinuation { (continuation: ChannelContinuation) in
+            continuations.append(continuation)
+            processNext()
+        }
+    }
+
+    public func send(_ element: Element) throws {
+        guard !terminated else {
+            throw InternalFailure.cannotSendAfterTerminated
+        }
+        elements.append(element)
+        processNext()
+    }
+
+
+    public func fail(_ error: Error) where Failure == Error {
+        self.error = error
+        processNext()
+    }
+
+    public func finish() {
+        terminated = true
+        processNext()
+    }
+
+    private func processNext() {
+        if canFail {
+            let contination = continuations.removeFirst()
+            assert(continuations.isEmpty)
+            assert(elements.isEmpty)
+            assert(error != nil)
+            if let error = error {
+                contination.resume(throwing: error)
+                return
+            }
+        }
+
+        if canTerminate {
+            let contination = continuations.removeFirst()
+            assert(continuations.isEmpty)
+            assert(elements.isEmpty)
+            contination.resume(returning: nil)
+            return
+        }
+
+        guard hasNext else {
+            return
+        }
+
+        assert(!continuations.isEmpty)
+        assert(!elements.isEmpty)
+
+        let contination = continuations.removeFirst()
+        let element = elements.removeFirst()
+
+        contination.resume(returning: element)
+    }
+}

Sources/AsyncChannelKit/AsyncChannelSequence.swift

@@ -17,7 +17,6 @@ final class AsyncChannelKitTests: XCTestCase {
     func testNumberSequence() async throws {
         let input = [1, 2, 3, 4, 5]
         let channel = AsyncChannel<Int>()
-        let sequence = AsyncChannelSequence(channel: channel)
 
         // load all numbers into the channel with delays
         Task {
@@ -27,7 +26,7 @@ final class AsyncChannelKitTests: XCTestCase {
         var output: [Int] = []
 
         print("-- before --")
-        for try await element in sequence {
+        for await element in channel {
             print(element)
             output.append(element)
         }
@@ -39,7 +38,6 @@ final class AsyncChannelKitTests: XCTestCase {
     func testStringSequence() async throws {
         let input = ["one", "two", "three", "four", "five"]
         let channel = AsyncChannel<String>()
-        let sequence = AsyncChannelSequence(channel: channel)
 
         // load all strings into the channel with delays
         Task {
@@ -49,7 +47,7 @@ final class AsyncChannelKitTests: XCTestCase {
         var output: [String] = []
 
         print("-- before --")
-        for try await element in sequence {
+        for await element in channel {
             print(element)
             output.append(element)
         }
@@ -58,11 +56,42 @@ final class AsyncChannelKitTests: XCTestCase {
         XCTAssertEqual(input, output)
     }
 
+    func testSucceedingSequence() async throws {
+        let input = [3, 7, 14, 21]
+        let channel = AsyncThrowingChannel<Int, Error>()
+
+        // load all numbers into the channel with delays
+        Task {
+            try await send(elements: input, channel: channel, sleepSeconds: sleepSeconds) { element in
+                if element == 13 {
+                    throw Failure.unluckyNumber
+                } else {
+                    return element
+                }
+            }
+        }
+
+        var output: [Int] = []
+        var thrown: Error? = nil
+
+        print("-- before --")
+        do {
+            for try await element in channel {
+                print(element)
+                output.append(element)
+            }
+        } catch {
+            thrown = error
+        }
+        print("-- after --")
+
+        XCTAssertNil(thrown)
+        XCTAssertEqual(input, output)
+    }
 
     func testFailingSequence() async throws {
         let input = [3, 7, 13, 21]
-        let channel = AsyncChannel<Int>()
-        let sequence = AsyncChannelSequence(channel: channel)
+        let channel = AsyncThrowingChannel<Int, Error>()
 
         // load all numbers into the channel with delays
         Task {
@@ -80,7 +109,7 @@ final class AsyncChannelKitTests: XCTestCase {
 
         print("-- before --")
         do {
-            for try await element in sequence {
+            for try await element in channel {
                 print(element)
                 output.append(element)
             }
@@ -94,25 +123,37 @@ final class AsyncChannelKitTests: XCTestCase {
         XCTAssertEqual(expected, output)
     }
 
-    private func send<Element>(elements: [Element], channel: AsyncChannel<Element>, sleepSeconds: Double = 0.1, processor: ((Element) throws -> Element)? = nil) async throws {
+    private func send<Element>(elements: [Element], channel: AsyncChannel<Element>, sleepSeconds: Double = 0.1) async throws {
+        var index = 0
+        while index < elements.count {
+            try await Task.sleep(seconds: sleepSeconds)
+            let element = elements[index]
+            try await channel.send(element)
+
+            index += 1
+        }
+        await channel.finish()
+    }
+
+    private func send<Element>(elements: [Element], channel: AsyncThrowingChannel<Element, Error>, sleepSeconds: Double = 0.1, processor: ((Element) throws -> Element)? = nil) async throws {
         var index = 0
         while index < elements.count {
             try await Task.sleep(seconds: sleepSeconds)
             let element = elements[index]
             if let processor = processor {
                 do {
                     let processed = try processor(element)
-                    try await channel.send(element: processed)
+                    try await channel.send(processed)
                 } catch {
                     print("throwing \(error)")
-                    try await channel.send(error: error)
+                    await channel.fail(error)
                 }
             } else {
-                try await channel.send(element: element)
+                try await channel.send(element)
             }
 
             index += 1
         }
-        await channel.terminate()
+        await channel.finish()
     }
 }