Add new experimental _AsyncStreaming module

Package.swift

@@ -37,6 +37,20 @@ let package = Package(
   products: [
     .library(name: "AsyncAlgorithms", targets: ["AsyncAlgorithms"])
   ],
+  traits: [
+    .default(
+      enabledTraits: [
+        //        "UnstableAsyncStreaming"
+      ]
+    ),
+    .trait(
+      name: "UnstableAsyncStreaming",
+      description: """
+        Enables source unstable async streaming components in the _AsyncStreaming
+        module. Do not rely on this module in API stable packages.
+        """
+    ),
+  ],
   targets: [
     .target(
       name: "AsyncAlgorithms",
@@ -48,6 +62,24 @@ let package = Package(
         .enableExperimentalFeature("StrictConcurrency=complete")
       ]
     ),
+    .target(
+      name: "AsyncStreaming",
+      dependencies: [
+        .product(name: "BasicContainers", package: "swift-collections"),
+        .product(name: "ContainersPreview", package: "swift-collections"),
+      ],
+      swiftSettings: [
+        .enableExperimentalFeature("SuppressedAssociatedTypesWithDefaults"),
+        .enableExperimentalFeature("LifetimeDependence"),
+        .enableExperimentalFeature("Lifetimes"),
+        .enableUpcomingFeature("LifetimeDependence"),
+        .enableUpcomingFeature("NonisolatedNonsendingByDefault"),
+        .enableUpcomingFeature("InferIsolatedConformances"),
+        .enableUpcomingFeature("ExistentialAny"),
+        .enableUpcomingFeature("MemberImportVisibility"),
+        .enableUpcomingFeature("InternalImportsByDefault"),
+      ]
+    ),
     .target(
       name: "AsyncSequenceValidation",
       dependencies: ["_CAsyncSequenceValidationSupport", "AsyncAlgorithms"],
@@ -104,12 +136,35 @@ let package = Package(
         .enableExperimentalFeature("StrictConcurrency=complete")
       ]
     ),
+    .testTarget(
+      name: "AsyncStreamingTests",
+      dependencies: [
+        "AsyncStreaming",
+        .product(name: "BasicContainers", package: "swift-collections"),
+        .product(name: "ContainersPreview", package: "swift-collections"),
+      ],
+      swiftSettings: [
+        .enableExperimentalFeature("SuppressedAssociatedTypesWithDefaults"),
+        .enableExperimentalFeature("LifetimeDependence"),
+        .enableExperimentalFeature("Lifetimes"),
+        .enableUpcomingFeature("LifetimeDependence"),
+        .enableUpcomingFeature("NonisolatedNonsendingByDefault"),
+        .enableUpcomingFeature("InferIsolatedConformances"),
+        .enableUpcomingFeature("ExistentialAny"),
+        .enableUpcomingFeature("MemberImportVisibility"),
+        .enableUpcomingFeature("InternalImportsByDefault"),
+      ]
+    ),
   ]
 )
 
 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/apple/swift-collections.git",
+      from: "1.5.0",
+      traits: [.trait(name: "UnstableContainersPreview", condition: .when(traits: ["UnstableAsyncStreaming"]))]
+    )
   ]
 } else {
   package.dependencies += [

Sources/AsyncStreaming/AsyncReader/AsyncReader.swift

@@ -0,0 +1,69 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2026 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
+//
+//===----------------------------------------------------------------------===//
+
+#if UnstableAsyncStreaming && compiler(>=6.4)
+public import ContainersPreview
+
+/// Reads elements asynchronously from a source.
+///
+/// Adopt ``AsyncReader`` when you need to provide callee-managed buffering,
+/// where the reader controls the buffer and passes a span of elements
+/// to the caller through the `body` closure.
+@available(macOS 10.14.4, iOS 12.2, watchOS 5.2, tvOS 12.2, visionOS 1.0, *)
+public protocol AsyncReader<ReadElement, ReadFailure>: ~Copyable, ~Escapable {
+  /// The type of elements this reader reads.
+  associatedtype ReadElement: ~Copyable
+
+  /// The error type that reading operations throw.
+  associatedtype ReadFailure: Error
+
+  /// Reads elements from the underlying source and passes them to the provided body closure.
+  ///
+  /// This method asynchronously reads a span of elements from the source,
+  /// then passes them to `body` for processing.
+  ///
+  /// ```swift
+  /// var fileReader: FileAsyncReader = ...
+  ///
+  /// // Read data from a file asynchronously and process it.
+  /// let result = try await fileReader.read { data in
+  ///     guard data.count > 0 else {
+  ///         return
+  ///     }
+  ///     return data
+  /// }
+  /// ```
+  ///
+  /// - Parameter maximumCount: The maximum count of items you're ready
+  ///   to process. Must be greater than zero.
+  /// - Parameter body: A closure that processes a span of read elements
+  ///   and returns a value of type `Return`. When the span is empty,
+  ///   it indicates the end of the stream.
+  /// - Returns: The value the body closure returns after processing the read elements.
+  /// - Throws: An `EitherError` containing either a `ReadFailure` from the read operation
+  ///   or a `Failure` from the body closure.
+  mutating func read<Return: ~Copyable, Failure: Error>(
+    maximumCount: Int,
+    body: (consuming InputSpan<ReadElement>) async throws(Failure) -> Return
+  ) async throws(EitherError<ReadFailure, Failure>) -> Return
+
+}
+
+@available(macOS 10.14.4, iOS 12.2, watchOS 5.2, tvOS 12.2, visionOS 1.0, *)
+extension AsyncReader where Self: ~Copyable, Self: ~Escapable {
+  /// Reads elements with no upper bound on span size.
+  public mutating func read<Return: ~Copyable, Failure: Error>(
+    body: (consuming InputSpan<ReadElement>) async throws(Failure) -> Return
+  ) async throws(EitherError<ReadFailure, Failure>) -> Return {
+    try await read(maximumCount: .max, body: body)
+  }
+}
+#endif

Sources/AsyncStreaming/AsyncWriter/AsyncWriter.swift

@@ -0,0 +1,56 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2026 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
+//
+//===----------------------------------------------------------------------===//
+
+#if UnstableAsyncStreaming && compiler(>=6.4)
+/// Writes elements asynchronously to a destination using a provided buffer.
+///
+/// Adopt ``AsyncWriter`` when you need to provide callee-managed buffering,
+/// where the writer supplies an output span buffer that the caller fills
+/// with elements to write.
+@available(macOS 26.2, iOS 26.2, watchOS 26.2, tvOS 26.2, visionOS 26.2, *)
+public protocol AsyncWriter<WriteElement, WriteFailure>: ~Copyable, ~Escapable {
+  /// The type of elements this writer writes.
+  associatedtype WriteElement: ~Copyable
+
+  /// The error type that writing operations throw.
+  associatedtype WriteFailure: Error
+
+  /// Provides a buffer for writing elements to the destination.
+  ///
+  /// The writer supplies an output span that `body` uses to append elements.
+  /// The writer manages the buffer allocation and handles the writing
+  /// operation once `body` completes.
+  ///
+  /// - Parameter body: A closure that receives an `OutputSpan` for appending elements
+  ///   to write. The closure returns a result of type `Return`.
+  ///
+  /// - Returns: The value the body closure returns.
+  ///
+  /// - Throws: An `EitherError` containing either a `WriteFailure` from the write operation
+  ///   or a `Failure` from the body closure.
+  ///
+  /// ## Example
+  ///
+  /// ```swift
+  /// var writer: SomeAsyncWriter = ...
+  ///
+  /// try await writer.write { outputSpan in
+  ///     for item in items {
+  ///         outputSpan.append(item)
+  ///     }
+  ///     return outputSpan.count
+  /// }
+  /// ```
+  mutating func write<Return: ~Copyable, Failure: Error>(
+    _ body: (inout OutputSpan<WriteElement>) async throws(Failure) -> Return
+  ) async throws(EitherError<WriteFailure, Failure>) -> Return
+}
+#endif

Sources/AsyncStreaming/CallerAsyncReader/CallerAsyncReader.swift

@@ -0,0 +1,37 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2026 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
+//
+//===----------------------------------------------------------------------===//
+
+#if UnstableAsyncStreaming && compiler(>=6.4)
+/// Reads elements asynchronously into a caller-provided buffer.
+///
+/// Adopt ``CallerAsyncReader`` when you need caller-managed buffering,
+/// where the caller supplies an output span that the reader fills
+/// with elements.
+@available(macOS 10.14.4, iOS 12.2, watchOS 5.2, tvOS 12.2, visionOS 1.0, *)
+public protocol CallerAsyncReader<ReadElement, ReadFailure>: ~Copyable, ~Escapable {
+  /// The type of elements this reader reads.
+  associatedtype ReadElement: ~Copyable
+
+  /// The error type that reading operations throw.
+  associatedtype ReadFailure: Error
+
+  /// Reads elements from the source into the provided buffer.
+  ///
+  /// This method appends elements into `buffer`. When the read operation
+  /// reaches the end of the source, it appends no elements.
+  ///
+  /// - Parameter buffer: The output span to fill with read elements.
+  /// - Throws: A `ReadFailure` from the underlying read operation.
+  mutating func read(
+    into buffer: inout OutputSpan<ReadElement>
+  ) async throws(ReadFailure)
+}
+#endif

Sources/AsyncStreaming/CallerAsyncWriter/CallerAsyncWriter.swift

@@ -0,0 +1,51 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2026 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
+//
+//===----------------------------------------------------------------------===//
+
+#if UnstableAsyncStreaming && compiler(>=6.4)
+public import ContainersPreview
+
+/// Writes elements asynchronously from a caller-provided span.
+///
+/// Adopt ``CallerAsyncWriter`` when you need caller-managed buffering,
+/// where the caller provides a span of elements for the writer
+/// to consume.
+@available(macOS 10.14.4, iOS 12.2, watchOS 5.2, tvOS 12.2, visionOS 1.0, *)
+public protocol CallerAsyncWriter<WriteElement, WriteFailure>: ~Copyable, ~Escapable {
+  /// The type of elements this writer writes.
+  associatedtype WriteElement: ~Copyable
+
+  /// The error type that writing operations throw.
+  associatedtype WriteFailure: Error
+
+  /// Writes a span of elements to the underlying destination.
+  ///
+  /// This method asynchronously writes all elements from the provided span to whatever destination
+  /// the writer represents. The operation may require multiple write calls to complete if the
+  /// writer cannot accept all elements at once.
+  ///
+  /// - Parameter span: The span of elements to write. If not all elements can be written, `span` will be non-empty after `write` returns
+  ///
+  /// - Throws: A `WriteFailure` from the underlying write operation
+  ///
+  /// ## Example
+  ///
+  /// ```swift
+  /// var fileWriter: FileAsyncWriter = ...
+  /// let dataBuffer: [UInt8] = [1, 2, 3, 4, 5]
+  ///
+  /// // Write the entire span to a file asynchronously
+  /// try await fileWriter.write(dataBuffer.span)
+  /// ```
+  mutating func write(
+    span: borrowing InputSpan<WriteElement>
+  ) async throws(WriteFailure)
+}
+#endif

Sources/AsyncStreaming/EitherError.swift

@@ -0,0 +1,56 @@
+//===----------------------------------------------------------------------===//
+//
+// 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
+//
+//===----------------------------------------------------------------------===//
+
+#if UnstableAsyncStreaming && compiler(>=6.4)
+/// A type-safe wrapper around one of two distinct error types.
+///
+/// Use ``EitherError`` when an operation can fail with errors from two
+/// different sources, such as a read failure and a body closure failure.
+@frozen
+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, unwrapping this either error.
+  ///
+  /// This method extracts and throws the error in the either error,
+  /// whether it's the first or second type. Use this 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
+    }
+  }
+}
+#endif