Move stream creation and configure to platform specific files

Author: Ryan Dingman

Sources/Subprocess/AsyncBufferSequence.swift

@@ -29,19 +29,17 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
         public typealias Element = SequenceOutput.Buffer
 
         private let diskIO: TrackedPlatformDiskIO
-        private let bufferSize: Int
         private var buffer: [UInt8]
         private var currentPosition: Int
         private var finished: Bool
-        private var streamIterator: AsyncThrowingStream<StreamStatus, Swift.Error>.AsyncIterator
+        private var streamIterator: AsyncThrowingStream<TrackedPlatformDiskIO.StreamStatus, Swift.Error>.AsyncIterator
 
-        internal init(diskIO: TrackedPlatformDiskIO, bufferSize: Int) {
+        internal init(diskIO: TrackedPlatformDiskIO) {
             self.diskIO = diskIO
-            self.bufferSize = bufferSize
             self.buffer = []
             self.currentPosition = 0
             self.finished = false
-            self.streamIterator = Self.createDataStream(with: diskIO.dispatchIO, bufferSize: bufferSize).makeAsyncIterator()
+            self.streamIterator = diskIO.readDataStream(upToLength: readBufferSize).makeAsyncIterator()
         }
 
         public mutating func next() async throws -> SequenceOutput.Buffer? {
@@ -51,7 +49,7 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
                     return data
 
                 case .endOfStream(let data):
-                    streamIterator = Self.createDataStream(with: diskIO.dispatchIO, bufferSize: bufferSize).makeAsyncIterator()
+                    streamIterator = diskIO.readDataStream(upToLength: readBufferSize).makeAsyncIterator()
                     return data
 
                 case .endOfFile:
@@ -63,66 +61,24 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
                 return nil
             }
         }
-
-        private enum StreamStatus {
-            case data(SequenceOutput.Buffer)
-            case endOfStream(SequenceOutput.Buffer)
-            case endOfFile
-        }
-
-        private static func createDataStream(with dispatchIO: DispatchIO, bufferSize: Int) -> AsyncThrowingStream<StreamStatus, Swift.Error> {
-            return AsyncThrowingStream<StreamStatus, Swift.Error> { continuation in
-                dispatchIO.read(
-                    offset: 0,
-                    length: bufferSize,
-                    queue: .global()
-                ) { done, data, error in
-                    if error != 0 {
-                        continuation.finish(throwing: SubprocessError(
-                            code: .init(.failedToReadFromSubprocess),
-                            underlyingError: .init(rawValue: error)
-                        ))
-                        return
-                    }
-
-                    // Treat empty data and nil as the same
-                    let buffer = data.map { $0.isEmpty ? nil : $0 } ?? nil
-                    let status: StreamStatus
-
-                    switch (buffer, done) {
-                    case (.some(let data), false):
-                        status = .data(SequenceOutput.Buffer(data: data))
-
-                    case (.some(let data), true):
-                        status = .endOfStream(SequenceOutput.Buffer(data: data))
-
-                    case (nil, false):
-                        return
-
-                    case (nil, true):
-                        status = .endOfFile
-                    }
-
-                    continuation.yield(status)
-
-                    if done {
-                        continuation.finish()
-                    }
-                }
-            }
-        }
     }
 
     private let diskIO: TrackedPlatformDiskIO
-    private let bufferSize: Int
 
-    internal init(diskIO: TrackedPlatformDiskIO, bufferSize: Int) {
+    internal init(diskIO: TrackedPlatformDiskIO) {
         self.diskIO = diskIO
-        self.bufferSize = bufferSize
     }
 
     public func makeAsyncIterator() -> Iterator {
-        return Iterator(diskIO: self.diskIO, bufferSize: bufferSize)
+        return Iterator(diskIO: self.diskIO)
+    }
+}
+
+extension TrackedPlatformDiskIO {
+    internal enum StreamStatus {
+        case data(SequenceOutput.Buffer)
+        case endOfStream(SequenceOutput.Buffer)
+        case endOfFile
     }
 }
 

Sources/Subprocess/Execution.swift

@@ -110,15 +110,8 @@ extension Execution where Output == SequenceOutput {
             fatalError("The standard output has already been consumed")
         }
 
-        if let lowWater = output.lowWater {
-            readFd.dispatchIO.setLimit(lowWater: lowWater)
-        }
-
-        if let highWater = output.highWater {
-            readFd.dispatchIO.setLimit(highWater: highWater)
-        }
-
-        return AsyncBufferSequence(diskIO: readFd, bufferSize: output.bufferSize)
+        // TODO: Make buffer size an option
+        return AsyncBufferSequence(diskIO: readFd)
     }
 }
 
@@ -142,15 +135,8 @@ extension Execution where Error == SequenceOutput {
             fatalError("The standard output has already been consumed")
         }
 
-        if let lowWater = error.lowWater {
-            readFd.dispatchIO.setLimit(lowWater: lowWater)
-        }
-
-        if let highWater = error.highWater {
-            readFd.dispatchIO.setLimit(highWater: highWater)
-        }
-
-        return AsyncBufferSequence(diskIO: readFd, bufferSize: error.bufferSize)
+        // TODO: Make buffer size and option
+        return AsyncBufferSequence(diskIO: readFd)
     }
 }
 

Sources/Subprocess/IO/Output.swift

@@ -210,15 +210,8 @@ public struct BytesOutput: OutputProtocol {
 #endif
 public struct SequenceOutput: OutputProtocol {
     public typealias OutputType = Void
-    internal let lowWater: Int?
-    internal let highWater: Int?
-    internal let bufferSize: Int
-
-    internal init(lowWater: Int? = nil, highWater: Int? = nil, bufferSize: Int = readBufferSize) {
-        self.lowWater = lowWater
-        self.highWater = highWater
-        self.bufferSize = bufferSize
-    }
+
+    internal init() {}
 }
 
 #if SubprocessSpan
@@ -291,9 +284,6 @@ extension OutputProtocol where Self == SequenceOutput {
     /// to the `.standardOutput` (or `.standardError`) property
     /// of `Execution` as `AsyncSequence<Data>`.
     public static var sequence: Self { .init() }
-    public static func sequence(lowWater: Int? = nil, highWater: Int? = nil, bufferSize: Int? = nil) -> Self {
-        .init(lowWater: lowWater, highWater: highWater, bufferSize: bufferSize ?? readBufferSize)
-    }
 }
 
 // MARK: - Span Default Implementations

Sources/Subprocess/Platforms/Subprocess+Darwin.swift

@@ -57,6 +57,8 @@ public struct PlatformOptions: Sendable {
     /// Creates a session and sets the process group ID
     /// i.e. Detach from the terminal.
     public var createSession: Bool = false
+    public var outputOptions: StreamOptions = .init()
+    public var errorOptions: StreamOptions = .init()
     /// An ordered list of steps in order to tear down the child
     /// process in case the parent task is cancelled before
     /// the child proces terminates.
@@ -126,6 +128,18 @@ extension PlatformOptions {
     #endif
 }
 
+extension PlatformOptions {
+    public struct StreamOptions: Sendable {
+        let lowWater: Int?
+        let highWater: Int?
+
+        init(lowWater: Int? = nil, highWater: Int? = nil) {
+            self.lowWater = lowWater
+            self.highWater = highWater
+        }
+    }
+}
+
 extension PlatformOptions: CustomStringConvertible, CustomDebugStringConvertible {
     internal func description(withIndent indent: Int) -> String {
         let indent = String(repeating: " ", count: indent * 4)
@@ -355,8 +369,8 @@ extension Configuration {
                     output: output,
                     error: error,
                     inputPipe: inputPipe.createInputPipe(),
-                    outputPipe: outputPipe.createOutputPipe(),
-                    errorPipe: errorPipe.createOutputPipe()
+                    outputPipe: outputPipe.createOutputPipe(with: platformOptions.outputOptions),
+                    errorPipe: errorPipe.createOutputPipe(with: platformOptions.errorOptions)
                 )
             }
 

Sources/Subprocess/Platforms/Subprocess+Linux.swift

@@ -127,8 +127,8 @@ extension Configuration {
                     output: output,
                     error: error,
                     inputPipe: inputPipe.createInputPipe(),
-                    outputPipe: outputPipe.createOutputPipe(),
-                    errorPipe: errorPipe.createOutputPipe()
+                    outputPipe: outputPipe.createOutputPipe(with: platformOptions.outputOptions),
+                    errorPipe: errorPipe.createOutputPipe(with: platformOptions.errorOptions)
                 )
             }
 
@@ -176,6 +176,8 @@ public struct PlatformOptions: Sendable {
     // Creates a session and sets the process group ID
     // i.e. Detach from the terminal.
     public var createSession: Bool = false
+    public var outputOptions: StreamOptions = .init()
+    public var errorOptions: StreamOptions = .init()
     /// An ordered list of steps in order to tear down the child
     /// process in case the parent task is cancelled before
     /// the child proces terminates.
@@ -197,6 +199,18 @@ public struct PlatformOptions: Sendable {
     public init() {}
 }
 
+extension PlatformOptions {
+    public struct StreamOptions: Sendable {
+        let lowWater: Int?
+        let highWater: Int?
+
+        init(lowWater: Int? = nil, highWater: Int? = nil) {
+            self.lowWater = lowWater
+            self.highWater = highWater
+        }
+    }
+}
+
 extension PlatformOptions: CustomStringConvertible, CustomDebugStringConvertible {
     internal func description(withIndent indent: Int) -> String {
         let indent = String(repeating: " ", count: indent * 4)

Sources/Subprocess/Platforms/Subprocess+Unix.swift

@@ -412,6 +412,7 @@ extension CreatedPipe {
                     }
                 }
             )
+
             writeEnd = .init(
                 dispatchIO,
                 closeWhenDone: writeFileDescriptor.closeWhenDone
@@ -423,7 +424,7 @@ extension CreatedPipe {
         )
     }
 
-    internal func createOutputPipe() -> OutputPipe {
+    internal func createOutputPipe(with options: PlatformOptions.StreamOptions) -> OutputPipe {
         var readEnd: TrackedPlatformDiskIO? = nil
         if let readFileDescriptor = self.readFileDescriptor {
             let dispatchIO: DispatchIO = DispatchIO(
@@ -437,6 +438,15 @@ extension CreatedPipe {
                     }
                 }
             )
+
+            if let lowWater = options.lowWater {
+                dispatchIO.setLimit(lowWater: lowWater)
+            }
+
+            if let highWater = options.highWater {
+                dispatchIO.setLimit(highWater: highWater)
+            }
+
             readEnd = .init(
                 dispatchIO,
                 closeWhenDone: readFileDescriptor.closeWhenDone
@@ -451,6 +461,48 @@ extension CreatedPipe {
 
 // MARK: - TrackedDispatchIO extensions
 extension TrackedDispatchIO {
+    internal func readDataStream(upToLength maxLength: Int) -> AsyncThrowingStream<StreamStatus, Swift.Error> {
+        return AsyncThrowingStream<StreamStatus, Swift.Error> { continuation in
+            self.dispatchIO.read(
+                offset: 0,
+                length: maxLength,
+                queue: .global()
+            ) { done, data, error in
+                if error != 0 {
+                    continuation.finish(throwing: SubprocessError(
+                        code: .init(.failedToReadFromSubprocess),
+                        underlyingError: .init(rawValue: error)
+                    ))
+                    return
+                }
+
+                // Treat empty data and nil as the same
+                let buffer = data.map { $0.isEmpty ? nil : $0 } ?? nil
+                let status: StreamStatus
+
+                switch (buffer, done) {
+                case (.some(let data), false):
+                    status = .data(SequenceOutput.Buffer(data: data))
+
+                case (.some(let data), true):
+                    status = .endOfStream(SequenceOutput.Buffer(data: data))
+
+                case (nil, false):
+                    return
+
+                case (nil, true):
+                    status = .endOfFile
+                }
+
+                continuation.yield(status)
+
+                if done {
+                    continuation.finish()
+                }
+            }
+        }
+    }
+
 #if SubprocessSpan
     @available(SubprocessSpan, *)
 #endif