Create a runPipe that sets up a multi-process pipeline from multiple Configurations

Author: cmcgee1024

Sources/Subprocess/API.swift

@@ -557,6 +557,100 @@ public func run<
     )
 }
 
+public enum PipeConfigurationError: Error {
+    case emptyConfiguration
+}
+
+/// 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> {
+    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
+    }
+
+    var prevInput: InputProtocol = input
+
+    for (idx, cmd) in pipeline.enumerated() {
+        let currentInput = prevInput
+        let currentOutput: any OutputProtocol
+
+        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
+        }
+
+        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()
+                }
+            }
+        }
+    }
+
+    // This should never happen because the list of configuration is guaranteed to be non-empty
+    // from the guard above.
+    fatalError()
+}
+
 /// Run an executable with given parameters specified by a `Configuration`
 /// - Parameters:
 ///   - configuration: The `Subprocess` configuration to run.