Consume process output before closing

lib/Internal/Posix/Handle.php

@@ -4,6 +4,7 @@ namespace Amp\Process\Internal\Posix;
 
 use Amp\Deferred;
 use Amp\Process\Internal\ProcessHandle;
+use Amp\Promise;
 
 /** @internal */
 final class Handle extends ProcessHandle

lib/Internal/Posix/Runner.php

@@ -44,7 +44,8 @@ final class Runner implements ProcessRunner
 
         $pid = \rtrim(@\fgets($stream));
 
-        /** @var $deferreds Deferred[] */
+        /** @var Handle $handle */
+        /** @var Deferred[] $deferreds */
         list($handle, $pipes, $deferreds) = $data;
 
         if (!$pid || !\is_numeric($pid)) {
@@ -67,11 +68,47 @@ final class Runner implements ProcessRunner
         $deferreds[1]->resolve($pipes[1]);
         $deferreds[2]->resolve($pipes[2]);
 
+        if ("" !== $exitCode = \rtrim(@\fgets($stream))) {
+            $handle->status = ProcessStatus::ENDED;
+            $handle->joinDeferred->resolve((int) $exitCode);
+
+            self::free($handle);
+
+            return;
+        }
+
         if ($handle->extraDataPipeWatcher !== null) {
             Loop::enable($handle->extraDataPipeWatcher);
         }
     }
 
+    private static function free(Handle $handle)
+    {
+        /** @var Handle $handle */
+        if ($handle->extraDataPipeWatcher !== null) {
+            Loop::cancel($handle->extraDataPipeWatcher);
+            $handle->extraDataPipeWatcher = null;
+        }
+
+        /** @var Handle $handle */
+        if ($handle->extraDataPipeStartWatcher !== null) {
+            Loop::cancel($handle->extraDataPipeStartWatcher);
+            $handle->extraDataPipeStartWatcher = null;
+        }
+
+        if (\is_resource($handle->extraDataPipe)) {
+            \fclose($handle->extraDataPipe);
+        }
+
+        $handle->stdin->close();
+        $handle->stdout->close();
+        $handle->stderr->close();
+
+        if (\is_resource($handle->proc)) {
+            \proc_close($handle->proc);
+        }
+    }
+
     /** @inheritdoc */
     public function start(string $command, string $cwd = null, array $env = [], array $options = []): ProcessHandle
     {
@@ -164,7 +201,7 @@ final class Runner implements ProcessRunner
             $handle->joinDeferred->fail(new ProcessException("The process was killed"));
         }
 
-        $this->free($handle);
+        self::free($handle);
     }
 
     /** @inheritdoc */
@@ -189,33 +226,6 @@ final class Runner implements ProcessRunner
             }
         }
 
-        $this->free($handle);
+        self::free($handle);
-    }
-
-    private function free(Handle $handle)
-    {
-        /** @var Handle $handle */
-        if ($handle->extraDataPipeWatcher !== null) {
-            Loop::cancel($handle->extraDataPipeWatcher);
-            $handle->extraDataPipeWatcher = null;
-        }
-
-        /** @var Handle $handle */
-        if ($handle->extraDataPipeStartWatcher !== null) {
-            Loop::cancel($handle->extraDataPipeStartWatcher);
-            $handle->extraDataPipeStartWatcher = null;
-        }
-
-        if (\is_resource($handle->extraDataPipe)) {
-            \fclose($handle->extraDataPipe);
-        }
-
-        $handle->stdin->close();
-        $handle->stdout->close();
-        $handle->stderr->close();
-
-        if (\is_resource($handle->proc)) {
-            \proc_close($handle->proc);
-        }
     }
 }

lib/ProcessInputStream.php

@@ -28,6 +28,9 @@ class ProcessInputStream implements InputStream
     /** @var StreamException|null */
     private $error;
 
+    /** @var string|null */
+    private $buffer;
+
     public function __construct(Promise $resourceStreamPromise)
     {
         $resourceStreamPromise->onResolve(function ($error, $resourceStream) {
@@ -48,14 +51,25 @@ class ProcessInputStream implements InputStream
             }
 
             if ($this->shouldClose) {
+                if ($this->resourceStream->getResource()) {
+                    $this->buffer .= \stream_get_contents($this->resourceStream->getResource());
+                }
+
                 $this->resourceStream->close();
             }
 
             if ($this->initialRead) {
                 $initialRead = $this->initialRead;
                 $this->initialRead = null;
+
+                if ($this->buffer !== null) {
+                    $buffer = $this->buffer;
+                    $this->buffer = null;
+                    $initialRead->resolve($buffer);
+                } else {
                     $initialRead->resolve($this->shouldClose ? null : $this->resourceStream->read());
                 }
+            }
         });
     }
 
@@ -66,6 +80,12 @@ class ProcessInputStream implements InputStream
             throw new PendingReadError;
         }
 
+        if ($this->buffer !== null) {
+            $buffer = $this->buffer;
+            $this->buffer = null;
+            return $buffer;
+        }
+
         if ($this->error) {
             throw $this->error;
         }
@@ -105,10 +125,21 @@ class ProcessInputStream implements InputStream
     {
         $this->shouldClose = true;
 
+        if ($this->resourceStream->getResource()) {
+            $this->buffer .= \stream_get_contents($this->resourceStream->getResource());
+        }
+
         if ($this->initialRead) {
             $initialRead = $this->initialRead;
             $this->initialRead = null;
-            $initialRead->resolve();
+
+            if ($this->buffer !== null) {
+                $buffer = $this->buffer;
+                $this->buffer = null;
+                $initialRead->resolve($buffer);
+            } else {
+                $initialRead->resolve(null);
+            }
         }
 
         if ($this->resourceStream) {