feat!: upgrade to bunny 0.6

composer.json

@@ -5,12 +5,15 @@
     "type": "library",
     "require": {
         "php": "^8.3",
-        "bunny/bunny": "^0.5.0",
+        "bunny/bunny": "^0.6.0-alpha.4",
         "myclabs/php-enum": "^1.8.4",
-        "react/promise": "^2.0",
+        "react/async": "^4.2",
+        "react/event-loop": "^1.2",
+        "react/promise": "^3",
         "symfony/config": "^7.4",
         "symfony/console": "^7.4",
         "symfony/dependency-injection": "^7.4",
+        "symfony/event-dispatcher": "^7.4",
         "symfony/framework-bundle": "^7.4",
         "symfony/http-kernel": "^7.4",
         "symfony/yaml": "^7.4"
@@ -26,6 +29,8 @@
         "phpunit/phpunit": "^12.5",
         "shipmonk/composer-dependency-analyser": "^1.4"
     },
+    "minimum-stability": "alpha",
+    "prefer-stable": true,
     "autoload": {
         "psr-4": {
             "Cdn77\\RabbitMQBundle\\": "src/"

docs/Consuming.md

@@ -65,4 +65,28 @@ Consumer configuration is done via `getConfiguration()` that returns instance of
 
 `maxMessages` & `maxSeconds` parameters are recommended to set to something else than `null` so consumer will shutdown gracefully after specified number of messages is consumed or seconds elapsed so it can start with fresh memory again.
 
+Both limits end the run between messages, never in the middle of one: a `consume()` that is still
+running when `maxSeconds` falls due is left to finish, and its acknowledge goes out on a connection
+that is still open.
+
 Consumer is registered under the name specified in `getName()` method. You can check whether it is successfully registered through `debug:rabbitmq:consumers` command. It can be run with `rabbitmq:consumer:run example_consumer`
+
+> **Heads up (Bunny 0.6): a `consume()` that blocks for longer than the heartbeat loses the
+> connection.** The event loop only turns while something awaits it, and a handler is ordinary
+> synchronous PHP - a subprocess, an SFTP upload, a long HTTP call - so no heartbeat frame can be
+> written for as long as it runs. The broker hangs up after two missed intervals, and the consumer
+> then dies with `Exception\ConnectionFailed` saying the channel closed with no error reported.
+>
+> Worse than the exception: an acknowledge issued *after* that point is a write with no reply to
+> wait for, so it silently goes nowhere. The message stays unacknowledged, the broker requeues it,
+> and the handler runs again on the next consumer - work you may have thought was committed.
+>
+> A message *published* from inside a handler does not share that fate: the bundle puts it on the
+> socket before `handle()` returns, so a handler that publishes and then blocks still gets it out.
+> The two are therefore not atomic - if the connection goes while the handler works on, the
+> published message stands while the consumed one is requeued and handled again.
+>
+> So a handler that can take minutes needs a `heartbeat` comfortably above twice its worst case
+> (`heartbeat: 3600` for handlers measured in minutes), or it needs to stop blocking the loop.
+> Acknowledging *before* the long stretch, rather than after it, also keeps the acknowledge on a
+> connection that is still alive - at the cost of at-most-once delivery for that message.

docs/Installation.md

@@ -6,6 +6,32 @@ Add as a dependency via Composer:
 composer require cdn77/rabbitmq-bundle
 ```
 
+> **Note:** This bundle requires `bunny/bunny` `0.6`, which is currently only available as a
+> pre-release (`alpha`). Until a stable `0.6.0` is tagged, your project must allow that stability,
+> e.g. by setting in your `composer.json`:
+>
+> ```json
+> {
+>     "minimum-stability": "alpha",
+>     "prefer-stable": true
+> }
+> ```
+>
+> Bunny `0.6` runs on a [ReactPHP](https://reactphp.org/) event loop using PHP Fibers, so the
+> bundle pulls in `react/event-loop`, `react/async` and `react/promise` as well. All broker I/O is
+> driven synchronously for you — see [Producing](Producing.md) for the one caveat in long-lived CLI
+> scripts.
+
+> **Upgrading from a Bunny `0.5` release of this bundle:** Bunny `0.6` has no read/write timeout, so
+> the `read_write_timeout` option is gone. A `read_write_timeout` query parameter in your DSN is
+> simply ignored, but the YAML key now fails the container build with
+> `Unrecognized option "read_write_timeout" under "rabbitmq"` — drop it from your configuration.
+>
+> Note also that Bunny `0.6` cannot recover within the same process from a *first* connection
+> attempt that failed (the broker being unreachable at start-up); every later attempt on that
+> connection reports `ConnectionFailed`. Losing an already established connection is handled and
+> reconnects on the next operation.
+
 If you're not using Symfony Flex, you will also need to enable the bundle by adding `Cdn77RabbitMQBundle` to `bundles.php`, that is required by `registerBundles()` in your `Kernel`:
 
 ```php

docs/Producing.md

@@ -9,11 +9,15 @@ use Cdn77\RabbitMQBundle\RabbitMQ\Operation\PublishOperation;
 
 final class ExampleProducer
 {
+    /** @var Connection */
+    private $connection;
+
     /** @var PublishOperation */
     private $publishOperation;
 
     public function __construct(Connection $connection, PublishOperation $publishOperation)
     {
+        $this->connection = $connection;
         $this->publishOperation = $publishOperation;
     }
 
@@ -28,11 +32,27 @@ final class ExampleProducer
         $message->makeTransient();
 
         $this->publishOperation->handle(
-            $this->connection
+            $this->connection,
             $message,
             $routingKey,
             'default_exchange', // Exchange to send the message to
         );
     }
 }
 ```
+
+> **Heads up (Bunny 0.6):** while connected, Bunny keeps a heartbeat timer on the ReactPHP event
+> loop, which would otherwise keep the php-fpm worker or console process alive after the work is
+> done. The bundle ships a `DisconnectConnection` subscriber that closes the connection on
+> `kernel.terminate` (after the HTTP response is flushed) and on `console.terminate` (after a
+> command returns), so HTTP and console producers are handled automatically. If you publish from a
+> context that dispatches neither event (e.g. a bare script using the event loop directly), call
+> `$connection->disconnect()` yourself when you're done.
+
+> **Heads up (Bunny 0.6):** a single `handle()` is fire-and-forget - the broker sends no reply to
+> wait for - so it tells you the message was written to the socket, not that the broker has it. The
+> bundle does make sure of the writing: publishing awaits nothing, so the event loop would otherwise
+> never turn and the message would sit in a buffer until some later operation happened to flush it,
+> which for a producer of ordinary messages could be never. Use `handleAll()` when you need to know
+> the broker has the messages before you carry on: it publishes in a transaction and returns once
+> the commit is confirmed.

docs/Setup.md

@@ -4,13 +4,35 @@ This bundle uses `rabbitmq` container extension key and is able to merge configu
 
 These configuration options are available to setup connection to your RabbitMQ instance. 
 
-Example DSN: `amqp://username:password@host:1234/vhost?heartbeat=60&connection_timeout=10&read_write_timeout=3`
+Example DSN: `amqp://username:password@host:1234/vhost?heartbeat=60&connection_timeout=10&operation_timeout=30`
 
 ```yaml
 rabbitmq:
     dsn: '%env(RABBITMQ_DSN)%'
+    heartbeat: 60           # seconds; also as a DSN parameter, where it wins over this
+    connection_timeout: 10  # seconds to establish the connection
+    operation_timeout: 30   # seconds a single broker operation may take before it is given up on
 ```
 
+`operation_timeout` bounds every single operation - publishing, acknowledging, declaring topology,
+connecting - but never the consume loop, which runs until the consumer's own message or time limit
+is reached. Exceeding it replaces the connection and throws `Exception\OperationFailed`, rather than
+leaving the process waiting for a broker that is not going to answer. Where the caller reports
+failures in its own terms it keeps doing so, carrying the `OperationFailed` as the cause: connecting
+throws `Exception\ConnectionFailed`, and `rabbitmq:setup` throws `Exception\ConfigurationFailed`
+naming the exchange or queue it got stuck on. All of them implement `Exception\Exception`. Any value
+of `0` or below disables the bound, at the risk of a process that can never finish.
+
+A connection idle for longer than `heartbeat` is replaced before the next operation: the event loop
+only turns while an operation is awaiting, so nothing can send a heartbeat frame in between, and the
+broker will have closed such a connection already.
+
+`heartbeat` has to be a positive number of seconds; `0`, which in AMQP switches heartbeats off, is
+rejected. Bunny 0.6 arms the timer whatever the interval is and re-arms it with the same value, so
+zero leaves a timer that is due again the moment it fires - spinning the event loop at a full core
+for as long as any operation is awaiting, and flooding the broker with heartbeat frames. Configure a
+long interval (say `3600`) if heartbeats are genuinely not wanted.
+
 Exchanges and Queues configuration can be done this way
 
 ```yaml

phpstan-baseline.neon

@@ -3,7 +3,7 @@ parameters:
 		-
 			message: '#^Cannot cast mixed to int\.$#'
 			identifier: cast.int
-			count: 3
+			count: 2
 			path: src/Configuration/Connection.php
 
 		-
@@ -55,7 +55,7 @@ parameters:
 			path: src/RabbitMQ/Binding.php
 
 		-
-			message: '#^Parameter \#3 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Binding constructor expects array\<mixed\>, mixed given\.$#'
+			message: '#^Parameter \#3 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Binding constructor expects array\<string, mixed\>, mixed given\.$#'
 			identifier: argument.type
 			count: 1
 			path: src/RabbitMQ/Binding.php
@@ -85,7 +85,7 @@ parameters:
 			path: src/RabbitMQ/Exchange.php
 
 		-
-			message: '#^Parameter \#6 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Exchange constructor expects array\<mixed\>, mixed given\.$#'
+			message: '#^Parameter \#6 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Exchange constructor expects array\<string, mixed\>, mixed given\.$#'
 			identifier: argument.type
 			count: 1
 			path: src/RabbitMQ/Exchange.php
@@ -109,7 +109,7 @@ parameters:
 			path: src/RabbitMQ/Queue.php
 
 		-
-			message: '#^Parameter \#5 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Queue constructor expects array\<mixed\>, mixed given\.$#'
+			message: '#^Parameter \#5 \$arguments of class Cdn77\\RabbitMQBundle\\RabbitMQ\\Queue constructor expects array\<string, mixed\>, mixed given\.$#'
 			identifier: argument.type
 			count: 1
 			path: src/RabbitMQ/Queue.php

phpunit.xml.dist

@@ -8,7 +8,7 @@
          bootstrap="tests/bootstrap.php"
 >
     <php>
-        <env name="RABBITMQ_DSN" value="amqp://127.0.0.1/?hearbeat=60&amp;connection_timeout=10&amp;read_write_timeout=3" />
+        <env name="RABBITMQ_DSN" value="amqp://127.0.0.1/?heartbeat=60&amp;connection_timeout=10" />
     </php>
     <testsuite name="Tests">
         <directory>tests</directory>

src/Configuration/Connection.php

@@ -5,12 +5,15 @@
 namespace Cdn77\RabbitMQBundle\Configuration;
 
 use Cdn77\RabbitMQBundle\DependencyInjection\Configuration;
+use Cdn77\RabbitMQBundle\Exception\ConfigurationFailed;
+
+use function is_numeric;
 
 final class Connection
 {
     private const int DEFAULT_HEARTBEAT = 60;
     private const int DEFAULT_CONNECTION_TIMEOUT = 3;
-    private const int DEFAULT_READ_WRITE_TIMEOUT = 5;
+    private const float DEFAULT_OPERATION_TIMEOUT = 30.0;
 
     /** @var string */
     private $host;
@@ -33,8 +36,8 @@
     /** @var int */
     private $connectionTimeout;
 
-    /** @var int */
-    private $readWriteTimeout;
+    /** @var float */
+    private $operationTimeout;
 
     public function __construct(
         string $host,
@@ -44,16 +47,16 @@
         string|null $password,
         int $heartbeat = self::DEFAULT_HEARTBEAT,
         int $connectionTimeout = self::DEFAULT_CONNECTION_TIMEOUT,
-        int $readWriteTimeout = self::DEFAULT_READ_WRITE_TIMEOUT,
+        float $operationTimeout = self::DEFAULT_OPERATION_TIMEOUT,
     ) {
         $this->host = $host;
         $this->port = $port;
         $this->vhost = $vhost;
         $this->user = $user;
         $this->password = $password;
-        $this->heartbeat = $heartbeat;
+        $this->heartbeat = self::validHeartbeat($heartbeat);
         $this->connectionTimeout = $connectionTimeout;
-        $this->readWriteTimeout = $readWriteTimeout;
+        $this->operationTimeout = $operationTimeout;
     }
 
     /** @param mixed[] $configuration */
@@ -63,24 +66,29 @@
         $new = self::fromDsn($dsn);
 
         if (
             isset($configuration[Configuration::KEY_CONFIGURATION_HEARTBEAT])
             && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_HEARTBEAT])
         ) {
-            $new->heartbeat = (int) $configuration[Configuration::KEY_CONFIGURATION_HEARTBEAT];
+            $new->heartbeat = self::validHeartbeat(
+                (int) $configuration[Configuration::KEY_CONFIGURATION_HEARTBEAT],
+            );
         }
 
         if (
             isset($configuration[Configuration::KEY_CONFIGURATION_CONNECTION_TIMEOUT])
             && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_CONNECTION_TIMEOUT])
         ) {
             $new->connectionTimeout = (int) $configuration[Configuration::KEY_CONFIGURATION_CONNECTION_TIMEOUT];
         }
 
+        // Only a number, never a blind cast: garbage would become 0.0, which is how the timeout
+        // is switched off - the one value nobody means to configure by accident.
+        $operationTimeout = $configuration[Configuration::KEY_CONFIGURATION_OPERATION_TIMEOUT] ?? null;
         if (
-            isset($configuration[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT])
-            && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT])
+            is_numeric($operationTimeout)
+            && ! isset($dsn->getParameters()[Configuration::KEY_CONFIGURATION_OPERATION_TIMEOUT])
         ) {
-            $new->readWriteTimeout = (int) $configuration[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT];
+            $new->operationTimeout = (float) $operationTimeout;
         }
 
         return $new;
@@ -89,6 +97,7 @@
     public static function fromDsn(Dsn $dsn): self
     {
         $parameters = $dsn->getParameters();
+        $operationTimeout = $parameters[Configuration::KEY_CONFIGURATION_OPERATION_TIMEOUT] ?? null;
 
         return new self(
             $dsn->getHost(),
@@ -99,8 +108,7 @@
             (int) ($parameters[Configuration::KEY_CONFIGURATION_HEARTBEAT] ?? self::DEFAULT_HEARTBEAT),
             (int) ($parameters[Configuration::KEY_CONFIGURATION_CONNECTION_TIMEOUT]
                 ?? self::DEFAULT_CONNECTION_TIMEOUT),
-            (int) ($parameters[Configuration::KEY_CONFIGURATION_READ_WRITE_TIMEOUT]
-                ?? self::DEFAULT_READ_WRITE_TIMEOUT),
+            is_numeric($operationTimeout) ? (float) $operationTimeout : self::DEFAULT_OPERATION_TIMEOUT,
         );
     }
 
@@ -139,8 +147,27 @@
         return $this->connectionTimeout;
     }
 
-    public function getReadWriteTimeout(): int
+    /** How long a single broker operation may take before it is given up on. Zero disables it. */
+    public function getOperationTimeout(): float
     {
-        return $this->readWriteTimeout;
+        return $this->operationTimeout;
+    }
+
+    /**
+     * Bunny 0.6.0-alpha.4 arms a heartbeat timer whatever the interval is and re-arms it with the
+     * same value, so 0 - the AMQP way of switching heartbeats off - leaves a timer that is due
+     * again the moment it fires. It then spins the event loop at a full core for the whole of every
+     * operation and floods the broker with heartbeat frames: measured at 0.78s of CPU for a
+     * one-second await, against 0.00s with an interval of 60. Rejected here rather than in
+     * BunnyConnection, so that a DSN parameter, a container key and a hand-built configuration are
+     * all covered - including the blind cast above, which turns any non-numeric value into a zero.
+     */
+    private static function validHeartbeat(int $heartbeat): int
+    {
+        if ($heartbeat > 0) {
+            return $heartbeat;
+        }
+
+        throw ConfigurationFailed::heartbeatMustBePositive($heartbeat);
     }
 }