Merge pull request #7 from Innmind/non-blocking-pool

Add non blocking stream pool

Author: Baptouuuu

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Added
+
+- `Innmind\IO\Streams\Stream\Read\Pool::nonBlocking()`
+
 ### Fixed
 
 - `Innmind\IO\Streams\Stream\Read\Pool` was losing the id type when calling `->watch()`, `->timeoutAfter()` or `->toEncoding()`

proofs/streams.php

@@ -306,6 +306,59 @@ static function($assert, $a, $b, $encoding) {
         },
     );
 
+    yield proof(
+        'IO::streams()->acquire()->read()->pool()->nonBlocking()->chunks()',
+        given(
+            $string,
+            $string,
+            Set::of(...Str\Encoding::cases()),
+        ),
+        static function($assert, $a, $b, $encoding) {
+            $tmpA = \tmpfile();
+            \fwrite($tmpA, $a);
+            $tmpB = \tmpfile();
+            \fwrite($tmpB, $b);
+
+            $io = IO::fromAmbientAuthority();
+            $chunks = $io
+                ->streams()
+                ->acquire($tmpA)
+                ->read()
+                ->pool('a')
+                ->with(
+                    'b',
+                    $io
+                        ->streams()
+                        ->acquire($tmpB)
+                        ->read(),
+                )
+                ->watch()
+                ->toEncoding($encoding)
+                ->nonBlocking()
+                ->chunks();
+
+            $assert->same(2, $chunks->size());
+            $chunks->foreach(static fn($chunk) => $assert->same(
+                $encoding,
+                $chunk->value()->encoding(),
+            ));
+            $assert->same(
+                [$a],
+                $chunks
+                    ->filter(static fn($chunk) => $chunk->key() === 'a')
+                    ->map(static fn($chunk) => $chunk->value()->toString())
+                    ->toList(),
+            );
+            $assert->same(
+                [$b],
+                $chunks
+                    ->filter(static fn($chunk) => $chunk->key() === 'b')
+                    ->map(static fn($chunk) => $chunk->value()->toString())
+                    ->toList(),
+            );
+        },
+    );
+
     yield proof(
         'IO::streams()->acquire()->close()',
         given($string),

src/Streams/Stream/Read.php

@@ -71,6 +71,11 @@ public function internal(): Stream
     }
 
     /**
+     * This doesn't affect the blocking when pooling.
+     *
+     * To prevent having some streams being blocking and others not, the non
+     * blocking have to be explicitly called on the pool.
+     *
      * @psalm-mutation-free
      */
     public function nonBlocking(): self

src/Streams/Stream/Read/Pool.php

@@ -30,6 +30,7 @@ private function __construct(
         private Map $streams,
         private ?Period $timeout,
         private ?Str\Encoding $encoding,
+        private bool $blocking,
     ) {
     }
 
@@ -51,6 +52,7 @@ public static function of(
             Map::of([$stream->internal(), $id]),
             null,
             null,
+            true,
         );
     }
 
@@ -69,6 +71,7 @@ public function with(mixed $id, Read $stream): self
             ($this->streams)($stream->internal(), $id),
             $this->timeout,
             $this->encoding,
+            $this->blocking,
         );
     }
 
@@ -90,6 +93,7 @@ public function watch(): self
             $this->streams,
             null,
             $this->encoding,
+            $this->blocking,
         );
     }
 
@@ -103,6 +107,7 @@ public function timeoutAfter(Period $timeout): self
             $this->streams,
             $timeout,
             $this->encoding,
+            $this->blocking,
         );
     }
 
@@ -116,6 +121,21 @@ public function toEncoding(Str\Encoding $encoding): self
             $this->streams,
             $this->timeout,
             $encoding,
+            $this->blocking,
+        );
+    }
+
+    /**
+     * @return self<T>
+     */
+    public function nonBlocking(): self
+    {
+        return new self(
+            $this->capabilities,
+            $this->streams,
+            $this->timeout,
+            $this->encoding,
+            false,
         );
     }
 
@@ -124,6 +144,7 @@ public function toEncoding(Str\Encoding $encoding): self
      */
     public function chunks(): Sequence
     {
+        $blocking = $this->blocking;
         $watch = $this->capabilities->watch();
         $watch = match ($this->timeout) {
             null => $watch->waitForever(),
@@ -133,6 +154,15 @@ public function chunks(): Sequence
             ->streams
             ->keys()
             ->filter(static fn($stream) => !$stream->closed())
+            ->flatMap(
+                static fn($stream) => (match ($blocking) {
+                    true => $stream->blocking(),
+                    false => $stream->nonBlocking(),
+                })
+                    ->map(static fn() => $stream)
+                    ->toSequence()
+                    ->toSet(),
+            )
             ->reduce(
                 $watch,
                 static fn(Internal\Watch $watch, $stream) => $watch->forRead($stream),