Async streaming

Author: FranzBusch

Package.swift

@@ -20,10 +20,24 @@ let availabilityMacros: [SwiftSetting] = [
   ),
 ]
 
+let extraSettings: [SwiftSetting] = [
+  .strictMemorySafety(),
+  .enableExperimentalFeature("SuppressedAssociatedTypes"),
+  .enableExperimentalFeature("LifetimeDependence"),
+  .enableExperimentalFeature("Lifetimes"),
+  .enableUpcomingFeature("LifetimeDependence"),
+  .enableUpcomingFeature("NonisolatedNonsendingByDefault"),
+  .enableUpcomingFeature("InferIsolatedConformances"),
+  .enableUpcomingFeature("ExistentialAny"),
+  .enableUpcomingFeature("MemberImportVisibility"),
+  .enableUpcomingFeature("InternalImportsByDefault"),
+]
+
 let package = Package(
   name: "swift-async-algorithms",
   products: [
-    .library(name: "AsyncAlgorithms", targets: ["AsyncAlgorithms"])
+    .library(name: "AsyncAlgorithms", targets: ["AsyncAlgorithms"]),
+    .library(name: "AsyncStreaming", targets: ["AsyncStreaming"])
   ],
   targets: [
     .target(
@@ -36,6 +50,14 @@ let package = Package(
         .enableExperimentalFeature("StrictConcurrency=complete")
       ]
     ),
+    .target(
+      name: "AsyncStreaming",
+      dependencies: [
+        .product(name: "DequeModule", package: "swift-collections"),
+        .product(name: "BasicContainers", package: "swift-collections"),
+      ],
+      swiftSettings: extraSettings + [.swiftLanguageMode(.v6)]
+    ),
     .target(
       name: "AsyncSequenceValidation",
       dependencies: ["_CAsyncSequenceValidationSupport", "AsyncAlgorithms"],
@@ -51,6 +73,13 @@ let package = Package(
         .enableExperimentalFeature("StrictConcurrency=complete")
       ]
     ),
+    .testTarget(
+      name: "AsyncStreamingTests",
+      dependencies: [
+        .target(name: "AsyncStreaming"),
+      ],
+      swiftSettings: extraSettings + [.swiftLanguageMode(.v6)]
+    ),
     .testTarget(
       name: "AsyncAlgorithmsTests",
       dependencies: [
@@ -95,7 +124,10 @@ let package = Package(
 
 if Context.environment["SWIFTCI_USE_LOCAL_DEPS"] == nil {
   package.dependencies += [
-    .package(url: "https://github.com/apple/swift-collections.git", from: "1.1.0")
+    .package(
+      url: "https://github.com/FranzBusch/swift-collections.git",
+      revision: "53408db248f5bea068579343c47aa0a542dce6c9",
+    )
   ]
 } else {
   package.dependencies += [

Sources/AsyncStreaming/Container/BorrowIteratorProtocol.swift

@@ -0,0 +1,11 @@
+@available(macOS 26.0, iOS 26.0, watchOS 26.0, tvOS 26.0, visionOS 26.0, *)
+public protocol BorrowMutableIteratorProtocol<Element>: ~Copyable, ~Escapable {
+  associatedtype Element: ~Copyable
+
+  // TODO: This is using a closure since we need the exclusive modifier to be
+  // able to extract mutable spans from the underlying owning buffer
+  mutating func nextSpan<Return, Failure>(
+    maximumCount: Int?,
+    body: (inout MutableSpan<Element>) async throws(Failure) -> Return
+  ) async throws(Failure) -> Return
+}

Sources/AsyncStreaming/EitherError.swift

@@ -0,0 +1,54 @@
+/// An enumeration that represents one of two possible error types.
+///
+/// ``EitherError`` provides a type-safe way to represent errors that can be one of two distinct
+/// error types.
+public enum EitherError<First: Error, Second: Error>: Error {
+    /// An error of the first type.
+    ///
+    /// The associated value contains the specific error instance of type `First`.
+    case first(First)
+
+    /// An error of the second type.
+    ///
+    /// The associated value contains the specific error instance of type `Second`.
+    case second(Second)
+
+    /// Throws the underlying error by unwrapping this either error.
+    ///
+    /// This method extracts and throws the actual error contained within the either error,
+    /// whether it's the first or second type. This is useful when you need to propagate
+    /// the original error without the either error wrapper.
+    ///
+    /// - Throws: The underlying error, either of type `First` or `Second`.
+    ///
+    /// ## Example
+    ///
+    /// ```swift
+    /// do {
+    ///     // Some operation that returns EitherError
+    ///     let result = try await operation()
+    /// } catch let eitherError as EitherError<NetworkError, ParseError> {
+    ///     try eitherError.unwrap() // Throws the original error
+    /// }
+    /// ```
+    public func unwrap() throws {
+        switch self {
+        case .first(let first):
+            throw first
+        case .second(let second):
+            throw second
+        }
+    }
+}
+
+@available(macOS 26.0, iOS 26.0, watchOS 26.0, tvOS 26.0, visionOS 26.0, *)
+struct File {
+    @_lifetime(&self)
+    mutating func read() async -> Span<UInt8> {
+        fatalError()
+    }
+    
+    mutating func write(span: Span<UInt8>) async {
+        fatalError()
+    }
+}

Sources/AsyncStreaming/Internal/InlineArray+convenience.swift

@@ -0,0 +1,10 @@
+@available(macOS 26.0, iOS 26.0, watchOS 26.0, tvOS 26.0, visionOS 26.0, *)
+extension InlineArray where Element: ~Copyable {
+    package static func one(value: consuming Element) -> InlineArray<1, Element> {
+        return InlineArray<1, Element>(first: value) { _ in fatalError() }
+    }
+
+    package static func zero(of elementType: Element.Type = Element.self) -> InlineArray<0, Element> {
+        return InlineArray<0, Element> { _ in }
+    }
+}

Sources/AsyncStreaming/Internal/Optional+SendingTake.swift

@@ -0,0 +1,8 @@
+extension Optional where Wrapped: ~Copyable {
+    @inlinable
+    mutating func takeSending() -> sending Self {
+        let result = consume self
+        self = nil
+        return result
+    }
+}

Sources/AsyncStreaming/Reader/Array+AsyncReader.swift

@@ -0,0 +1,85 @@
+@available(macOS 26.0, iOS 26.0, watchOS 26.0, tvOS 26.0, visionOS 26.0, *)
+extension Array {
+    /// Creates an async reader that provides access to the array's elements.
+    ///
+    /// This method converts an array into an ``AsyncReader`` implementation, allowing
+    /// the array's elements to be read through the async reader interface.
+    ///
+    /// - Returns: An ``AsyncReader`` that produces all elements of the array.
+    ///
+    /// ## Example
+    ///
+    /// ```swift
+    /// let numbers = [1, 2, 3, 4, 5]
+    /// var reader = numbers.asyncReader()
+    ///
+    /// try await reader.forEach { span in
+    ///     print("Read \(span.count) numbers")
+    /// }
+    /// ```
+  public func asyncReader() -> some AsyncReader<Element, Never> & SendableMetatype & ~Escapable {
+      return ArrayAsyncReader(array: self)
+    }
+}
+
+/// An async reader implementation that provides array elements through the AsyncReader interface.
+///
+/// This internal reader type wraps an array and delivers its elements through the ``AsyncReader``
+/// protocol. It maintains a current read position and can deliver elements in chunks based on
+/// the requested maximum count.
+@available(macOS 26.0, iOS 26.0, watchOS 26.0, tvOS 26.0, visionOS 26.0, *)
+struct ArrayAsyncReader<Element>: AsyncReader, BorrowMutableIteratorProtocol {
+    typealias ReadIterator = Self
+    typealias Element = Element
+    typealias ReadElement = Element
+    typealias ReadFailure = Never
+
+    var array: [Element]
+    var index: Array<Element>.Index
+
+    init(array: [Element]) {
+        self.array = array
+        self.index = array.startIndex
+    }
+
+  mutating func nextSpan<Return, Failure>(
+    maximumCount: Int?,
+    body: (inout MutableSpan<Element>) async throws(Failure) -> Return
+  ) async throws(Failure) -> Return {
+    guard self.index < self.array.endIndex else {
+      var mutableSpan = self.array.mutableSpan
+      var empty = mutableSpan._mutatingExtracting(last: 0)
+
+      return try await body(&empty)
+    }
+
+    guard let maximumCount else {
+      let index = self.index
+      self.index = self.array.span.indices.endIndex
+      var mutableSpan = self.array.mutableSpan
+      var sizedMutableSpan = mutableSpan._mutatingExtracting(index...)
+      return try await body(&sizedMutableSpan)
+    }
+    let endIndex = min(
+      self.array.span.indices.endIndex,
+      self.index.advanced(
+        by: maximumCount
+      )
+    )
+    self.index = endIndex
+    let index = self.index
+    var mutableSpan = self.array.mutableSpan
+    var sizedMutableSpan = mutableSpan._mutatingExtracting(index..<endIndex)
+    return try await body(&sizedMutableSpan)
+  }
+
+    mutating func read<Return, Failure>(
+      body: (inout ArrayAsyncReader<Element>) async throws(Failure) -> Return
+    ) async throws(EitherError<Never, Failure>) -> Return {
+      do {
+        return try await body(&self)
+      } catch {
+        throw.second(error)
+      }
+    }
+}

Sources/AsyncStreaming/Reader/AsyncReader+forEach.swift

@@ -0,0 +1,94 @@
+// swift-format-ignore: AmbiguousTrailingClosureOverload
+@available(macOS 26.0, iOS 26.0, watchOS 26.0, tvOS 26.0, visionOS 26.0, *)
+extension AsyncReader where Self: ~Copyable, Self: ~Escapable {
+    /// Iterates over all elements from the reader, executing the provided body for each span.
+    ///
+    /// This method continuously reads elements from the async reader until the stream ends,
+    /// executing the provided closure for each span of elements read. The iteration terminates
+    /// when the reader produces an empty span, indicating the end of the stream.
+    ///
+    /// - Parameter body: An asynchronous closure that processes each span of elements read
+    ///   from the stream. The closure receives a `Span<ReadElement>` for each read operation.
+    ///
+    /// - Throws: An `EitherError` containing either a `ReadFailure` from the read operation
+    ///   or a `Failure` from the body closure.
+    ///
+    /// ## Example
+    ///
+    /// ```swift
+    /// var fileReader: FileAsyncReader = ...
+    ///
+    /// // Process each chunk of data from the file
+    /// try await fileReader.forEach { chunk in
+    ///     print("Processing \(chunk.count) elements")
+    ///     // Process the chunk
+    /// }
+    /// ```
+    public consuming func forEach<Failure: Error>(
+        body: (inout MutableSpan<ReadElement>) async throws(Failure) -> Void
+    ) async throws(EitherError<ReadFailure, Failure>) {
+        var shouldContinue = true
+        while shouldContinue {
+            try await self.read { (iterator) throws(Failure) -> Void in
+              try await iterator.nextSpan(maximumCount: 0) { (next) throws(Failure) -> Void in
+                  guard next.count > 0 else {
+                      shouldContinue = false
+                      return
+                  }
+
+                  try await body(&next)
+              }
+            }
+        }
+    }
+
+    /// Iterates over all elements from the reader, executing the provided body for each span.
+    ///
+    /// This method continuously reads elements from the async reader until the stream ends,
+    /// executing the provided closure for each span of elements read. The iteration terminates
+    /// when the reader produces an empty span, indicating the end of the stream.
+    ///
+    /// - Parameter body: An asynchronous closure that processes each span of elements read
+    ///   from the stream. The closure receives a `Span<ReadElement>` for each read operation.
+    ///
+    /// - Throws: An error of type `Failure` from the body closure. Since this reader never fails,
+    ///   only the body closure can throw errors.
+    ///
+    /// ## Example
+    ///
+    /// ```swift
+    /// var fileReader: FileAsyncReader = ...
+    ///
+    /// // Process each chunk of data from the file
+    /// try await fileReader.forEach { chunk in
+    ///     print("Processing \(chunk.count) elements")
+    ///     // Process the chunk
+    /// }
+    /// ```
+    public consuming func forEach<Failure: Error>(
+        body: (inout MutableSpan<ReadElement>) async throws(Failure) -> Void
+    ) async throws(Failure) where ReadFailure == Never {
+        var shouldContinue = true
+        while shouldContinue {
+            do {
+              try await self.read { (iterator) throws(Failure) -> Void in
+                try await iterator.nextSpan(maximumCount: 0) { (next) throws(Failure) -> Void in
+                  guard next.count > 0 else {
+                    shouldContinue = false
+                    return
+                  }
+
+                  try await body(&next)
+                }
+              }
+            } catch {
+                switch error {
+                case .first:
+                    fatalError()
+                case .second(let error):
+                    throw error
+                }
+            }
+        }
+    }
+}