Remove definition and use of StreamOptions on Windows

Author: rdingman

Sources/Subprocess/API.swift

@@ -200,7 +200,7 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
             }
 
             // Body runs in the same isolation
-            let outputSequence = AsyncBufferSequence(diskIO: outputIOBox.take()!.consumeDiskIO(), streamOptions: platformOptions.streamOptions)
+            let outputSequence = AsyncBufferSequence(diskIO: outputIOBox.take()!.consumeDiskIO())
             let result = try await body(execution, outputSequence)
             try await group.waitForAll()
             return result
@@ -250,7 +250,7 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
             }
 
             // Body runs in the same isolation
-            let errorSequence = AsyncBufferSequence(diskIO: errorIOBox.take()!.consumeDiskIO(), streamOptions: platformOptions.streamOptions)
+            let errorSequence = AsyncBufferSequence(diskIO: errorIOBox.take()!.consumeDiskIO())
             let result = try await body(execution, errorSequence)
             try await group.waitForAll()
             return result
@@ -286,7 +286,7 @@ public func run<Result, Error: OutputProtocol>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO(), streamOptions: platformOptions.streamOptions)
+        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO())
         return try await body(execution, writer, outputSequence)
     }
 }
@@ -319,7 +319,7 @@ public func run<Result, Output: OutputProtocol>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO(), streamOptions: platformOptions.streamOptions)
+        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO())
         return try await body(execution, writer, errorSequence)
     }
 }
@@ -376,8 +376,8 @@ public func run<Result>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO(), streamOptions: platformOptions.streamOptions)
-        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO(), streamOptions: platformOptions.streamOptions)
+        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO())
+        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO())
         return try await body(execution, writer, outputSequence, errorSequence)
     }
 }
@@ -503,8 +503,8 @@ public func run<Result>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO(), streamOptions: configuration.platformOptions.streamOptions)
-        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO(), streamOptions: configuration.platformOptions.streamOptions)
+        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO())
+        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO())
         return try await body(execution, writer, outputSequence, errorSequence)
     }
 }

Sources/Subprocess/AsyncBufferSequence.swift

@@ -45,7 +45,7 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
         private let continuation: Stream.Continuation
         private var bytesRemaining: Int
 
-        internal init(diskIO: DiskIO, streamOptions: PlatformOptions.StreamOptions) {
+        internal init(diskIO: DiskIO) {
             self.diskIO = diskIO
             self.buffer = []
             self.currentPosition = 0
@@ -54,16 +54,6 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
             self.streamIterator = stream.makeAsyncIterator()
             self.continuation = continuation
             self.bytesRemaining = 0
-
-            #if !os(Windows)
-            if let minimumBufferSize = streamOptions.minimumBufferSize {
-                diskIO.setLimit(lowWater: minimumBufferSize)
-            }
-
-            if let maximumBufferSize = streamOptions.maximumBufferSize {
-                diskIO.setLimit(highWater: maximumBufferSize)
-            }
-            #endif
         }
 
         public mutating func next() async throws -> Buffer? {
@@ -103,15 +93,13 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
     }
 
     private let diskIO: DiskIO
-    private let streamOptions: PlatformOptions.StreamOptions
 
-    internal init(diskIO: DiskIO, streamOptions: PlatformOptions.StreamOptions) {
+    internal init(diskIO: DiskIO) {
         self.diskIO = diskIO
-        self.streamOptions = streamOptions
     }
 
     public func makeAsyncIterator() -> Iterator {
-        return Iterator(diskIO: self.diskIO, streamOptions: streamOptions)
+        return Iterator(diskIO: self.diskIO)
     }
 }
 

Sources/Subprocess/Platforms/Subprocess+Darwin.swift

@@ -456,9 +456,9 @@ extension Configuration {
                 )
                 return SpawnResult(
                     execution: execution,
-                    inputWriteEnd: inputWriteFileDescriptor?.createPlatformDiskIO(),
-                    outputReadEnd: outputReadFileDescriptor?.createPlatformDiskIO(),
-                    errorReadEnd: errorReadFileDescriptor?.createPlatformDiskIO()
+                    inputWriteEnd: inputWriteFileDescriptor?.createPlatformDiskIO(with: platformOptions.streamOptions),
+                    outputReadEnd: outputReadFileDescriptor?.createPlatformDiskIO(with: platformOptions.streamOptions),
+                    errorReadEnd: errorReadFileDescriptor?.createPlatformDiskIO(with: platformOptions.streamOptions)
                 )
             }
 

Sources/Subprocess/Platforms/Subprocess+Unix.swift

@@ -393,7 +393,7 @@ internal typealias PlatformFileDescriptor = CInt
 internal typealias TrackedPlatformDiskIO = TrackedDispatchIO
 
 extension TrackedFileDescriptor {
-    internal consuming func createPlatformDiskIO() -> TrackedPlatformDiskIO {
+    internal consuming func createPlatformDiskIO(with streamOptions: PlatformOptions.StreamOptions) -> TrackedPlatformDiskIO {
         let dispatchIO: DispatchIO = DispatchIO(
             type: .stream,
             fileDescriptor: self.platformDescriptor(),
@@ -406,6 +406,14 @@ extension TrackedFileDescriptor {
             }
         )
 
+        if let minimumBufferSize = streamOptions.minimumBufferSize {
+            dispatchIO.setLimit(lowWater: minimumBufferSize)
+        }
+
+        if let maximumBufferSize = streamOptions.maximumBufferSize {
+            dispatchIO.setLimit(highWater: maximumBufferSize)
+        }
+
         return .init(dispatchIO, closeWhenDone: self.closeWhenDone)
     }
 }

Sources/Subprocess/Platforms/Subprocess+Windows.swift

@@ -417,7 +417,6 @@ public struct PlatformOptions: Sendable {
     /// process in case the parent task is cancelled before
     /// the child proces terminates.
     /// Always ends in forcefully terminate at the end.
-    internal var streamOptions: StreamOptions = .init()
     public var teardownSequence: [TeardownStep] = []
     /// A closure to configure platform-specific
     /// spawning constructs. This closure enables direct
@@ -442,12 +441,6 @@ public struct PlatformOptions: Sendable {
     public init() {}
 }
 
-extension PlatformOptions {
-    internal struct StreamOptions: Sendable {
-        internal init() {}
-    }
-}
-
 extension PlatformOptions: CustomStringConvertible, CustomDebugStringConvertible {
     internal func description(withIndent indent: Int) -> String {
         let indent = String(repeating: " ", count: indent * 4)