Skeleton for an AsyncSubject/AsyncThrowingSubject implementation

apple#176

Author: rustle

Diff for: Sources/AsyncAlgorithms/AsyncAlgorithms.docc/AsyncAlgorithms.md

@@ -31,6 +31,7 @@ This package has three main goals:
 - <doc:Reductions>
 - <doc:RemoveDuplicates>
 - <doc:Select>
+- <doc:Subject>
 - <doc:Throttle>
 - <doc:Timer>
 - <doc:Zip>

Diff for: Sources/AsyncAlgorithms/AsyncAlgorithms.docc/Guides/Subject.md

@@ -0,0 +1,28 @@
+# Subject
+
+* Author(s): 
+
+[
+[Source](https://github.com/apple/swift-async-algorithms/blob/main/Sources/AsyncAlgorithms/AsyncSubject.swift),
+[Source](https://github.com/apple/swift-async-algorithms/blob/main/Sources/AsyncAlgorithms/AsyncThrowingSubject.swift) |
+[Tests](https://github.com/apple/swift-async-algorithms/blob/main/Tests/AsyncAlgorithmsTests/TestSubject.swift)
+]
+
+## Introduction
+
+ 
+
+## Proposed Solution
+
+
+
+## Detailed Design
+
+ 
+
+## Alternatives Considered
+
+
+
+## Credits/Inspiration
+

Diff for: Sources/AsyncAlgorithms/AsyncSubject.swift

@@ -0,0 +1,80 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2022 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+//
+//===----------------------------------------------------------------------===//
+
+///
+public final class AsyncSubject<Element: Sendable>: AsyncSequence, Sendable {
+  /// The iterator for an `AsyncSubject` instance.
+  public struct Iterator: AsyncIteratorProtocol, Sendable {
+    fileprivate let subject: AsyncSubject<Element>
+    private var active: Bool = true
+
+    fileprivate init(_ subject: AsyncSubject<Element>) {
+      self.subject = subject
+    }
+
+    /// Await the next sent element or finish.
+    public mutating func next() async -> Element? {
+      guard active else {
+        return nil
+      }
+
+      let value = await withTaskCancellationHandler {
+        await subject.next()
+      } onCancel: { [subject] in
+        subject.cancelNext()
+      }
+
+      if let value {
+        return value
+      } else {
+        active = false
+        return nil
+      }
+    }
+  }
+
+  fileprivate func cancelNext() {
+    
+  }
+
+  fileprivate func next() async -> Element? {
+    return await withUnsafeContinuation { (continuation: UnsafeContinuation<Element?, Never>) in
+      fatalError()
+    }
+  }
+
+  func cancelSend() {
+    fatalError()
+  }
+
+  fileprivate func send(_ element: Element) async {
+    fatalError()
+  }
+
+  /// Create a new `AsyncSubject` given an element type.
+  public init(element elementType: Element.Type = Element.self) { }
+
+  ///
+  public func send(_ element: Element) {
+    fatalError()
+  }
+
+  /// Send a finish to all awaiting iterations.
+  /// All subsequent calls to `next(_:)` will resume immediately.
+  public func finish() {
+    
+  }
+
+  /// Create an `Iterator` for iteration of an `AsyncChannel`
+  public func makeAsyncIterator() -> Iterator {
+    Iterator(self)
+  }
+}

Diff for: Sources/AsyncAlgorithms/AsyncThrowingSubject.swift

@@ -0,0 +1,93 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2022 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+//
+//===----------------------------------------------------------------------===//
+
+///
+public final class AsyncThrowingSubject<Element: Sendable, Failure: Error>: AsyncSequence, Sendable {
+  public struct Iterator: AsyncIteratorProtocol, Sendable {
+    let subject: AsyncThrowingSubject<Element, Failure>
+    var active: Bool = true
+    
+    init(_ subject: AsyncThrowingSubject<Element, Failure>) {
+      self.subject = subject
+    }
+
+    public mutating func next() async throws -> Element? {
+      guard active else {
+        return nil
+      }
+
+      do {
+        let value = try await withTaskCancellationHandler {
+          try await subject.next()
+        } onCancel: { [subject] in
+          subject.cancelNext()
+        }
+        
+        if let value = value {
+          return value
+        } else {
+          active = false
+          return nil
+        }
+      } catch {
+        active = false
+        throw error
+      }
+    }
+  }
+
+  /// Create a new `AsyncThrowingSubject` given an element type.
+  public init(_ elementType: Element.Type = Element.self) { }
+
+  fileprivate func cancelNext() {
+    fatalError()
+  }
+
+  fileprivate func next() async throws -> Element? {
+    fatalError()
+  }
+
+  fileprivate func cancelSend() {
+    fatalError()
+  }
+
+  fileprivate func send(_ element: Element) async {
+    let continuation: UnsafeContinuation<Element?, Error>? = await withUnsafeContinuation { continuation in
+      fatalError()
+    }
+    continuation?.resume(returning: element)
+  }
+
+  fileprivate func terminateAll(error: Failure? = nil) {
+    fatalError()
+  }
+
+  ///
+  public func send(_ element: Element) {
+    fatalError()
+  }
+
+  /// Send an error to all awaiting iterations.
+  /// All subsequent calls to `next(_:)` will resume immediately.
+  public func fail(_ error: Error) where Failure == Error {
+    terminateAll(error: error)
+  }
+
+  /// Send a finish to all awaiting iterations.
+  /// All subsequent calls to `next(_:)` will resume immediately.
+  public func finish() {
+    terminateAll()
+  }
+
+  public func makeAsyncIterator() -> Iterator {
+    Iterator(self)
+  }
+}

Diff for: Tests/AsyncAlgorithmsTests/TestSubject.swift

@@ -0,0 +1,17 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2022 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+//
+//===----------------------------------------------------------------------===//
+
+@preconcurrency import XCTest
+import AsyncAlgorithms
+
+final class TestChannel: XCTestCase {
+    
+}