Merge branch 'develop'

* develop: (62 commits) specify next release update benchmark load message body in memory when it is less than 2Mo fix example update changelog discard psalm error require innmind/operating-system 5 remove no longer relevant comment remove dead code ignore cs config from code coverage automatically translate values based on their type fix closing the connection when the process receives a signal remove dead code fix tests CS use IO socket wrapper remove todo as it is already handled add a message for further improvement on how to read the messages rely on the os filesystem to create a temporary file content fix using $this in static context ...
Innmind · Mar 10, 2024 · ef2a68c · ef2a68c
2 parents 7e3040c + 62c2579
commit ef2a68c
Show file tree

Hide file tree

Showing 102 changed files with 2,083 additions and 2,200 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,16 @@
 # Changelog
 
+## 5.0.0 - 2024-03-10
+
+### Changed
+
+- Requires `innmind/immutable:~5.2`
+- Requires `innmind/operating-system:~5.0`
+- Requires `innmind/filesystem:~7.0`
+- Requires `innmind/io:~2.6`
+- Carried state inside a `Innmind\AMQP\Command` is now wrapped inside a `Innmind\AMQP\Client\State`
+- `Innmind\AMQP\Client::of()` now requires an instance of `Innmind\OperatingSystem\Filesystem` as a second argument
+
 ## 4.3.0 - 2023-09-23
 
 ### Added

diff --git a/README.md b/README.md
@@ -110,10 +110,10 @@ Feel free to look at the `Command` namespace to explore all capabilities.
 make benchmark
 Publishing 4000 msgs with 1KB of content:
 php benchmark/producer.php 4000
-0.39038109779358
+0.48978996276855
 Consuming 4000:
 php benchmark/consumer.php
-Pid: 701, Count: 4000, Time: 1.6017
+Pid: 701, Count: 4000, Time: 2.3580
 ```
 
 By comparison, the `php-amqplib` produces this result:

diff --git a/codecov.yml b/codecov.yml
@@ -1,3 +1,4 @@
 ignore:
     - benchmark
     - fixtures
+    - .php-cs-fixer.dist.php
diff --git a/composer.json b/composer.json
@@ -16,15 +16,16 @@
     },
     "require": {
         "php": "~8.2",
-        "innmind/immutable": "~4.15|~5.0",
+        "innmind/immutable": "~5.2",
         "innmind/time-continuum": "~3.1",
         "innmind/math": "~6.0",
         "innmind/url": "~4.1",
         "ramsey/uuid": "~4.0",
-        "innmind/operating-system": "~3.1",
+        "innmind/operating-system": "~5.0",
         "innmind/media-type": "~2.0",
-        "innmind/filesystem": "~5.2|~6.0",
-        "innmind/stream": "~4.0"
+        "innmind/filesystem": "~7.0",
+        "innmind/stream": "~4.0",
+        "innmind/io": "~2.6"
     },
     "autoload": {
         "psr-4": {

diff --git a/docs/Publish many messages.md b/docs/Publish many messages.md
@@ -43,7 +43,7 @@ $client = Factory::of($os)->make(/* details */);
 $os
     ->filesystem()
     ->mount(Path::of('/path/to/some/directory/'))
-    ->get(new Name('leads.csv'))
+    ->get(Name::of('leads.csv'))
     ->map(
         static fn($file) => $file
             ->content()

diff --git a/src/Client.php b/src/Client.php
@@ -10,10 +10,9 @@
     Transport\Frame\Method,
     Model\Channel\Close as CloseChannel,
 };
-use Innmind\OperatingSystem\CurrentProcess;
-use Innmind\Stream\{
-    Capabilities,
-    Streams,
+use Innmind\OperatingSystem\{
+    CurrentProcess,
+    Filesystem,
 };
 use Innmind\Immutable\{
     Either,
@@ -27,7 +26,7 @@ final class Client
     private Maybe $command;
     /** @var callable(): Maybe<Connection> */
     private $load;
-    private Capabilities $streams;
+    private Filesystem $filesystem;
     /** @var Maybe<CurrentProcess> */
     private Maybe $signals;
 
@@ -39,19 +38,19 @@ final class Client
     private function __construct(
         Maybe $command,
         callable $load,
-        Capabilities $streams,
+        Filesystem $filesystem,
         Maybe $signals,
     ) {
         $this->command = $command;
         $this->load = $load;
-        $this->streams = $streams;
+        $this->filesystem = $filesystem;
         $this->signals = $signals;
     }
 
     /**
      * @param callable(): Maybe<Connection> $load
      */
-    public static function of(callable $load, Capabilities $streams = null): self
+    public static function of(callable $load, Filesystem $filesystem): self
     {
         /** @var Maybe<Command> */
         $command = Maybe::nothing();
@@ -61,7 +60,7 @@ public static function of(callable $load, Capabilities $streams = null): self
         return new self(
             $command,
             $load,
-            $streams ?? Streams::fromAmbientAuthority(),
+            $filesystem,
             $signals,
         );
     }
@@ -74,22 +73,17 @@ public function with(Command $command): self
                 ->map(static fn($previous) => new Command\Pipe($previous, $command))
                 ->otherwise(static fn() => Maybe::just($command)),
             $this->load,
-            $this->streams,
+            $this->filesystem,
             $this->signals,
         );
     }
 
     public function listenSignals(CurrentProcess $currentProcess): self
     {
-        // We ask for the current process instead of the signals wrapper directly
-        // because the user may fork the process between the time this method is
-        // called and the time the listeners are installed (when run is called).
-        // This would result on the listeners being installed for the parent
-        // process instead of the child.
         return new self(
             $this->command,
             $this->load,
-            $this->streams,
+            $this->filesystem,
             Maybe::just($currentProcess),
         );
     }
@@ -108,12 +102,12 @@ public function run(mixed $state): Either
                 ->openChannel()
                 ->flatMap(function($in) use ($command, $state) {
                     [$connection, $channel] = $in;
-                    $read = MessageReader::of($this->streams);
+                    $read = MessageReader::of($this->filesystem);
 
-                    return $command($connection, $channel, $read, $state)->flatMap(
-                        fn($clientState) => $this
-                            ->close($clientState->connection(), $channel)
-                            ->map(static fn(): mixed => $clientState->userState()),
+                    return $command($connection, $channel, $read, Client\State::of($state))->flatMap(
+                        fn($state) => $this
+                            ->close($connection, $channel)
+                            ->map(static fn(): mixed => $state->unwrap()),
                     );
                 }),
             static fn() => Either::right($state),
@@ -134,21 +128,20 @@ private function openChannel(): Either
         return ($this->load)()
             ->either()
             ->leftMap(static fn() => Failure::toOpenConnection())
-            ->map(static fn($connection) => $connection->send(
-                static fn($protocol) => $protocol->channel()->open($channel),
-            ))
-            ->map(static fn($continuation) => $continuation->wait(Method::channelOpenOk))
             ->flatMap(
-                fn($continuation) => $continuation
-                    ->connection()
-                    ->map(fn($connection) => $this->signals->match(
+                fn($connection) => $connection
+                    ->request(
+                        static fn($protocol) => $protocol->channel()->open($channel),
+                        Method::channelOpenOk,
+                    )
+                    ->map(fn() => $this->signals->match(
                         static fn($process) => $connection->listenSignals(
                             $process->signals(),
                             $channel,
                         ),
-                        static fn() => $connection,
+                        static fn() => null,
                     ))
-                    ->map(static fn($connection) => [$connection, $channel])
+                    ->map(static fn() => [$connection, $channel])
                     ->leftMap(static fn() => Failure::toOpenChannel()),
             );
     }
@@ -160,15 +153,16 @@ private function close(Connection $connection, Channel $channel): Either
     {
         /** @var Either<Failure, SideEffect> */
         return $connection
-            ->send(static fn($protocol) => $protocol->channel()->close(
-                $channel,
-                CloseChannel::demand(),
-            ))
-            ->wait(Method::channelCloseOk)
-            ->connection()
+            ->request(
+                static fn($protocol) => $protocol->channel()->close(
+                    $channel,
+                    CloseChannel::demand(),
+                ),
+                Method::channelCloseOk,
+            )
             ->leftMap(static fn() => Failure::toCloseChannel())
             ->flatMap(
-                static fn($connection) => $connection
+                static fn() => $connection
                     ->close()
                     ->either()
                     ->leftMap(static fn() => Failure::toCloseConnection()),

diff --git a/src/Client/State.php b/src/Client/State.php
@@ -3,34 +3,22 @@
 
 namespace Innmind\AMQP\Client;
 
-use Innmind\AMQP\Transport\Connection;
-
-/**
- * @internal
- */
 final class State
 {
-    private Connection $connection;
-    private mixed $userState;
-
-    private function __construct(Connection $connection, mixed $userState)
-    {
-        $this->connection = $connection;
-        $this->userState = $userState;
-    }
+    private mixed $value;
 
-    public static function of(Connection $connection, mixed $userState): self
+    private function __construct(mixed $value)
     {
-        return new self($connection, $userState);
+        $this->value = $value;
     }
 
-    public function connection(): Connection
+    public static function of(mixed $value): self
     {
-        return $this->connection;
+        return new self($value);
     }
 
-    public function userState(): mixed
+    public function unwrap(): mixed
     {
-        return $this->userState;
+        return $this->value;
     }
 }
diff --git a/src/Command.php b/src/Command.php
@@ -19,6 +19,6 @@ public function __invoke(
         Connection $connection,
         Channel $channel,
         MessageReader $read,
-        mixed $state,
+        Client\State $state,
     ): Either;
 }
diff --git a/src/Command/Bind.php b/src/Command/Bind.php
@@ -9,11 +9,15 @@
     Failure,
     Transport\Connection,
     Transport\Connection\MessageReader,
+    Transport\Protocol,
     Transport\Frame\Channel,
     Transport\Frame\Method,
     Model\Queue\Binding,
 };
-use Innmind\Immutable\Either;
+use Innmind\Immutable\{
+    Either,
+    Sequence,
+};
 
 final class Bind implements Command
 {
@@ -28,17 +32,20 @@ public function __invoke(
         Connection $connection,
         Channel $channel,
         MessageReader $read,
-        mixed $state,
+        State $state,
     ): Either {
-        /** @var Either<Failure, State> */
-        return $connection
-            ->send(fn($protocol) => $protocol->queue()->bind(
-                $channel,
-                $this->command,
-            ))
-            ->maybeWait($this->command->shouldWait(), Method::queueBindOk)
-            ->connection()
-            ->map(static fn($connection) => State::of($connection, $state))
+        $frames = fn(Protocol $protocol): Sequence => $protocol->queue()->bind(
+            $channel,
+            $this->command,
+        );
+
+        $sideEffect = match ($this->command->shouldWait()) {
+            true => $connection->request($frames, Method::queueBindOk),
+            false => $connection->send($frames),
+        };
+
+        return $sideEffect
+            ->map(static fn() => $state)
             ->leftMap(fn() => Failure::toBind($this->command));
     }