Merge pull request #5 from Innmind/fix-output-streaming-with-long-input

Process output should not be kept in memory while writing to the input stream

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## [Unreleased]
+
+### Fixed
+
+- Output generated while writing the input is now directly available (previously it was when the whole input was written)
+
 ## 5.2.1 - 2023-11-11
 
 ### Fixed

src/Server/Process.php

@@ -6,7 +6,6 @@
 use Innmind\Server\Control\Server\Process\{
     Pid,
     Output,
-    ExitCode,
     TimedOut,
     Failed,
     Signaled,

src/Server/Process/Background.php

@@ -20,7 +20,7 @@ public function __construct(Started $process)
         // wait for the process to be started in the background otherwise the
         // process will be killed
         // this also allows to send any input to the stream
-        $process->wait();
+        $process->output()->memoize();
         /** @var Sequence<array{0: Str, 1: Output\Type}> */
         $output = Sequence::of();
         $this->output = new Output\Output($output);

src/Server/Process/Foreground.php

@@ -3,12 +3,16 @@
 
 namespace Innmind\Server\Control\Server\Process;
 
-use Innmind\Server\Control\Server\Process;
+use Innmind\Server\Control\{
+    Server\Process,
+    Server\Process\Output\Chunk,
+    Exception\RuntimeException,
+};
 use Innmind\Immutable\{
     Sequence,
-    Str,
     Maybe,
     Either,
+    Predicate\Instance,
 };
 
 final class Foreground implements Process
@@ -22,29 +26,35 @@ public function __construct(Started $process, bool $streamOutput = false)
     {
         $this->process = $process;
         $yieldOutput = function() use ($process): \Generator {
-            $output = $process->output();
-
-            foreach ($output as $chunk) {
-                yield $chunk;
-            }
+            yield $process
+                ->output()
+                ->map(function($chunk) {
+                    if ($chunk instanceof Either) {
+                        $this->status = $chunk
+                            ->map(fn() => new Success($this->output))
+                            ->leftMap(fn($error) => match ($error) {
+                                'timed-out' => new TimedOut($this->output),
+                                'signaled' => new Signaled($this->output),
+                                default => new Failed($error, $this->output),
+                            });
+                    }
 
-            $this->status = $output
-                ->getReturn()
-                ->map(fn() => new Success($this->output))
-                ->leftMap(fn($error) => match ($error) {
-                    'timed-out' => new TimedOut($this->output),
-                    'signaled' => new Signaled($this->output),
-                    default => new Failed($error, $this->output),
-                });
+                    return $chunk;
+                })
+                ->keep(Instance::of(Chunk::class));
         };
 
         if ($streamOutput) {
-            $output = Sequence::lazy($yieldOutput);
+            $output = Sequence::lazy($yieldOutput)->flatMap(
+                static fn($chunks) => $chunks,
+            );
         } else {
-            $output = Sequence::defer($yieldOutput());
+            $output = Sequence::defer($yieldOutput())->flatMap(
+                static fn($chunks) => $chunks,
+            );
         }
 
-        $this->output = new Output\Output($output);
+        $this->output = Output\Output::of($output);
     }
 
     public function pid(): Maybe
@@ -67,16 +77,12 @@ public function wait(): Either
             $_ = $this->output->foreach(static fn() => null);
         }
 
+        if (\is_null($this->status)) {
+            throw new RuntimeException('Unable to retrieve the status');
+        }
+
         // the status should always be set here because we iterated over the
-        // output above but we stil coalesce to the wait() call to please psalm
-        return $this->status ??= $this
-            ->process
-            ->wait()
-            ->map(fn() => new Success($this->output))
-            ->leftMap(fn($error) => match ($error) {
-                'timed-out' => new TimedOut($this->output),
-                'signaled' => new Signaled($this->output),
-                default => new Failed($error, $this->output),
-            });
+        // output above
+        return $this->status;
     }
 }

src/Server/Process/Logger.php

@@ -68,18 +68,18 @@ public function wait(): Either
             ->wait()
             ->leftMap(function($e) {
                 [$message, $context] = match (\get_class($e)) {
-                    Process\Signaled::class => [
+                    Signaled::class => [
                         'Command {command} stopped due to external signal',
                         ['command' => $this->command->toString()],
                     ],
-                    Process\Failed::class => [
+                    Failed::class => [
                         'Command {command} failed with {exitCode}',
                         [
                             'command' => $this->command->toString(),
                             'exitCode' => $e->exitCode()->toInt(),
                         ],
                     ],
-                    Process\TimedOut::class => [
+                    TimedOut::class => [
                         'Command {command} timed out',
                         ['command' => $this->command->toString()],
                     ],

src/Server/Process/Output/Chunk.php

@@ -0,0 +1,36 @@
+<?php
+declare(strict_types = 1);
+
+namespace Innmind\Server\Control\Server\Process\Output;
+
+use Innmind\Immutable\Str;
+
+/**
+ * @psalm-immutable
+ */
+final class Chunk
+{
+    private function __construct(
+        private Str $data,
+        private Type $type,
+    ) {
+    }
+
+    /**
+     * @psalm-pure
+     */
+    public static function of(Str $data, Type $type): self
+    {
+        return new self($data, $type);
+    }
+
+    public function data(): Str
+    {
+        return $this->data;
+    }
+
+    public function type(): Type
+    {
+        return $this->type;
+    }
+}

src/Server/Process/Output/Output.php

@@ -27,6 +27,17 @@ public function __construct(Sequence $output)
         $this->output = $output;
     }
 
+    /**
+     * @param Sequence<Chunk> $chunks
+     */
+    public static function of(Sequence $chunks): self
+    {
+        return new self($chunks->map(static fn($chunk) => [
+            $chunk->data(),
+            $chunk->type(),
+        ]));
+    }
+
     /**
      * @param callable(Str, Type): void $function
      */