allow to pool servers together when accepting incoming connections

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - `Innmind\IO\Sockets::servers()`
 - `Innmind\IO\Sockets\Servers`
+- `Innmind\IO\Sockets\Server\Pool`
 
 ## 2.6.0 - 2024-03-09
 

src/Sockets/Server.php

@@ -65,6 +65,16 @@ public static function of(
         );
     }
 
+    /**
+     * @param self<T> $socket
+     *
+     * @return Server\Pool<T>
+     */
+    public function with(self $socket): Server\Pool
+    {
+        return Server\Pool::of($this->watch, $this->socket, $socket->unwrap());
+    }
+
     /**
      * @return T
      */

src/Sockets/Server/Pool.php

@@ -0,0 +1,174 @@
+<?php
+declare(strict_types = 1);
+
+namespace Innmind\IO\Sockets\Server;
+
+use Innmind\IO\Sockets\{
+    Server,
+    Client,
+};
+use Innmind\TimeContinuum\ElapsedPeriod;
+use Innmind\Socket\{
+    Server as Socket,
+    Server\Connection,
+};
+use Innmind\Stream\Watch;
+use Innmind\Immutable\{
+    Sequence,
+    Predicate\Instance,
+};
+
+/**
+ * @template-covariant T of Socket
+ */
+final class Pool
+{
+    /** @var non-empty-list<T> */
+    private array $sockets;
+    /** @var callable(?ElapsedPeriod): Watch */
+    private $watch;
+    /** @var callable(T): Sequence<T> */
+    private $wait;
+
+    /**
+     * @psalm-mutation-free
+     *
+     * @param callable(?ElapsedPeriod): Watch $watch
+     * @param non-empty-list<T> $sockets
+     * @param callable(T): Sequence<T> $wait
+     */
+    private function __construct(
+        callable $watch,
+        array $sockets,
+        callable $wait,
+    ) {
+        $this->watch = $watch;
+        $this->sockets = $sockets;
+        $this->wait = $wait;
+    }
+
+    /**
+     * @psalm-mutation-free
+     * @internal
+     * @template A of Socket
+     *
+     * @param callable(?ElapsedPeriod): Watch $watch
+     * @param A $first
+     * @param A $second
+     *
+     * @return self<A>
+     */
+    public static function of(
+        callable $watch,
+        Socket $first,
+        Socket $second,
+    ): self {
+        /** @var self<A> */
+        return new self(
+            $watch,
+            [$first, $second],
+            static fn(Socket $socket) => Sequence::of($socket),
+        );
+    }
+
+    /**
+     * @param Server<T> $server
+     *
+     * @return self<T>
+     */
+    public function with(Server $server): self
+    {
+        return new self(
+            $this->watch,
+            [...$this->sockets, $server->unwrap()],
+            $this->wait,
+        );
+    }
+
+    /**
+     * @return Sequence<T>
+     */
+    public function unwrap(): Sequence
+    {
+        return Sequence::of(...$this->sockets);
+    }
+
+    /**
+     * Wait forever for the socket to be ready to read before tryin to use it
+     *
+     * @psalm-mutation-free
+     *
+     * @return self<T>
+     */
+    public function watch(): self
+    {
+        /** @var self<T> */
+        return new self(
+            $this->watch,
+            $this->sockets,
+            fn(Socket $socket, Socket ...$sockets) => ($this->watch)(null)
+                ->forRead($socket, ...$sockets)()
+                ->map(
+                    static fn($ready) => $ready
+                        ->toRead()
+                        ->filter(static fn($ready) => \in_array(
+                            $ready,
+                            [$socket, ...$sockets],
+                            true,
+                        ))
+                        ->keep(Instance::of(Socket::class)),
+                )
+                ->toSequence()
+                ->flatMap(
+                    static fn($toRead) => Sequence::of(...$toRead->toList()),
+                ),
+        );
+    }
+
+    /**
+     * @psalm-mutation-free
+     *
+     * @return self<T>
+     */
+    public function timeoutAfter(ElapsedPeriod $timeout): self
+    {
+        /** @var self<T> */
+        return new self(
+            $this->watch,
+            $this->sockets,
+            fn(Socket $socket, Socket ...$sockets) => ($this->watch)($timeout)
+                ->forRead($socket, ...$sockets)()
+                ->map(
+                    static fn($ready) => $ready
+                        ->toRead()
+                        ->filter(static fn($ready) => \in_array(
+                            $ready,
+                            [$socket, ...$sockets],
+                            true,
+                        ))
+                        ->keep(Instance::of(Socket::class)),
+                )
+                ->toSequence()
+                ->flatMap(
+                    static fn($toRead) => Sequence::of(...$toRead->toList()),
+                ),
+        );
+    }
+
+    /**
+     * @return Sequence<Client<Connection>>
+     */
+    public function accept(): Sequence
+    {
+        return ($this->wait)(...$this->sockets)
+            ->flatMap(
+                static fn($socket) => $socket
+                    ->accept()
+                    ->toSequence(),
+            )
+            ->map(fn($client) => Client::of(
+                $this->watch,
+                $client,
+            ));
+    }
+}

tests/FunctionalTest.php

@@ -773,4 +773,111 @@ public function testServerAcceptConnection()
         $client->close();
         $server->close();
     }
+
+    public function testServerPool()
+    {
+        @\unlink('/tmp/foo.sock');
+        @\unlink('/tmp/bar.sock');
+        @\unlink('/tmp/baz.sock');
+        $addressFoo = Address\Unix::of('/tmp/foo');
+        $addressBar = Address\Unix::of('/tmp/bar');
+        $addressBaz = Address\Unix::of('/tmp/baz');
+        $serverFoo = Server\Unix::recoverable($addressFoo)->match(
+            static fn($server) => $server,
+            static fn() => null,
+        );
+
+        $this->assertNotNull($serverFoo);
+
+        $serverBar = Server\Unix::recoverable($addressBar)->match(
+            static fn($server) => $server,
+            static fn() => null,
+        );
+
+        $this->assertNotNull($serverBar);
+
+        $serverBaz = Server\Unix::recoverable($addressBaz)->match(
+            static fn($server) => $server,
+            static fn() => null,
+        );
+
+        $this->assertNotNull($serverBaz);
+
+        $clientFoo = Client\Unix::of($addressFoo)->match(
+            static fn($socket) => $socket,
+            static fn() => null,
+        );
+
+        $this->assertNotNull($clientFoo);
+
+        $clientBar = Client\Unix::of($addressBar)->match(
+            static fn($socket) => $socket,
+            static fn() => null,
+        );
+
+        $this->assertNotNull($clientBar);
+
+        $clientBaz = Client\Unix::of($addressBaz)->match(
+            static fn($socket) => $socket,
+            static fn() => null,
+        );
+
+        $this->assertNotNull($clientBaz);
+
+        $_ = IO::of(Select::timeoutAfter(...))
+            ->sockets()
+            ->clients()
+            ->wrap($clientFoo)
+            ->send(Sequence::of(Str::of('foo')))
+            ->match(
+                static fn() => null,
+                static fn() => null,
+            );
+        $_ = IO::of(Select::timeoutAfter(...))
+            ->sockets()
+            ->clients()
+            ->wrap($clientBar)
+            ->send(Sequence::of(Str::of('bar')))
+            ->match(
+                static fn() => null,
+                static fn() => null,
+            );
+        $_ = IO::of(Select::timeoutAfter(...))
+            ->sockets()
+            ->clients()
+            ->wrap($clientBaz)
+            ->send(Sequence::of(Str::of('baz')))
+            ->match(
+                static fn() => null,
+                static fn() => null,
+            );
+
+        $servers = IO::of(Select::timeoutAfter(...))
+            ->sockets()
+            ->servers();
+        $result = $servers
+            ->wrap($serverFoo)
+            ->with($servers->wrap($serverBar))
+            ->with($servers->wrap($serverBaz))
+            ->timeoutAfter(ElapsedPeriod::of(1_000))
+            ->accept()
+            ->flatMap(
+                static fn($client) => $client
+                    ->frames(Frame\Chunk::of(3))
+                    ->one()
+                    ->toSequence(),
+            )
+            ->map(static fn($data) => $data->toString());
+
+        $this->assertCount(3, $result);
+        $this->assertTrue($result->contains('foo'));
+        $this->assertTrue($result->contains('bar'));
+        $this->assertTrue($result->contains('baz'));
+        $clientFoo->close();
+        $clientBar->close();
+        $clientBaz->close();
+        $serverFoo->close();
+        $serverBar->close();
+        $serverBaz->close();
+    }
 }