Refactor runPipe to use a task group for all asynchronous tasks

Change runPipe into another variant of run using variadic parameters for the configuration

Author: cmcgee1024

Sources/Subprocess/API.swift

@@ -557,8 +557,8 @@ public func run<
     )
 }
 
-public enum PipeConfigurationError: Error {
-    case emptyConfiguration
+public enum PipelineError: Error {
+    case runErrors([Swift.Error])
 }
 
 /// Run a series of `Configuration` asynchronously and returns
@@ -567,88 +567,77 @@ public enum PipeConfigurationError: Error {
 /// to standard output.
 ///
 /// - Parameters:
-///   - configurations: The `Subprocess` configuration to run in a pipeline.
+///   - configurations: The `Subprocess` configurations for the start of the pipeline
+///   - last: The final configuration that produces the output
 ///   - input: The input to send to the first executable.
 ///   - output: The method to use for redirecting the standard output of the last executable.
 ///   - error: The method to use for redirecting the standard error of all executables.
 /// - Returns a CollectedResult containing the result of the pipeline and the termination status of the final executable.
-public func runPipe<
-    Input: InputProtocol,
-    Output: OutputProtocol,
-    Error: OutputProtocol
->(
-    _ pipeline: Configuration...,
-    input: Input = .none,
-    output: Output,
-    error: Error = .discarded
-) async throws -> CollectedResult<Output, Error> {
-    try await runPipe(pipeline, input: input, output: output, error: error)
-}
-
-/// Run a series of `Configuration` asynchronously and returns
-/// a `CollectedResult` collecting the result of the final process at the
-/// end of a pipeline of all processes joined together with pipes from standard input
-/// to standard output.
-///
-/// - Parameters:
-///   - configurations: The `Subprocess` configuration to run in a pipeline.
-///   - input: The input to send to the first executable.
-///   - output: The method to use for redirecting the standard output of the last executable.
-///   - error: The method to use for redirecting the standard error of all executables.
-/// - Returns a CollectedResult containing the result of the pipeline and the termination status of the final executable.
-public func runPipe<
-    Input: InputProtocol,
-    Output: OutputProtocol,
-    Error: OutputProtocol
->(
-    _ pipeline: [Configuration],
-    input: Input = .none,
-    output: Output,
-    error: Error = .discarded
-) async throws -> CollectedResult<Output, Error> {
-    guard !pipeline.isEmpty else {
-        throw PipeConfigurationError.emptyConfiguration
+func run<Input: InputProtocol, Output, Error: OutputProtocol>(_ first: Configuration, _ pipeline: Configuration..., to last: Configuration, input: Input = .none, output: Output, error: Error = .discarded) async throws -> CollectedResult<Output, Error> {
+    var pipes = [try FileDescriptor.pipe()]
+    for _ in pipeline {
+        pipes.append(try FileDescriptor.pipe())
     }
 
-    var prevInput: InputProtocol = input
-
-    for (idx, cmd) in pipeline.enumerated() {
-        let currentInput = prevInput
-        let currentOutput: any OutputProtocol
+    let finalPipes = pipes
 
-        if pipeline.count == 1 {
-            currentOutput = output
-        } else if idx != pipeline.count - 1 {
-            let (reader, writer) = try FileDescriptor.pipe()
-            prevInput = .fileDescriptor(reader, closeAfterSpawningProcess: true)
-            currentOutput = .fileDescriptor(writer, closeAfterSpawningProcess: true)
-        } else {
-            currentOutput = output
+    let result = await withTaskGroup { taskGroup -> Result<CollectedResult<Output,Error>,PipelineError> in
+        taskGroup.addTask { () -> Result<CollectedResult<Output,Error>?,Swift.Error> in
+            do {
+                _ = try await run(first, input: input, output: .fileDescriptor(finalPipes.first!.writeEnd, closeAfterSpawningProcess: true), error: error)
+                return Result.success(nil)
+            } catch {
+                return Result.failure(error)
+            }
         }
 
-        if idx == pipeline.count - 1 {
-            let result = await Task {
-                try await run(cmd, input: currentInput, output: output, error: error)
-            }.result
-
-            return try result.get()
-        } else {
-            Task {
-                if let currentOutput = currentOutput as? FileDescriptorOutput {
-                    _ = try await run(cmd, input: currentInput, output: currentOutput, error: error)
-                } else {
-                    // In this case the task is guaranteed to not be the final configuration,
-                    // and therefore the output must be a file descriptor output from a pipe that was
-                    // created in the previous run through the loop.
-                    fatalError()
+        for (idx, p) in pipeline.enumerated() {
+            taskGroup.addTask { () -> Result<CollectedResult<Output,Error>?,Swift.Error> in
+                do {
+                    _ = try await run(p, input: .fileDescriptor(finalPipes[idx+1].readEnd, closeAfterSpawningProcess: true), output: .fileDescriptor(finalPipes[idx].writeEnd, closeAfterSpawningProcess: true), error: error)
+                    return Result.success(nil)
+                } catch {
+                    return Result.failure(error)
                 }
             }
         }
+
+        taskGroup.addTask { () -> Result<CollectedResult<Output,Error>?,Swift.Error> in
+            do {
+                let collectedResult = try await run(last, input: .fileDescriptor(finalPipes.last!.readEnd, closeAfterSpawningProcess: true), output: output, error: error)
+                return Result.success(collectedResult)
+            } catch {
+                return Result.failure(error)
+            }
+        }
+
+        var errors: [Swift.Error] = []
+        var collection: CollectedResult<Output,Error>?
+
+        while let taskVal = await taskGroup.next() {
+            switch taskVal {
+                case .failure(let error):
+                    errors.append(error)
+                case .success(.some(let c)):
+                    collection = c
+                case .success(.none):
+                    continue
+            }
+        }
+
+        if errors.isEmpty {
+            return Result.failure(PipelineError.runErrors(errors))
+        }
+
+        return Result.success(collection!)
     }
 
-    // This should never happen because the list of configuration is guaranteed to be non-empty
-    // from the guard above.
-    fatalError()
+    switch result {
+        case .success(let collection):
+            return collection
+        case .failure(let pipelineError):
+            throw pipelineError
+    }
 }
 
 /// Run an executable with given parameters specified by a `Configuration`