Merge pull request #214 from jakubkulhan/0.6.x-make-connection-take-clientinterface

[0.6.x] Make connection take ClientInterface

Author: WyriHaximus

spec/generate.php

@@ -230,6 +230,7 @@ function amqpTypeToLength(string $type, string $e): array
 $connectionContent .= "use Bunny\\Protocol\\MethodFrame;\n";
 $connectionContent .= "use Bunny\\Protocol\\ProtocolReader;\n";
 $connectionContent .= "use Bunny\\Protocol\\ProtocolWriter;\n";
+$connectionContent .= "use Closure;\n";
 $connectionContent .= "use Evenement\\EventEmitterInterface;\n";
 $connectionContent .= "use Evenement\\EventEmitterTrait;\n";
 $connectionContent .= "use React\\EventLoop\\Loop;\n";
@@ -270,15 +271,17 @@ function amqpTypeToLength(string $type, string $e): array
 $connectionContent .= "    /** @var array<array{filter: (callable(\Bunny\Protocol\AbstractFrame): bool), promise: \React\Promise\Deferred<\Bunny\Protocol\AbstractFrame>}> */\n";
 $connectionContent .= "    private array \$awaitList = [];\n";
 $connectionContent .= "\n";
+$connectionContent .= "    /** @param (Closure(): int) \$frameMax */\n";
 $connectionContent .= "    public function __construct(\n";
-$connectionContent .= "        private readonly Client \$client,\n";
+$connectionContent .= "        private readonly ClientInterface \$client,\n";
 $connectionContent .= "        private readonly ConnectionInterface \$connection,\n";
 $connectionContent .= "        private readonly Buffer \$readBuffer,\n";
 $connectionContent .= "        private readonly Buffer \$writeBuffer,\n";
 $connectionContent .= "        private readonly ProtocolReader \$reader,\n";
 $connectionContent .= "        private readonly ProtocolWriter \$writer,\n";
 $connectionContent .= "        private readonly Channels \$channels,\n";
 $connectionContent .= "        private readonly Configuration \$configuration,\n";
+$connectionContent .= "        private readonly Closure \$frameMax,\n";
 $connectionContent .= "    ) {\n";
 $connectionContent .= "        \$this->connection->on('data', function (string \$data): void {\n";
 $connectionContent .= "            \$this->readBuffer->append(\$data);\n";
@@ -899,7 +902,7 @@ function amqpTypeToLength(string $type, string $e): array
                     $connectionContent .= "        }\n\n";
                 }
 
-                $connectionContent .= "        for (\$payloadMax = \$this->client->frameMax - 8 /* frame preface and frame end */, \$i = 0, \$l = strlen(\$body); \$i < \$l; \$i += \$payloadMax) {\n";
+                $connectionContent .= "        for (\$payloadMax = (\$this->frameMax)() - 8 /* frame preface and frame end */, \$i = 0, \$l = strlen(\$body); \$i < \$l; \$i += \$payloadMax) {\n";
                 $connectionContent .= "            \$payloadSize = \$l - \$i;\n";
                 $connectionContent .= "            if (\$payloadSize > \$payloadMax) {\n";
                 $connectionContent .= "                \$payloadSize = \$payloadMax;\n";

src/Client.php

@@ -56,11 +56,11 @@ class Client implements ClientInterface, EventEmitterInterface
 
     private Channels $channels;
 
-    public int $frameMax = 0xFFFF;
+    public int $frameMax = Constants::FRAME_MAX;
 
     private int $nextChannelId = 1;
 
-    private int $channelMax = 0xFFFF;
+    private int $channelMax = Constants::CHANNEL_MAX;
 
     /**
      * @var list<\React\Promise\Deferred<null>>
@@ -209,6 +209,9 @@ public function connect(): self
                 new ProtocolWriter(),
                 $this->channels,
                 $this->configuration,
+                function (): int {
+                    return $this->channelMax;
+                },
             );
             $this->connection->on('error', function (Throwable $error): void {
                 $this->emit('error', [$error]);

src/Connection.php

@@ -14,6 +14,7 @@
 use Bunny\Protocol\MethodFrame;
 use Bunny\Protocol\ProtocolReader;
 use Bunny\Protocol\ProtocolWriter;
+use Closure;
 use Evenement\EventEmitterInterface;
 use Evenement\EventEmitterTrait;
 use React\EventLoop\Loop;
@@ -54,15 +55,17 @@ final class Connection implements EventEmitterInterface
     /** @var array<array{filter: (callable(\Bunny\Protocol\AbstractFrame): bool), promise: \React\Promise\Deferred<\Bunny\Protocol\AbstractFrame>}> */
     private array $awaitList = [];
 
+    /** @param (Closure(): int) $frameMax */
     public function __construct(
-        private readonly Client $client,
+        private readonly ClientInterface $client,
         private readonly ConnectionInterface $connection,
         private readonly Buffer $readBuffer,
         private readonly Buffer $writeBuffer,
         private readonly ProtocolReader $reader,
         private readonly ProtocolWriter $writer,
         private readonly Channels $channels,
         private readonly Configuration $configuration,
+        private readonly Closure $frameMax,
     ) {
         $this->connection->on('data', function (string $data): void {
             $this->readBuffer->append($data);
@@ -1699,7 +1702,7 @@ public function publish(int $channel, string $body, array $headers = [], string
             }
         }
 
-        for ($payloadMax = $this->client->frameMax - 8 /* frame preface and frame end */, $i = 0, $l = strlen($body); $i < $l; $i += $payloadMax) {
+        for ($payloadMax = ($this->frameMax)() - 8 /* frame preface and frame end */, $i = 0, $l = strlen($body); $i < $l; $i += $payloadMax) {
             $payloadSize = $l - $i;
             if ($payloadSize > $payloadMax) {
                 $payloadSize = $payloadMax;

src/Constants.php

@@ -24,6 +24,10 @@ final class Constants
 
     public const FRAME_END = 0xCE;
 
+    public const FRAME_MAX = 0xFFFF;
+
+    public const CHANNEL_MAX = 0xFFFF;
+
     // connection channel
     public const CONNECTION_CHANNEL = 0;
 

test/ConnectionTest.php

@@ -66,6 +66,9 @@ public function testThrowOn(): void
             new ProtocolWriter(),
             new Channels(),
             new Configuration(),
+            static function (): int {
+                return Constants::FRAME_MAX;
+            },
         );
         $deferred = new Deferred();
         Loop::addTimer(0.1, async(static function () use ($deferred, $connection): void {
@@ -108,6 +111,9 @@ public function testNoMethodConnectionCloseOkFrameIsWrittenWhenConnectionIsClose
             new ProtocolWriter(),
             new Channels(),
             new Configuration(),
+            static function (): int {
+                return Constants::FRAME_MAX;
+            },
         );
         $baseBuffer = $mockConnection->getWrittenData();
         self::assertSame('', $baseBuffer);
@@ -152,6 +158,9 @@ public function testNoMethodConnectionCloseOkFrameIsWrittenWhenConnectionIsClose
             new ProtocolWriter(),
             new Channels(),
             new Configuration(),
+            static function (): int {
+                return Constants::FRAME_MAX;
+            },
         );
         $baseBuffer = $mockConnection->getWrittenData();
         self::assertSame('', $baseBuffer);