Add connection blocked/unblocked support

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_low_memory/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf

docker/rabbitmq_low_memory/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/Client.php

@@ -31,14 +31,16 @@ final class Client
 
     public function __construct(
         public readonly Config $config,
+        public readonly EventDispatcher $eventDispatcher = new EventDispatcher(),
     ) {
         $properties = Properties::createDefault();
         $this->hooks = new Hooks();
 
         $this->connectionFactory = new AmqpConnectionFactory(
-            $this->config,
-            $properties,
-            $this->hooks,
+            config: $this->config,
+            properties: $properties,
+            hooks: $this->hooks,
+            eventDispatcher: $this->eventDispatcher,
         );
 
         $this->channelFactory = new ChannelFactory(

src/Event/ConnectionBlocked.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Amqp\Event;
+
+final readonly class ConnectionBlocked
+{
+    public function __construct(
+        public string $reason,
+    ) {}
+}

src/Event/ConnectionUnblocked.php

@@ -0,0 +1,7 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Amqp\Event;
+
+final readonly class ConnectionUnblocked {}

src/EventDispatcher.php

@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Amqp;
+
+use Revolt\EventLoop;
+
+/**
+ * @api
+ * @phpstan-type Events = Event\ConnectionBlocked|Event\ConnectionUnblocked
+ */
+final class EventDispatcher
+{
+    /**
+     * @var array<class-string<Events>, non-empty-list<callable(Events): void>>
+     */
+    private array $listeners = [];
+
+    /**
+     * @template TEvent of Events
+     * @param class-string<TEvent>|non-empty-list<class-string<TEvent>> $events
+     * @param callable(TEvent): void $listener
+     */
+    public function listen(string|array $events, callable $listener): self
+    {
+        $dispatcher = clone $this;
+
+        foreach ((array) $events as $event) {
+            /** @phpstan-ignore assign.propertyType */
+            $dispatcher->listeners[$event][] = $listener;
+        }
+
+        return $dispatcher;
+    }
+
+    /**
+     * @param Events $event
+     */
+    public function dispatch(object $event): void
+    {
+        foreach ($this->listeners[$event::class] ?? [] as $listener) {
+            /** @phpstan-ignore argument.type */
+            EventLoop::queue($listener, $event);
+        }
+    }
+}

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,10 @@
 use Amp\Socket\Socket;
 use Revolt\EventLoop;
 use Thesis\AmpBridge\ReaderWriter as AmpReaderWriter;
+use Thesis\Amqp\Event\ConnectionBlocked;
+use Thesis\Amqp\Event\ConnectionUnblocked;
+use Thesis\Amqp\EventDispatcher;
+use Thesis\Amqp\Exception\ConnectionIsBlocked;
 use Thesis\Amqp\Exception\UnexpectedFrameReceived;
 use Thesis\Amqp\Internal\Hooks;
 use Thesis\Amqp\Internal\Protocol;
@@ -29,10 +33,17 @@ 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,
+        private readonly EventDispatcher $eventDispatcher,
     ) {
         $this->buffer = Buffer::empty();
         $this->reader = new Protocol\Reader(
@@ -91,8 +102,31 @@ public function write(string $bytes): void
         $this->lastWrite = Amp\now();
     }
 
-    public function ioLoop(Hooks $hooks): void
+    public function setup(): void
     {
+        $hooks = $this->hooks;
+        $eventDispatcher = $this->eventDispatcher;
+
+        $blockedReason = &$this->blockedReason;
+
+        $hooks->anyOf(
+            0,
+            Protocol\Frame\ConnectionBlocked::class,
+            static function (Protocol\Frame\ConnectionBlocked $blocked) use (&$blockedReason, $eventDispatcher): void {
+                $blockedReason = $blocked->reason;
+                $eventDispatcher->dispatch(new ConnectionBlocked($blockedReason));
+            },
+        );
+
+        $hooks->anyOf(
+            0,
+            Protocol\Frame\ConnectionUnblocked::class,
+            static function () use (&$blockedReason, $eventDispatcher): void {
+                $blockedReason = null;
+                $eventDispatcher->dispatch(new ConnectionUnblocked());
+            },
+        );
+
         $reader = $this->reader;
         $isClosed = &$this->closed;
 
@@ -127,6 +161,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

@@ -7,6 +7,7 @@
 use Amp\Cancellation;
 use Amp\Socket;
 use Thesis\Amqp\Config;
+use Thesis\Amqp\EventDispatcher;
 use Thesis\Amqp\Exception;
 use Thesis\Amqp\Internal\Hooks;
 use Thesis\Amqp\Internal\Properties;
@@ -24,6 +25,7 @@ public function __construct(
         private Config $config,
         private Properties $properties,
         private Hooks $hooks,
+        private EventDispatcher $eventDispatcher,
     ) {}
 
     public function connect(): AmqpConnection
@@ -61,7 +63,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 +95,11 @@ private function createConnection(): AmqpConnection
 
         foreach ($this->config->connectionUrls() as $url) {
             try {
-                return new AmqpConnection($this->createSocket($url));
+                return new AmqpConnection(
+                    socket: $this->createSocket($url),
+                    hooks: $this->hooks,
+                    eventDispatcher: $this->eventDispatcher,
+                );
             } catch (\Throwable $e) {
                 $exceptions[] = "{$url}: {$e->getMessage()}";
             }

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,

tests/RabbitMQTest.php

@@ -0,0 +1,63 @@
+<?php
+
+declare(strict_types=1);
+
+use Amp\DeferredFuture;
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\Attributes\DoesNotPerformAssertions;
+use PHPUnit\Framework\TestCase;
+use Thesis\Amqp\Channel;
+use Thesis\Amqp\Client;
+use Thesis\Amqp\Config;
+use Thesis\Amqp\Event\ConnectionBlocked;
+use Thesis\Amqp\Event\ConnectionUnblocked;
+use Thesis\Amqp\EventDispatcher;
+use Thesis\Amqp\Exception\ConnectionIsBlocked;
+use Thesis\Amqp\Message;
+
+#[CoversClass(Client::class)]
+#[CoversClass(Channel::class)]
+final class RabbitMQTest extends TestCase
+{
+    #[DoesNotPerformAssertions]
+    public function testConnectionBlockedUnblockedOnLowMemory(): void
+    {
+        self::markTestSkipped();
+
+        /** @phpstan-ignore deadCode.unreachable */
+        $deferredBlocked = new DeferredFuture();
+        $deferredUnblocked = new DeferredFuture();
+
+        $publishClient = new Client(
+            config: new Config(urls: ['localhost:5673']),
+            eventDispatcher: (new EventDispatcher())
+                ->listen(ConnectionBlocked::class, static function () use ($deferredBlocked): void {
+                    $deferredBlocked->complete();
+                })
+                ->listen(ConnectionUnblocked::class, static function () use ($deferredUnblocked): void {
+                    $deferredUnblocked->complete();
+                }),
+        );
+        $publishChannel = $publishClient->channel();
+
+        $fixClient = new Client($publishClient->config);
+        $fixChannel = $fixClient->channel();
+
+        $queue = $publishChannel->queueDeclare(autoDelete: true);
+
+        $message = new Message(body: str_repeat('x', 1024 * 10));
+
+        try {
+            while (true) {
+                $publishChannel->publish($message, routingKey: $queue->name);
+            }
+        } catch (ConnectionIsBlocked) {
+        }
+
+        $deferredBlocked->getFuture()->await();
+
+        $fixChannel->queueDelete($queue->name);
+
+        $deferredUnblocked->getFuture()->await();
+    }
+}