Implement nextElement() on most async sequences that have rethrowing next()

Start implementing `nextElement()` for the various async sequences
provided by the concurrency library, starting with those that
currently depend on `rethrows` with conformances, an undocumented
feature we're trying to stage out of the language.

With the exception of `AsyncFlatMapSequence`, all of these async
sequences have obvious implementations of `nextElement()`. We'll save
that one for later.

Author: DougGregor

stdlib/public/Concurrency/AsyncCompactMapSequence.swift

@@ -81,6 +81,12 @@ extension AsyncCompactMapSequence: AsyncSequence {
   /// The compact map sequence produces whatever type of element its
   /// transforming closure produces.
   public typealias Element = ElementOfResult
+  /// The type of the error that can be produced by the sequence.
+  ///
+  /// The compact map sequence produces whatever type of error its
+  /// base sequence does.
+  @available(SwiftStdlib 5.11, *)
+  public typealias Failure = Base.Failure
   /// The type of iterator that produces elements of the sequence.
   public typealias AsyncIterator = Iterator
 
@@ -123,6 +129,28 @@ extension AsyncCompactMapSequence: AsyncSequence {
         }
       }
     }
+
+    /// Produces the next element in the compact map sequence.
+    ///
+    /// This iterator calls `next()` on its base iterator; if this call returns
+    /// `nil`, `next()` returns `nil`. Otherwise, `next()` calls the
+    /// transforming closure on the received element, returning it if the
+    /// transform returns a non-`nil` value. If the transform returns `nil`,
+    /// this method continues to wait for further elements until it gets one
+    /// that transforms to a non-`nil` value.
+    @available(SwiftStdlib 5.11, *)
+    @inlinable
+    public mutating func nextElement() async throws(Failure) -> ElementOfResult? {
+      while true {
+        guard let element = try await baseIterator.nextElement() else {
+          return nil
+        }
+
+        if let transformed = await transform(element) {
+          return transformed
+        }
+      }
+    }
   }
 
   @inlinable

stdlib/public/Concurrency/AsyncDropFirstSequence.swift

@@ -70,6 +70,12 @@ extension AsyncDropFirstSequence: AsyncSequence {
   /// The drop-first sequence produces whatever type of element its base
   /// iterator produces.
   public typealias Element = Base.Element
+  /// The type of errors produced by this asynchronous sequence.
+  ///
+  /// The drop-first sequence produces whatever type of error its base
+  /// sequence produces.
+  @available(SwiftStdlib 5.11, *)
+  public typealias Failure = Base.Failure
   /// The type of iterator that produces elements of the sequence.
   public typealias AsyncIterator = Iterator
 
@@ -108,6 +114,29 @@ extension AsyncDropFirstSequence: AsyncSequence {
       count = 0
       return try await baseIterator.next()
     }
+
+    /// Produces the next element in the drop-first sequence.
+    ///
+    /// Until reaching the number of elements to drop, this iterator calls
+    /// `nextElement()` on its base iterator and discards the result. If the
+    /// base iterator returns `nil`, indicating the end of the sequence, this
+    /// iterator returns `nil`. After reaching the number of elements to drop,
+    /// this iterator passes along the result of calling `nextElement()` on the
+    /// base iterator.
+    @available(SwiftStdlib 5.11, *)
+    @inlinable
+    public mutating func nextElement() async throws(Failure) -> Base.Element? {
+      var remainingToDrop = count
+      while remainingToDrop > 0 {
+        guard try await baseIterator.nextElement() != nil else {
+          count = 0
+          return nil
+        }
+        remainingToDrop -= 1
+      }
+      count = 0
+      return try await baseIterator.nextElement()
+    }
   }
 
   @inlinable

stdlib/public/Concurrency/AsyncDropWhileSequence.swift

@@ -79,6 +79,12 @@ extension AsyncDropWhileSequence: AsyncSequence {
   /// The drop-while sequence produces whatever type of element its base
   /// sequence produces.
   public typealias Element = Base.Element
+  /// The type of errors produced by this asynchronous sequence.
+  ///
+  /// The drop-while sequence produces whatever type of error its base
+  /// sequence produces.
+  @available(SwiftStdlib 5.11, *)
+  public typealias Failure = Base.Failure
   /// The type of iterator that produces elements of the sequence.
   public typealias AsyncIterator = Iterator
 
@@ -120,6 +126,30 @@ extension AsyncDropWhileSequence: AsyncSequence {
       }
       return try await baseIterator.next()
     }
+
+    /// Produces the next element in the drop-while sequence.
+    ///
+    /// This iterator calls `nextElement()` on its base iterator and evaluates
+    /// the result with the `predicate` closure. As long as the predicate
+    /// returns `true`, this method returns `nil`. After the predicate returns
+    /// `false`, for a value received from the base iterator, this method
+    /// returns that value. After that, the iterator returns values received
+    /// from its base iterator as-is, and never executes the predicate closure
+    /// again.
+    @available(SwiftStdlib 5.11, *)
+    @inlinable
+    public mutating func nextElement() async throws(Failure) -> Base.Element? {
+      while let predicate = self.predicate {
+        guard let element = try await baseIterator.nextElement() else {
+          return nil
+        }
+        if await predicate(element) == false {
+          self.predicate = nil
+          return element
+        }
+      }
+      return try await baseIterator.nextElement()
+    }
   }
 
   /// Creates an instance of the drop-while sequence iterator.

stdlib/public/Concurrency/AsyncFilterSequence.swift

@@ -69,6 +69,12 @@ extension AsyncFilterSequence: AsyncSequence {
   /// The filter sequence produces whatever type of element its base
   /// sequence produces.
   public typealias Element = Base.Element
+  /// The type of the error that can be produced by the sequence.
+  ///
+  /// The filter sequence produces whatever type of error its
+  /// base sequence does.
+  @available(SwiftStdlib 5.11, *)
+  public typealias Failure = Base.Failure
   /// The type of iterator that produces elements of the sequence.
   public typealias AsyncIterator = Iterator
 
@@ -107,6 +113,26 @@ extension AsyncFilterSequence: AsyncSequence {
         }
       }
     }
+
+    /// Produces the next element in the filter sequence.
+    ///
+    /// This iterator calls `nextelement()` on its base iterator; if this call
+    /// returns `nil`, `nextElement()` returns nil. Otherwise, `nextElement()`
+    /// evaluates the result with the `predicate` closure. If the closure
+    /// returns `true`, `nextElement()` returns the received element; otherwise
+    /// it awaits the next element from the base iterator.
+    @available(SwiftStdlib 5.11, *)
+    @inlinable
+    public mutating func nextElement() async throws(Failure) -> Base.Element? {
+      while true {
+        guard let element = try await baseIterator.nextElement() else {
+          return nil
+        }
+        if await isIncluded(element) {
+          return element
+        }
+      }
+    }
   }
 
   @inlinable

stdlib/public/Concurrency/AsyncIteratorProtocol.swift

@@ -112,7 +112,8 @@ public protocol AsyncIteratorProtocol<Element, Failure> {
 extension AsyncIteratorProtocol {
   /// Default implementation of `nextElement()` in terms of `next()`, which is
   /// required to maintain backward compatibility with existing async iterators.
-  @_alwaysEmitIntoClient
+  @available(SwiftStdlib 5.11, *)
+  @inlinable
   public mutating func nextElement() async throws(Failure) -> Element? {
     do {
       return try await next()

stdlib/public/Concurrency/AsyncMapSequence.swift

@@ -79,6 +79,12 @@ extension AsyncMapSequence: AsyncSequence {
   /// The map sequence produces whatever type of element its transforming
   /// closure produces.
   public typealias Element = Transformed
+  /// The type of the error that can be produced by the sequence.
+  ///
+  /// The map sequence produces whatever type of error its
+  /// base sequence does.
+  @available(SwiftStdlib 5.11, *)
+  public typealias Failure = Base.Failure
   /// The type of iterator that produces elements of the sequence.
   public typealias AsyncIterator = Iterator
 
@@ -111,6 +117,20 @@ extension AsyncMapSequence: AsyncSequence {
       }
       return await transform(element)
     }
+
+    /// Produces the next element in the map sequence.
+    ///
+    /// This iterator calls `next()` on its base iterator; if this call returns
+    /// `nil`, `next()` returns `nil`. Otherwise, `next()` returns the result of
+    /// calling the transforming closure on the received element.
+    @available(SwiftStdlib 5.11, *)
+    @inlinable
+    public mutating func nextElement() async throws(Failure) -> Transformed? {
+      guard let element = try await baseIterator.nextElement() else {
+        return nil
+      }
+      return await transform(element)
+    }
   }
 
   @inlinable

stdlib/public/Concurrency/AsyncPrefixSequence.swift

@@ -70,6 +70,12 @@ extension AsyncPrefixSequence: AsyncSequence {
   /// The prefix sequence produces whatever type of element its base iterator
   /// produces.
   public typealias Element = Base.Element
+  /// The type of the error that can be produced by the sequence.
+  ///
+  /// The prefix sequence produces whatever type of error its
+  /// base sequence does.
+  @available(SwiftStdlib 5.11, *)
+  public typealias Failure = Base.Failure
   /// The type of iterator that produces elements of the sequence.
   public typealias AsyncIterator = Iterator
 
@@ -102,6 +108,23 @@ extension AsyncPrefixSequence: AsyncSequence {
         return nil
       }
     }
+
+    /// Produces the next element in the prefix sequence.
+    ///
+    /// Until reaching the number of elements to include, this iterator calls
+    /// `nextElement()` on its base iterator and passes through the
+    /// result. After reaching the maximum number of elements, subsequent calls
+    /// to `nextElement()` return `nil`.
+    @available(SwiftStdlib 5.11, *)
+    @inlinable
+    public mutating func nextElement() async throws(Failure) -> Base.Element? {
+      if remaining != 0 {
+        remaining &-= 1
+        return try await baseIterator.nextElement()
+      } else {
+        return nil
+      }
+    }
   }
 
   @inlinable

stdlib/public/Concurrency/AsyncPrefixWhileSequence.swift

@@ -74,6 +74,12 @@ extension AsyncPrefixWhileSequence: AsyncSequence {
   /// The prefix-while sequence produces whatever type of element its base
   /// iterator produces.
   public typealias Element = Base.Element
+  /// The type of the error that can be produced by the sequence.
+  ///
+  /// The prefix-while sequence produces whatever type of error its
+  /// base sequence does.
+  @available(SwiftStdlib 5.11, *)
+  public typealias Failure = Base.Failure
   /// The type of iterator that produces elements of the sequence.
   public typealias AsyncIterator = Iterator
 
@@ -114,6 +120,25 @@ extension AsyncPrefixWhileSequence: AsyncSequence {
       }
       return nil
     }
+
+    /// Produces the next element in the prefix-while sequence.
+    ///
+    /// If the predicate hasn't yet failed, this method gets the next element
+    /// from the base sequence and calls the predicate with it. If this call
+    /// succeeds, this method passes along the element. Otherwise, it returns
+    /// `nil`, ending the sequence.
+    @available(SwiftStdlib 5.11, *)
+    @inlinable
+    public mutating func nextElement() async throws(Failure) -> Base.Element? {
+      if !predicateHasFailed, let nextElement = try await baseIterator.nextElement() {
+        if await predicate(nextElement) {
+          return nextElement
+        } else {
+          predicateHasFailed = true
+        }
+      }
+      return nil
+    }
   }
 
   @inlinable