Isolated When Possible

README.md

@@ -27,7 +27,10 @@ Please check the [releases](https://github.com/mochidev/AsyncSequenceReader/rele
 
 ```swift
 dependencies: [
-    .package(url: "https://github.com/mochidev/AsyncSequenceReader.git", .upToNextMinor(from: "0.4.1")),
+    .package(
+        url: "https://github.com/mochidev/AsyncSequenceReader.git",
+        .upToNextMinor(from: "0.5.0")
+    ),
 ],
 ...
 targets: [

Sources/AsyncSequenceReader/AsyncIteratorMapSequence.swift

@@ -155,7 +155,7 @@ public struct AsyncIteratorMapSequence<Base: AsyncSequence, Transformed, Transfo
     @usableFromInline
     init(
         _ base: Base,
-        transform: @escaping (_ iterator: inout AsyncBufferedIterator<Base.AsyncIterator>) async throws(TransformFailure) -> Transformed
+        transform: sending @escaping (_ iterator: inout AsyncBufferedIterator<Base.AsyncIterator>) async throws(TransformFailure) -> Transformed
     ) {
         self.base = base
         self.transform = transform

Sources/AsyncSequenceReader/AsyncReadSequence.swift

@@ -9,7 +9,7 @@
 
 /// An ``/AsyncSequenceReader/_Concurrency/AsyncSequence`` subtype suitable for reading an existing iterator in place.
 ///
-/// Note that to conform to this protocol, your type must be a reference type. After iterating, you'll also likely want to copy the base iterator back into your starting iterator, as shown in ``/AsyncSequenceReader/_Concurrency/AsyncIteratorProtocol/transform(with:readSequenceFactory:)``.
+/// Note that to conform to this protocol, your type must be a reference type. After iterating, you'll also likely want to copy the base iterator back into your starting iterator, as shown in ``AsyncBufferedIterator/transform(isolation:with:readSequenceFactory:)``.
 public protocol AsyncReadSequence: AsyncSequence, AnyObject {
     associatedtype BaseIterator: AsyncIteratorProtocol where BaseIterator.Element == Element
 
@@ -22,14 +22,16 @@ extension AsyncIteratorProtocol {
     ///
     /// - Note: Iterating over the read sequence multiple times will result in undefined behavior.
     ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter sequenceTransform: A transformation that accepts a sequence that can be read from, or stopped prematurely by returning `nil`. The receiving iterator will have moved forward by the same amount of items consumed within `sequenceTransform`.
     /// - Parameter readSequenceFactory: A factory to create a suitable ``AsyncReadSequence`` that will determine the logical bounds of the transformation within the receiving iterator.
     /// - Returns: A transformed value read from the iterator, or `nil` if there were no values left to read.
     public mutating func transform<
         Transformed, ReadSequence: AsyncReadSequence,
         TransformFailure: Error
     >(
-        with sequenceTransform: sending (sending ReadSequence) async throws(TransformFailure) -> Transformed,
+        isolation actor: isolated (any Actor)? = #isolation,
+        with sequenceTransform: sending (sending ReadSequence) async throws(TransformFailure) -> sending Transformed,
         readSequenceFactory: (inout AsyncBufferedIterator<Self>) -> ReadSequence
     ) async throws(TransformFailure) -> Transformed? where ReadSequence.BaseIterator == Self {
         var results: Transformed? = nil
@@ -50,15 +52,17 @@ extension AsyncBufferedIterator {
     /// Transform the receiving iterator using the specified sequence transformer and configured read sequence.
     ///
     /// - Note: Iterating over the read sequence multiple times will result in undefined behavior.
-    ///  
+    ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter sequenceTransform: A transformation that accepts a sequence that can be read from, or stopped prematurely by returning `nil`. The receiving iterator will have moved forward by the same amount of items consumed within `sequenceTransform`.
     /// - Parameter readSequenceFactory: A factory to create a suitable ``AsyncReadSequence`` that will determine the logical bounds of the transformation within the receiving iterator.
     /// - Returns: A transformed value read from the iterator, or `nil` if there were no values left to read.
     public mutating func transform<
         Transformed, ReadSequence: AsyncReadSequence,
         TransformFailure: Error
     >(
-        with sequenceTransform: sending (sending ReadSequence) async throws(TransformFailure) -> Transformed,
+        isolation actor: isolated (any Actor)? = #isolation,
+        with sequenceTransform: sending (sending ReadSequence) async throws(TransformFailure) -> sending Transformed,
         readSequenceFactory: (inout Self) -> ReadSequence
     ) async throws(TransformFailure) -> Transformed? where ReadSequence.BaseIterator == BaseIterator {
         var results: Transformed? = nil

Sources/AsyncSequenceReader/AsyncReadUpToCountSequence.swift

@@ -11,24 +11,35 @@ extension AsyncIteratorProtocol {
     /// Asynchronously advances by the specified number of elements, or ends the sequence if there is no next element.
     ///
     /// If a complete array could not be collected, an error is thrown and the sequence should be considered finished.
+    ///
+    /// - Parameter actor: The isolation context to run the reciever on. 
     /// - Parameter count: The number of elements to collect.
     /// - Returns: A collection with exactly `count` elements, or `nil` if the sequence is finished.
     /// - Throws: ``AsyncSequenceReaderError/insufficientElements(minimum:actual:)`` if a complete byte sequence could not be returned by the time the sequence ended.
     @inlinable
-    public mutating func collect(_ count: Int) async throws -> [Element]? {
+    public mutating func collect(
+        isolation actor: isolated (any Actor)? = #isolation,
+        _ count: Int
+    ) async throws -> [Element]? {
         assert(count >= 0, "count must be larger than or equal to 0")
         return try await collect(min: count, max: count)
     }
 
     /// Asynchronously advances by the specified minimum number of elements, continuing until the specified maximum number of elements, or ends the sequence if there is no next element.
     ///
     /// If a complete array larger than `minCount` could not be constructed, an error is thrown and the sequence should be considered finished.
+    ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter minCount: The minimum number of elements to collect.
     /// - Parameter maxCount: The maximum number of elements to collect.
     /// - Returns: A collection with at least `minCount` and at most `maxCount` elements, or `nil` if the sequence is finished.
     /// - Throws: ``AsyncSequenceReaderError/insufficientElements(minimum:actual:)`` if a complete byte sequence could not be returned by the time the sequence ended.
     @inlinable
-    public mutating func collect(min minCount: Int = 0, max maxCount: Int) async throws -> [Element]? {
+    public mutating func collect(
+        isolation actor: isolated (any Actor)? = #isolation,
+        min minCount: Int = 0,
+        max maxCount: Int
+    ) async throws -> [Element]? {
         precondition(minCount <= maxCount, "maxCount must be larger than or equal to minCount")
         precondition(minCount >= 0, "minCount must be larger than or equal to 0")
         if maxCount == 0 { return [] }
@@ -59,11 +70,16 @@ extension AsyncIteratorProtocol where Failure == Never {
     /// Asynchronously advances by the specified number of elements, or ends the sequence if there is no next element.
     ///
     /// If a complete array could not be collected, an error is thrown and the sequence should be considered finished.
+    ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter count: The number of elements to collect.
     /// - Returns: A collection with exactly `count` elements, or `nil` if the sequence is finished.
     /// - Throws: ``AsyncSequenceReaderError/insufficientElements(minimum:actual:)`` if a complete byte sequence could not be returned by the time the sequence ended.
     @inlinable
-    public mutating func collect(_ count: Int) async throws(AsyncSequenceReaderError) -> [Element]? {
+    public mutating func collect(
+        isolation actor: isolated (any Actor)? = #isolation,
+        _ count: Int
+    ) async throws(AsyncSequenceReaderError) -> [Element]? {
         assert(count >= 0, "count must be larger than or equal to 0")
         #if compiler(<6.1.3) || compiler(>=6.2)
         return try await collect(min: count, max: count)
@@ -94,12 +110,18 @@ extension AsyncIteratorProtocol where Failure == Never {
     /// Asynchronously advances by the specified minimum number of elements, continuing until the specified maximum number of elements, or ends the sequence if there is no next element.
     ///
     /// If a complete array larger than `minCount` could not be constructed, an error is thrown and the sequence should be considered finished.
+    ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter minCount: The minimum number of elements to collect.
     /// - Parameter maxCount: The maximum number of elements to collect.
     /// - Returns: A collection with at least `minCount` and at most `maxCount` elements, or `nil` if the sequence is finished.
     /// - Throws: ``AsyncSequenceReaderError/insufficientElements(minimum:actual:)`` if a complete byte sequence could not be returned by the time the sequence ended.
     @inlinable
-    public mutating func collect(min minCount: Int = 0, max maxCount: Int) async throws(AsyncSequenceReaderError) -> [Element]? {
+    public mutating func collect(
+        isolation actor: isolated (any Actor)? = #isolation,
+        min minCount: Int = 0,
+        max maxCount: Int
+    ) async throws(AsyncSequenceReaderError) -> [Element]? {
         precondition(minCount <= maxCount, "maxCount must be larger than or equal to minCount")
         precondition(minCount >= 0, "minCount must be larger than or equal to 0")
         if maxCount == 0 { return [] }
@@ -153,6 +175,7 @@ extension AsyncIteratorProtocol {
     /// // Prints: "Hello, World!", "My name is Dimitri.", "", "Bye!"
     /// ```
     ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter count: The number of elements the `sequenceTransform` closure will have access to.
     /// - Parameter sequenceTransform: A transformation that accepts a sequence of the specified size that can be read from, or stopped prematurely by returning early. The receiving iterator will have moved forward by the same amount of items consumed within `sequenceTransform`.
     /// - Returns: A transformed value as returned by `sequenceTransform`, or `nil` if the sequence was already finished.
@@ -162,8 +185,9 @@ extension AsyncIteratorProtocol {
         Transformed,
         TransformFailure
     >(
+        isolation actor: isolated (any Actor)? = #isolation,
         _ count: Int,
-        sequenceTransform: sending (sending AsyncReadUpToCountSequence<Self>) async throws(TransformFailure) -> Transformed
+        sequenceTransform: sending (sending AsyncReadUpToCountSequence<Self>) async throws(TransformFailure) -> sending Transformed
     ) async throws(TransformFailure) -> Transformed? {
         assert(count >= 0, "count must be larger than or equal to 0")
         return try await collect(min: count, max: count, sequenceTransform: sequenceTransform)
@@ -198,6 +222,7 @@ extension AsyncIteratorProtocol {
     ///
     /// - Important: This variation reads ahead a single byte
     ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter minCount: The minimum number of elements the `sequenceTransform` closure will attempt have access to. If this number cannot be guaranteed, an error will be thrown.
     /// - Parameter maxCount: The maximum number of elements the `sequenceTransform` closure will have access to.
     /// - Parameter sequenceTransform: A transformation that accepts a sequence of the specified size that can be read from, or stopped prematurely by returning early. The receiving iterator will have moved forward by the same amount of items consumed within `sequenceTransform`.
@@ -208,9 +233,10 @@ extension AsyncIteratorProtocol {
         Transformed,
         TransformFailure
     >(
+        isolation actor: isolated (any Actor)? = #isolation,
         min minCount: Int = 1,
         max maxCount: Int,
-        sequenceTransform: sending (sending AsyncReadUpToCountSequence<Self>) async throws(TransformFailure) -> Transformed
+        sequenceTransform: sending (sending AsyncReadUpToCountSequence<Self>) async throws(TransformFailure) -> sending Transformed
     ) async throws(TransformFailure) -> Transformed? {
         /// It is unsafe to read ahead in this case, so exit early if we know we won't need to read.
         if maxCount == 0 { return nil }
@@ -247,6 +273,7 @@ extension AsyncBufferedIterator {
     /// // Prints: "Hello, World!", "My name is Dimitri.", "", "Bye!"
     /// ```
     ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter count: The number of elements the `sequenceTransform` closure will have access to.
     /// - Parameter sequenceTransform: A transformation that accepts a sequence of the specified size that can be read from, or stopped prematurely by returning early. The receiving iterator will have moved forward by the same amount of items consumed within `sequenceTransform`.
     /// - Returns: A transformed value as returned by `sequenceTransform`, or `nil` if the sequence was already finished.
@@ -256,8 +283,9 @@ extension AsyncBufferedIterator {
         Transformed,
         TransformFailure
     >(
+        isolation actor: isolated (any Actor)? = #isolation,
         _ count: Int,
-        sequenceTransform: sending (sending AsyncReadUpToCountSequence<BaseIterator>) async throws(TransformFailure) -> Transformed
+        sequenceTransform: sending (sending AsyncReadUpToCountSequence<BaseIterator>) async throws(TransformFailure) -> sending Transformed
     ) async throws(TransformFailure) -> Transformed? {
         assert(count >= 0, "count must be larger than 0")
         return try await collect(min: count, max: count, sequenceTransform: sequenceTransform)
@@ -290,6 +318,7 @@ extension AsyncBufferedIterator {
     /// // Prints: "Hello, World!", "My name is Dimitri.", "", "Bye?"
     /// ```
     ///
+    /// - Parameter actor: The isolation context to run the reciever on.
     /// - Parameter minCount: The minimum number of elements the `sequenceTransform` closure will attempt have access to. If this number cannot be guaranteed, an error will be thrown.
     /// - Parameter maxCount: The maximum number of elements the `sequenceTransform` closure will have access to.
     /// - Parameter sequenceTransform: A transformation that accepts a sequence of the specified size that can be read from, or stopped prematurely by returning early. The receiving iterator will have moved forward by the same amount of items consumed within `sequenceTransform`.
@@ -300,9 +329,10 @@ extension AsyncBufferedIterator {
         Transformed,
         TransformFailure: Error
     >(
+        isolation actor: isolated (any Actor)? = #isolation,
         min minCount: Int = 0,
         max maxCount: Int,
-        sequenceTransform: sending (sending AsyncReadUpToCountSequence<BaseIterator>) async throws(TransformFailure) -> Transformed
+        sequenceTransform: sending (sending AsyncReadUpToCountSequence<BaseIterator>) async throws(TransformFailure) -> sending Transformed
     ) async throws(TransformFailure) -> Transformed? {
         try await transform(with: sequenceTransform) { .init($0, minCount: minCount, maxCount: maxCount) }
     }
@@ -358,7 +388,16 @@ extension AsyncReadUpToCountSequence: AsyncSequence {
         ///
         /// This iterator checks if `numberOfElementsRead` has exceeded the max size for the sequence. If it has not, then it'll read until it does. If the next value read marks the end of the sequence, but the minimum size has not yet been reached, an error is thrown.
         @inlinable
-        public mutating func next() async throws -> Element? {
+        @_disfavoredOverload
+        public mutating func next() async rethrows -> Element? {
+            try await next()
+        }
+
+        /// Produces the next element in the sequence.
+        ///
+        /// This iterator checks if `numberOfElementsRead` has exceeded the max size for the sequence. If it has not, then it'll read until it does. If the next value read marks the end of the sequence, but the minimum size has not yet been reached, an error is thrown.
+        @inlinable
+        public mutating func next(isolation actor: isolated (any Actor)? = #isolation) async throws -> Element? {
             guard numberOfElementsRead < readSequence.maxCount else { return nil }
             guard let next = try await readSequence.baseIterator.next() else {