Implement AsyncThrowingStream.nextElement() with typed throws

Author: DougGregor

stdlib/public/Concurrency/AsyncThrowingStream.swift

@@ -254,9 +254,9 @@ public struct AsyncThrowingStream<Element, Failure: Error> {
 
   final class _Context {
     let storage: _Storage?
-    let produce: () async throws -> Element?
+    let produce: () async throws(Failure) -> Element?
 
-    init(storage: _Storage? = nil, produce: @escaping () async throws -> Element?) {
+    init(storage: _Storage? = nil, produce: @escaping () async throws(Failure) -> Element?) {
       self.storage = storage
       self.produce = produce
     }
@@ -415,6 +415,24 @@ extension AsyncThrowingStream: AsyncSequence {
     public mutating func next() async throws -> Element? {
       return try await context.produce()
     }
+
+    /// The next value from the asynchronous stream.
+    ///
+    /// When `nextElement()` returns `nil`, this signifies the end of the
+    /// `AsyncThrowingStream`.
+    ///
+    /// It is a programmer error to invoke `nextElement()` from a concurrent
+    /// context that contends with another such call, which results in a call to
+    /// `fatalError()`.
+    ///
+    /// If you cancel the task this iterator is running in while `nextElement()`
+    /// is awaiting a value, the `AsyncThrowingStream` terminates. In this case,
+    /// `nextElement()` may return `nil` immediately, or else return `nil` on
+    /// subsequent calls.
+    @available(SwiftStdlib 5.11, *)
+    public mutating func nextElement() async throws(Failure) -> Element? {
+      return try await context.produce()
+    }
   }
 
   /// Creates the asynchronous iterator that produces elements of this