Connection blocked/unblocked support

.php-cs-fixer.dist.php

@@ -7,7 +7,7 @@
 use PhpCsFixer\Runner\Parallel\ParallelConfigFactory;
 use PHPyh\CodingStandard\PhpCsFixerCodingStandard;
 
-$config = (new Config())
+$config = new Config()
     ->setFinder(
         Finder::create()
             ->in(__DIR__ . '/src')
@@ -20,7 +20,7 @@
     ->setParallelConfig(ParallelConfigFactory::detect())
     ->setCacheFile(__DIR__ . '/var/' . basename(__FILE__) . '.cache');
 
-(new PhpCsFixerCodingStandard())->applyTo($config, [
+new PhpCsFixerCodingStandard()->applyTo($config, [
     'numeric_literal_separator' => false,
 ]);
 

docker-compose.yaml

@@ -14,3 +14,12 @@ services:
     ports:
       - "127.0.0.1:5672:5672"
       - "127.0.0.1:15672:15672"
+
+  rabbitmq_low_memory:
+    image: rabbitmq:4.0-management-alpine
+    restart: always
+    ports:
+      - "127.0.0.1:5673:5672"
+      - "127.0.0.1:15673:15672"
+    volumes:
+      - ./docker/rabbitmq/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf

docker/rabbitmq/rabbitmq.conf

@@ -0,0 +1 @@
+vm_memory_high_watermark.absolute = 130MiB

src/Channel.php

@@ -86,6 +86,8 @@ public function publish(
         bool $mandatory = false,
         bool $immediate = false,
     ): ?PublishConfirmation {
+        $this->connection->ensureNotBlocked();
+
         /** @var ?PublishConfirmation $confirmation */
         $confirmation = null;
 
@@ -110,6 +112,8 @@ public function publish(
      */
     public function publishBatch(array $publishMessages): PublishBatchConfirmation
     {
+        $this->connection->ensureNotBlocked();
+
         /** @var list<PublishConfirmation> $confirmations */
         $confirmations = [];
 

src/Exception/ConnectionIsBlocked.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Amqp\Exception;
+
+use Thesis\Amqp\AmqpException;
+
+/**
+ * @api
+ */
+final class ConnectionIsBlocked extends \RuntimeException implements AmqpException
+{
+    public function __construct(
+        public readonly string $reason,
+        ?\Throwable $previous = null,
+    ) {
+        parent::__construct(
+            message: 'Connection is blocked: ' . $reason,
+            previous: $previous,
+        );
+    }
+}

src/Internal/Io/AmqpConnection.php

@@ -8,6 +8,7 @@
 use Amp\Socket\Socket;
 use Revolt\EventLoop;
 use Thesis\AmpBridge\ReaderWriter as AmpReaderWriter;
+use Thesis\Amqp\Exception\ConnectionIsBlocked;
 use Thesis\Amqp\Exception\UnexpectedFrameReceived;
 use Thesis\Amqp\Internal\Hooks;
 use Thesis\Amqp\Internal\Protocol;
@@ -29,10 +30,16 @@ final class AmqpConnection implements Writer
 
     private float $lastWrite = 0;
 
+    /**
+     * @phpstan-ignore property.unusedType
+     */
+    private ?string $blockedReason = null;
+
     private bool $closed = false;
 
     public function __construct(
         private readonly Socket $socket,
+        private readonly Hooks $hooks,
     ) {
         $this->buffer = Buffer::empty();
         $this->reader = new Protocol\Reader(
@@ -91,8 +98,28 @@ public function write(string $bytes): void
         $this->lastWrite = Amp\now();
     }
 
-    public function ioLoop(Hooks $hooks): void
+    public function setup(): void
     {
+        $hooks = $this->hooks;
+
+        $blockedReason = &$this->blockedReason;
+
+        $hooks->anyOf(
+            0,
+            Protocol\Frame\ConnectionBlocked::class,
+            static function (Protocol\Frame\ConnectionBlocked $blocked) use (&$blockedReason): void {
+                $blockedReason = $blocked->reason;
+            },
+        );
+
+        $hooks->anyOf(
+            0,
+            Protocol\Frame\ConnectionUnblocked::class,
+            static function () use (&$blockedReason): void {
+                $blockedReason = null;
+            },
+        );
+
         $reader = $this->reader;
         $isClosed = &$this->closed;
 
@@ -127,6 +154,13 @@ public function heartbeat(int $interval): void
         });
     }
 
+    public function ensureNotBlocked(): void
+    {
+        if ($this->blockedReason !== null) {
+            throw new ConnectionIsBlocked($this->blockedReason);
+        }
+    }
+
     public function close(): void
     {
         if (!$this->socket->isClosed()) {

src/Internal/Io/AmqpConnectionFactory.php

@@ -61,7 +61,7 @@ public function connect(): AmqpConnection
             Frame\ConnectionOpenOk::class,
         );
 
-        $connection->ioLoop($this->hooks);
+        $connection->setup();
 
         $this->hooks->anyOf(0, Frame\ConnectionClose::class, static function () use ($connection): void {
             $connection->writeFrame(Protocol\Method::connectionCloseOk());
@@ -93,7 +93,7 @@ private function createConnection(): AmqpConnection
 
         foreach ($this->config->connectionUrls() as $url) {
             try {
-                return new AmqpConnection($this->createSocket($url));
+                return new AmqpConnection($this->createSocket($url), $this->hooks);
             } catch (\Throwable $e) {
                 $exceptions[] = "{$url}: {$e->getMessage()}";
             }
@@ -109,7 +109,7 @@ private function createConnection(): AmqpConnection
      */
     private function createSocket(string $url): Socket\Socket
     {
-        $context = (new Socket\ConnectContext())
+        $context = new Socket\ConnectContext()
             ->withConnectTimeout($this->config->connectionTimeout);
 
         if ($this->config->tcpNoDelay) {

src/Internal/Protocol/Frame/ConnectionBlocked.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Amqp\Internal\Protocol\Frame;
+
+use Thesis\Amqp\Internal\Io;
+use Thesis\Amqp\Internal\Protocol\Frame;
+
+/**
+ * @internal
+ */
+final readonly class ConnectionBlocked implements Frame
+{
+    public function __construct(
+        public string $reason,
+    ) {}
+
+    public static function read(Io\ReadBytes $reader): self
+    {
+        return new self($reader->readString());
+    }
+
+    public function write(Io\WriteBytes $writer): void
+    {
+        $writer->writeString($this->reason);
+    }
+}

src/Internal/Protocol/Frame/ConnectionUnblocked.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Amqp\Internal\Protocol\Frame;
+
+use Thesis\Amqp\Internal\Io;
+use Thesis\Amqp\Internal\Protocol\Frame;
+
+/**
+ * @internal
+ */
+enum ConnectionUnblocked implements Frame
+{
+    case frame;
+
+    public static function read(Io\ReadBytes $reader): self
+    {
+        return self::frame;
+    }
+
+    public function write(Io\WriteBytes $writer): void {}
+}

src/Internal/Protocol/Protocol.php

@@ -25,6 +25,8 @@ enum Protocol
             ClassMethod::CONNECTION_OPEN_OK => Frame\ConnectionOpenOk::class,
             ClassMethod::CONNECTION_CLOSE => Frame\ConnectionClose::class,
             ClassMethod::CONNECTION_CLOSE_OK => Frame\ConnectionCloseOk::class,
+            ClassMethod::CONNECTION_BLOCKED => Frame\ConnectionBlocked::class,
+            ClassMethod::CONNECTION_UNBLOCKED => Frame\ConnectionUnblocked::class,
         ],
         ClassType::CHANNEL => [
             ClassMethod::CHANNEL_OPEN_OK => Frame\ChannelOpenOkFrame::class,