[10.x] Add pipe function to Process layer (#46527)

* [10.x] Add pipe function to Process layer

* Fixing code style

* Update pipe to accept fully configured processes as parameters

* Fixing code style

* Remove unused use statement on test file

* formatting

* exception message

* Wrap error in process pipe with a custom ProcessPipeException

* Fixed code style issues

* Updated pipe to return last failed process instead of throwing exception and added test case for failed pipe run

* Update pipe to handle keys for processes

---------

Co-authored-by: Taylor Otwell <[email protected]>

Author: WendellAdriel

src/Illuminate/Process/Factory.php

@@ -271,6 +271,17 @@ public function pool(callable $callback)
         return new Pool($this, $callback);
     }
 
+    /**
+     * Start defining a series of piped processes.
+     *
+     * @param  callable  $callback
+     * @return \Illuminate\Process\Pipe
+     */
+    public function pipe(callable $callback)
+    {
+        return new Pipe($this, $callback);
+    }
+
     /**
      * Run a pool of processes and wait for them to finish executing.
      *

src/Illuminate/Process/PendingProcess.php

@@ -5,13 +5,16 @@
 use Closure;
 use Illuminate\Process\Exceptions\ProcessTimedOutException;
 use Illuminate\Support\Str;
+use Illuminate\Support\Traits\Conditionable;
 use LogicException;
 use RuntimeException;
 use Symfony\Component\Process\Exception\ProcessTimedOutException as SymfonyTimeoutException;
 use Symfony\Component\Process\Process;
 
 class PendingProcess
 {
+    use Conditionable;
+
     /**
      * The process factory instance.
      *

src/Illuminate/Process/Pipe.php

@@ -0,0 +1,102 @@
+<?php
+
+namespace Illuminate\Process;
+
+use InvalidArgumentException;
+
+/**
+ * @mixin \Illuminate\Process\Factory
+ * @mixin \Illuminate\Process\PendingProcess
+ */
+class Pipe
+{
+    /**
+     * The process factory instance.
+     *
+     * @var \Illuminate\Process\Factory
+     */
+    protected $factory;
+
+    /**
+     * The callback that resolves the pending processes.
+     *
+     * @var callable
+     */
+    protected $callback;
+
+    /**
+     * The array of pending processes.
+     *
+     * @var array
+     */
+    protected $pendingProcesses = [];
+
+    /**
+     * Create a new series of piped processes.
+     *
+     * @param  \Illuminate\Process\Factory  $factory
+     * @param  callable  $callback
+     * @return void
+     */
+    public function __construct(Factory $factory, callable $callback)
+    {
+        $this->factory = $factory;
+        $this->callback = $callback;
+    }
+
+    /**
+     * Add a process to the pipe with a key.
+     *
+     * @param  string  $key
+     * @return \Illuminate\Process\PendingProcess
+     */
+    public function as(string $key)
+    {
+        return tap($this->factory->newPendingProcess(), function ($pendingProcess) use ($key) {
+            $this->pendingProcesses[$key] = $pendingProcess;
+        });
+    }
+
+    /**
+     * Runs the processes in the pipe.
+     *
+     * @param  callable|null  $output
+     * @return \Illuminate\Contracts\Process\ProcessResult
+     */
+    public function run(?callable $output = null)
+    {
+        call_user_func($this->callback, $this);
+
+        return collect($this->pendingProcesses)
+                ->reduce(function ($previousProcessResult, $pendingProcess, $key) use ($output) {
+                    if (! $pendingProcess instanceof PendingProcess) {
+                        throw new InvalidArgumentException('Process pipe must only contain pending processes.');
+                    }
+
+                    if ($previousProcessResult && $previousProcessResult->failed()) {
+                        return $previousProcessResult;
+                    }
+
+                    return $pendingProcess->when(
+                        $previousProcessResult,
+                        fn () => $pendingProcess->input($previousProcessResult->output())
+                    )->run(output: $output ? function ($type, $buffer) use ($key, $output) {
+                        $output($type, $buffer, $key);
+                    } : null);
+                });
+    }
+
+    /**
+     * Dynamically proxy methods calls to a new pending process.
+     *
+     * @param  string  $method
+     * @param  array  $parameters
+     * @return \Illuminate\Process\PendingProcess
+     */
+    public function __call($method, $parameters)
+    {
+        return tap($this->factory->{$method}(...$parameters), function ($pendingProcess) {
+            $this->pendingProcesses[] = $pendingProcess;
+        });
+    }
+}

src/Illuminate/Support/Facades/Process.php

@@ -33,6 +33,7 @@
  * @method static \Illuminate\Process\Factory assertDidntRun(\Closure|string $callback)
  * @method static \Illuminate\Process\Factory assertNothingRan()
  * @method static \Illuminate\Process\Pool pool(callable $callback)
+ * @method static \Illuminate\Process\Pipe pipe(callable $callback)
  * @method static \Illuminate\Process\ProcessPoolResults concurrently(callable $callback, callable|null $output = null)
  * @method static \Illuminate\Process\PendingProcess newPendingProcess()
  * @method static void macro(string $name, object|callable $macro)

tests/Process/ProcessTest.php

@@ -496,6 +496,44 @@ public function testRealProcessesCanUseStandardInput()
         $this->assertSame('foobar', $result->output());
     }
 
+    public function testProcessPipe()
+    {
+        if (windows_os()) {
+            $this->markTestSkipped('Requires Linux.');
+        }
+
+        $factory = new Factory;
+        $factory->fake([
+            'cat *' => "Hello, world\nfoo\nbar",
+        ]);
+
+        $pipe = $factory->pipe(function ($pipe) {
+            $pipe->command('cat test');
+            $pipe->command('grep -i "foo"');
+        });
+
+        $this->assertSame("foo\n", $pipe->run()->output());
+    }
+
+    public function testProcessPipeFailed()
+    {
+        if (windows_os()) {
+            $this->markTestSkipped('Requires Linux.');
+        }
+
+        $factory = new Factory;
+        $factory->fake([
+            'cat *' => $factory->result(exitCode: 1),
+        ]);
+
+        $pipe = $factory->pipe(function ($pipe) {
+            $pipe->command('cat test');
+            $pipe->command('grep -i "foo"');
+        });
+
+        $this->assertTrue($pipe->run()->failed());
+    }
+
     public function testFakeInvokedProcessOutputWithLatestOutput()
     {
         $factory = new Factory;