Removed executing: label from .run and made CollectedResults.standardOutput non-optional with fatalError

Author: iCharlesHu

Sources/FoundationEssentials/Subprocess/Platforms/Subprocess+Unix.swift

@@ -281,7 +281,7 @@ internal func monitorProcessTermination(
         source.setEventHandler {
             source.cancel()
             var status: Int32 = -1
-            waitpid(pid.value, &status, WNOHANG)
+            waitpid(pid.value, &status, 0)
             if _was_process_exited(status) != 0 {
                 continuation.resume(returning: .exit(_get_exit_code(status)))
                 return

Sources/FoundationEssentials/Subprocess/Subprocess+API.swift

@@ -13,7 +13,7 @@ import SystemPackage
 
 extension Subprocess {
     public static func run(
-        executing executable: Executable,
+        _ executable: Executable,
         arguments: Arguments = [],
         environment: Environment = .inherit,
         workingDirectory: FilePath? = nil,
@@ -23,7 +23,7 @@ extension Subprocess {
         error: CollectedOutputMethod = .collect
     ) async throws -> CollectedResult {
         let result = try await self.run(
-            executing: executable,
+            executable,
             arguments: arguments,
             environment: environment,
             workingDirectory: workingDirectory,
@@ -48,7 +48,7 @@ extension Subprocess {
     }
 
     public static func run(
-        executing executable: Executable,
+        _ executable: Executable,
         arguments: Arguments = [],
         environment: Environment = .inherit,
         workingDirectory: FilePath? = nil,
@@ -58,7 +58,7 @@ extension Subprocess {
         error: CollectedOutputMethod = .collect
     ) async throws -> CollectedResult {
         let result = try await self.run(
-            executing: executable,
+            executable,
             arguments: arguments,
             environment: environment,
             workingDirectory: workingDirectory,
@@ -84,7 +84,7 @@ extension Subprocess {
     }
 
     public static func run<S: AsyncSequence>(
-        executing executable: Executable,
+        _ executable: Executable,
         arguments: Arguments = [],
         environment: Environment = .inherit,
         workingDirectory: FilePath? = nil,
@@ -94,7 +94,7 @@ extension Subprocess {
         error: CollectedOutputMethod = .collect
     ) async throws -> CollectedResult where S.Element == UInt8 {
         let result =  try await self.run(
-            executing: executable,
+            executable,
             arguments: arguments,
             environment: environment,
             workingDirectory: workingDirectory,
@@ -123,7 +123,7 @@ extension Subprocess {
 // MARK: Custom Execution Body
 extension Subprocess {
     public static func run<R>(
-        executing executable: Executable,
+        _ executable: Executable,
         arguments: Arguments = [],
         environment: Environment = .inherit,
         workingDirectory: FilePath? = nil,
@@ -144,7 +144,7 @@ extension Subprocess {
     }
 
     public static func run<R>(
-        executing executable: Executable,
+        _ executable: Executable,
         arguments: Arguments = [],
         environment: Environment = .inherit,
         workingDirectory: FilePath? = nil,
@@ -169,7 +169,7 @@ extension Subprocess {
     }
 
     public static func run<R, S: AsyncSequence>(
-        executing executable: Executable,
+        _ executable: Executable,
         arguments: Arguments = [],
         environment: Environment = .inherit,
         workingDirectory: FilePath? = nil,
@@ -194,7 +194,7 @@ extension Subprocess {
     }
 
     public static func run<R>(
-        executing executable: Executable,
+        _ executable: Executable,
         arguments: Arguments = [],
         environment: Environment = .inherit,
         workingDirectory: FilePath? = nil,

Sources/FoundationEssentials/Subprocess/Subprocess.swift

@@ -44,7 +44,7 @@ public struct Subprocess: Sendable {
             fatalError("The standard output was not redirected")
         }
         guard let fd = fd else {
-            fatalError("The standard output has already been consumed")
+            fatalError("The standard output has already been closed")
         }
         return AsyncBytes(fileDescriptor: fd)
     }
@@ -55,7 +55,7 @@ public struct Subprocess: Sendable {
             fatalError("The standard error was not redirected")
         }
         guard let fd = fd else {
-            fatalError("The standard error has already been consumed")
+            fatalError("The standard error has already been closed")
         }
         return AsyncBytes(fileDescriptor: fd)
     }
@@ -114,8 +114,20 @@ extension Subprocess {
     public struct CollectedResult: Sendable, Hashable {
         public let processIdentifier: ProcessIdentifier
         public let terminationStatus: TerminationStatus
-        public let standardOutput: Data?
-        public let standardError: Data?
+        private let _standardOutput: Data?
+        private let _standardError: Data?
+        public var standardOutput: Data {
+            guard let output = self._standardOutput else {
+                fatalError("standardOutput is only available if the Subprocess was ran with .collect as output")
+            }
+            return output
+        }
+        public var standardError: Data {
+            guard let output = self._standardError else {
+                fatalError("standardError is only available if the Subprocess was ran with .collect as error ")
+            }
+            return output
+        }
 
         internal init(
             processIdentifier: ProcessIdentifier,
@@ -124,8 +136,8 @@ extension Subprocess {
             standardError: Data?) {
             self.processIdentifier = processIdentifier
             self.terminationStatus = terminationStatus
-            self.standardOutput = standardOutput
-            self.standardError = standardError
+            self._standardOutput = standardOutput
+            self._standardError = standardError
         }
     }
 }
@@ -144,7 +156,7 @@ extension Subprocess {
     private func capture(fileDescriptor: FileDescriptor, maxLength: Int) async throws -> Data{
         let chunkSize: Int = min(Subprocess.readBufferSize, maxLength)
         var buffer: [UInt8] = []
-        while buffer.count < maxLength {
+        while buffer.count <= maxLength {
             let captured = try await fileDescriptor.read(upToLength: chunkSize)
             buffer += captured
             if captured.count < chunkSize {
@@ -171,7 +183,7 @@ extension Subprocess {
         }
         return try await self.capture(fileDescriptor: readFd, maxLength: limit)
     }
-    
+
     internal func captureIOs() async throws -> (standardOut: Data?, standardError: Data?) {
         return try await withThrowingTaskGroup(of: OutputCapturingState.self) { group in
             group.addTask {

Tests/FoundationEssentialsTests/SubprocessTests.swift

@@ -19,24 +19,40 @@
 import TestSupport
 #endif
 
+import SystemPackage
+
 final class SubprocessTests: XCTestCase {
     func testSimple() async throws {
-        let ls = try await Subprocess.run(executing: .named("ls"), output: .collect, error: .discard)
-        let result = String(data: ls.standardOutput!, encoding: .utf8)!
+        let ls = try await Subprocess.run(.named("ls"))
+        let result = String(data: ls.standardOutput, encoding: .utf8)!
         XCTAssert(ls.terminationStatus.isSuccess)
         XCTAssert(!result.isEmpty)
         print(result)
     }
-    
+
+    func testInteractive() async throws {
+        let su = try await Subprocess.run(.at("/Users/icharleshu/Developer/super.sh"), input: .readFrom(.standardInput, closeWhenDone: false))
+        XCTAssert(!su.terminationStatus.isSuccess)
+    }
+
+    func testChained() async throws {
+        let (readFd, writeFd) = try FileDescriptor.pipe()
+        try await Subprocess.run(
+            .named("ls"), output: .writeTo(writeFd, closeWhenDone: true), error: .discard)
+        let grep = try await Subprocess.run(.named("grep"), arguments: ["com"], input: .readFrom(readFd, closeWhenDone: true))
+        var output = String(data: grep.standardOutput, encoding: .utf8) ?? "Failed to decode"
+        print("Output: \(output)")
+    }
+
     func testLongText() async throws {
         let cat = try await Subprocess.run(
-            executing: .named("cat"),
+            .named("cat"),
             arguments: ["/Users/icharleshu/Downloads/PaP.txt"],
             output: .collect(limit: 1024 * 1024),
             error: .discard
         )
         print("after")
-        print("Result: \(cat.standardOutput?.count ?? -1)")
+        print("Result: \(cat.standardOutput.count)")
     }
 
     func testComplex() async throws {
@@ -45,7 +61,7 @@ final class SubprocessTests: XCTestCase {
         }
 
         let result = try await Subprocess.run(
-            executing: .named("curl"),
+            .named("curl"),
             arguments: ["http://ip.jsontest.com/"]
         ) { execution in
             let output: [UInt8] = try await Array(execution.standardOutput)
@@ -57,8 +73,10 @@ final class SubprocessTests: XCTestCase {
     }
 
     func testShell() async throws {
-        let result = try await Subprocess.run(executing: .named("sh")) { subprocess, writer in
-            // Stream all outputs
+        let result = try await Subprocess.run(.named("sh")) { subprocess, writer in
+            try await writer.write("ls | grep 'apple'\n".utf8)
+            try await writer.finish()
+            // Stream outputs
             try await withThrowingTaskGroup(of: Void.self) { group in
                 group.addTask {
                     // Stream output line by line
@@ -67,10 +85,6 @@ final class SubprocessTests: XCTestCase {
                         print("> \(line)")
                     }
                 }
-                group.addTask {
-                    try await writer.write("ls\n".utf8)
-                    try await writer.finish()
-                }
                 try await group.waitForAll()
             }
         }