From ba6bca974616c16009f6c166a7c5904a26ba26a2 Mon Sep 17 00:00:00 2001 From: Vitalii Zhuk Date: Mon, 25 May 2026 19:31:02 +0300 Subject: [PATCH 1/2] Allow only loop consume strategy for Spool --- src/Adapter/Amqp/Queue/AmqpQueueFactory.php | 14 +------ src/Adapter/AmqpLib/Queue/AmqpQueue.php | 3 +- .../AmqpLib/Queue/AmqpQueueFactory.php | 9 ---- src/Consumer/AbstractConsumer.php | 14 ------- src/Consumer/Spool/SpoolConsumer.php | 29 ++++++------- src/Queue/QueueFactoryInterface.php | 7 ---- .../Adapter/SpoolConsumerTestCase.php | 42 +++++++++++++------ .../ConsumerSignalsInConsumeStrategyTest.php | 5 ++- 8 files changed, 50 insertions(+), 73 deletions(-) diff --git a/src/Adapter/Amqp/Queue/AmqpQueueFactory.php b/src/Adapter/Amqp/Queue/AmqpQueueFactory.php index 3cf89ce..3705048 100644 --- a/src/Adapter/Amqp/Queue/AmqpQueueFactory.php +++ b/src/Adapter/Amqp/Queue/AmqpQueueFactory.php @@ -29,11 +29,6 @@ public function __construct( ) { } - public function withPassive(): QueueFactoryInterface - { - return new AmqpQueueFactory($this->channelFactory, $this->definition->withPassive(true)); - } - public function create(): QueueInterface { if ($this->queue) { @@ -59,14 +54,7 @@ public function create(): QueueInterface $queue->declareQueue(); - $amqpQueue = new AmqpQueue($channel, $queue); - - if ($this->definition->passive) { - return $amqpQueue; - } - - $this->queue = $amqpQueue; - + $this->queue = new AmqpQueue($channel, $queue); $channel->getConnection()->attach($this); foreach ($this->definition->bindings as $binding) { diff --git a/src/Adapter/AmqpLib/Queue/AmqpQueue.php b/src/Adapter/AmqpLib/Queue/AmqpQueue.php index f2c56d8..d506960 100644 --- a/src/Adapter/AmqpLib/Queue/AmqpQueue.php +++ b/src/Adapter/AmqpLib/Queue/AmqpQueue.php @@ -52,13 +52,14 @@ public function consume(\Closure $handler, string $tag = ''): void false, $this->definition->exclusive, false, - function (AMQPMessage $message) use ($handler, $queueName, &$stopConsuming): void { + function (AMQPMessage $message) use ($handler, $queueName, &$stopConsuming, &$amqplibChannel): void { $receivedMessage = new AmqpReceivedMessage($message, $queueName); $result = $handler($receivedMessage); if (false === $result) { $stopConsuming = true; + $amqplibChannel->basic_cancel($message->getConsumerTag(), false, true); } } ); diff --git a/src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php b/src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php index 3f5cdd2..64cbcf3 100644 --- a/src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php +++ b/src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php @@ -30,11 +30,6 @@ public function __construct( ) { } - public function withPassive(): QueueFactoryInterface - { - return new AmqpQueueFactory($this->channelFactory, $this->definition->withPassive(true)); - } - public function create(): QueueInterface { if ($this->queue) { @@ -55,10 +50,6 @@ public function create(): QueueInterface $queue = new AmqpQueue($channel, $this->definition); $queue->declare(); - if ($this->definition->passive) { - return $queue; - } - $connection->attach($this); foreach ($this->definition->bindings as $binding) { diff --git a/src/Consumer/AbstractConsumer.php b/src/Consumer/AbstractConsumer.php index 5f46d78..c359230 100644 --- a/src/Consumer/AbstractConsumer.php +++ b/src/Consumer/AbstractConsumer.php @@ -124,18 +124,4 @@ protected function doRun(bool $autoAck = true, ?\Closure $beforeCallback = null, } }, $this->configuration->tagGenerator->generate()); } - - /** - * Performs a graceful AMQP disconnect. - * - * A passively declared queue is created first to force a synchronous AMQP - * round-trip on the current channel. This guarantees that all previously - * sent frames (e.g. acknowledgements) were processed by the broker before - * the connection is closed. - */ - protected function gracefulDisconnect(): void - { - $queue = $this->queueFactory->withPassive()->create(); - $queue->getChannel()->getConnection()->disconnect(); - } } diff --git a/src/Consumer/Spool/SpoolConsumer.php b/src/Consumer/Spool/SpoolConsumer.php index de5b9aa..bc9a599 100644 --- a/src/Consumer/Spool/SpoolConsumer.php +++ b/src/Consumer/Spool/SpoolConsumer.php @@ -16,12 +16,18 @@ use FiveLab\Component\Amqp\AmqpEvents; use FiveLab\Component\Amqp\Channel\ChannelInterface; use FiveLab\Component\Amqp\Consumer\AbstractConsumer; +use FiveLab\Component\Amqp\Consumer\ConsumerConfiguration; use FiveLab\Component\Amqp\Consumer\ConsumerStoppedReason; +use FiveLab\Component\Amqp\Consumer\Handler\MessageHandlerInterface; +use FiveLab\Component\Amqp\Consumer\Strategy\ConsumeStrategyInterface; +use FiveLab\Component\Amqp\Consumer\Strategy\DefaultConsumeStrategy; +use FiveLab\Component\Amqp\Consumer\Strategy\LoopConsumeStrategy; use FiveLab\Component\Amqp\Event\ConsumerStartedEvent; use FiveLab\Component\Amqp\Event\ConsumerStoppedEvent; use FiveLab\Component\Amqp\Exception\ConsumerTimeoutExceedException; use FiveLab\Component\Amqp\Message\MutableReceivedMessages; use FiveLab\Component\Amqp\Message\ReceivedMessage; +use FiveLab\Component\Amqp\Queue\QueueFactoryInterface; use FiveLab\Component\Amqp\Queue\QueueInterface; /** @@ -35,6 +41,13 @@ { public function run(): void { + if (!$this->strategy instanceof LoopConsumeStrategy) { + throw new \LogicException(\sprintf( + 'The "%s" consume strategy is not supported for spool consumer (only loop supported).', + \get_class($this->strategy) + )); + } + $this->allowConsuming(); $this->getEventDispatcher()?->dispatch(new ConsumerStartedEvent($this), AmqpEvents::CONSUMER_STARTED); @@ -49,16 +62,13 @@ public function run(): void $receivedMessages = new MutableReceivedMessages(); $endTime = \microtime(true) + $this->configuration->timeout; - $countOfProcessedMessages = 0; - try { $this->doRun( false, - function (ReceivedMessage $message) use ($receivedMessages, &$countOfProcessedMessages, &$endTime): void { + function (ReceivedMessage $message) use ($receivedMessages, &$endTime): void { $receivedMessages->push($message); - $countOfProcessedMessages++; - if ($countOfProcessedMessages >= $this->configuration->prefetchCount) { + if (\count($receivedMessages) >= $this->configuration->prefetchCount) { $this->strategy->stopConsume(); } @@ -77,8 +87,6 @@ static function (ReceivedMessage $message): void { } catch (ConsumerTimeoutExceedException) { $this->flushMessages($receivedMessages); - $this->gracefulDisconnect(); - $this->getEventDispatcher()?->dispatch(new ConsumerStoppedEvent($this, ConsumerStoppedReason::Timeout), AmqpEvents::CONSUMER_STOPPED); continue; @@ -91,18 +99,13 @@ static function (ReceivedMessage $message): void { $receivedMessages->clear(); - $this->gracefulDisconnect(); - throw $error; } $this->flushMessages($receivedMessages); - // @todo: critical case, we can't reconnect per each batch block. - $this->gracefulDisconnect(); } $this->flushMessages($receivedMessages); - $this->gracefulDisconnect(); } private function flushMessages(MutableReceivedMessages $messages): void @@ -123,8 +126,6 @@ private function flushMessages(MutableReceivedMessages $messages): void $messages->clear(); - $this->gracefulDisconnect(); - throw $e; } diff --git a/src/Queue/QueueFactoryInterface.php b/src/Queue/QueueFactoryInterface.php index c119404..cfe68e0 100644 --- a/src/Queue/QueueFactoryInterface.php +++ b/src/Queue/QueueFactoryInterface.php @@ -21,11 +21,4 @@ interface QueueFactoryInterface * @return QueueInterface */ public function create(): QueueInterface; - - /** - * Create the queue factory with passive mode - * - * @return QueueFactoryInterface - */ - public function withPassive(): QueueFactoryInterface; } diff --git a/tests/Functional/Adapter/SpoolConsumerTestCase.php b/tests/Functional/Adapter/SpoolConsumerTestCase.php index dc6cf9b..0af62b0 100644 --- a/tests/Functional/Adapter/SpoolConsumerTestCase.php +++ b/tests/Functional/Adapter/SpoolConsumerTestCase.php @@ -71,7 +71,8 @@ public function shouldSuccessConsume(): void $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(10, 1) + new SpoolConsumerConfiguration(10, 1), + new LoopConsumeStrategy(), ); $this->runConsumer($consumer); @@ -99,7 +100,8 @@ public function shouldSuccessReturnMessagesToBrokerIfSpoolFailed(): void $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(10, 1, 0, true) + new SpoolConsumerConfiguration(10, 1, 0, true), + new LoopConsumeStrategy(), ); try { @@ -131,7 +133,8 @@ public function shouldNotReturnMessagesToBrokerIfSpoolFailedIfRequeueIsFalse(): $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(10, 1, 0, false) + new SpoolConsumerConfiguration(10, 1, 0, false), + new LoopConsumeStrategy(), ); try { @@ -158,7 +161,8 @@ public function shouldReturnMessagesToBrokerIfFlushFailed(): void $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(5, 1, 0, true) + new SpoolConsumerConfiguration(5, 1, 0, true), + new LoopConsumeStrategy(), ); try { @@ -185,7 +189,8 @@ public function shouldNotReturnMessagesToBrokerIfFlushFailedAndRequeueIsFalse(): $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(5, 1, 0, false) + new SpoolConsumerConfiguration(5, 1, 0, false), + new LoopConsumeStrategy(), ); try { @@ -221,7 +226,8 @@ public function shouldReturnMessagesToBrokerOnlyNotAckedMessagesIfFlushFalied(): $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(5, 1) + new SpoolConsumerConfiguration(5, 1), + new LoopConsumeStrategy(), ); try { @@ -260,7 +266,8 @@ public function shouldThrowExceptionIfMessageHandlerTryAnsweringToBroker(): void $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(10, 1) + new SpoolConsumerConfiguration(10, 1), + new LoopConsumeStrategy(), ); $this->expectException(\LogicException::class); @@ -284,7 +291,8 @@ public function shouldNotThrowOrphanedEnvelope(): void $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(10, 1) + new SpoolConsumerConfiguration(10, 1), + new LoopConsumeStrategy() ); $this->messageHandler->setHandlerCallback(function (ReceivedMessage $message) use (&$handledMessages, $consumer) { @@ -329,7 +337,7 @@ public function shouldSuccessProcessOnStopAfterNExecutes(): void { $this->publishMessages(12); - $consumer = new SpoolConsumer($this->queueFactory, $this->messageHandler, new SpoolConsumerConfiguration(100, 1)); + $consumer = new SpoolConsumer($this->queueFactory, $this->messageHandler, new SpoolConsumerConfiguration(100, 1), new LoopConsumeStrategy()); $consumer->setEventDispatcher($eventDispatcher = new EventDispatcher()); $eventDispatcher->addListener(AmqpEvents::PROCESSED_MESSAGE, (new StopAfterNExecutesListener(5))->onProcessedMessage(...)); @@ -363,7 +371,8 @@ public function shouldFlushWithZeroReadTimeout(int $prefetchCount, int $messageC $consumer = new SpoolConsumer( $this->queueFactory, $this->messageHandler, - new SpoolConsumerConfiguration($prefetchCount, 1, 1, true) + new SpoolConsumerConfiguration($prefetchCount, 1, 1, true), + new LoopConsumeStrategy(), ); $this->queueFactory->create()->getChannel()->getConnection()->setReadTimeout(0); @@ -388,7 +397,7 @@ public function shouldSuccessReRunAfterStop(): void { $this->publishMessages(10); - $consumer = new SpoolConsumer($this->queueFactory, $this->messageHandler, new SpoolConsumerConfiguration(10, 1, 1, true)); + $consumer = new SpoolConsumer($this->queueFactory, $this->messageHandler, new SpoolConsumerConfiguration(10, 1, 1, true), new LoopConsumeStrategy()); $consumer->setEventDispatcher($eventDispatcher = new EventDispatcher()); $eventDispatcher->addListener(AmqpEvents::PROCESSED_MESSAGE, static function (ProcessedMessageEvent $event): void { @@ -461,7 +470,8 @@ public function shouldSuccessAck2KMessagesForQuorumQueue(): void $consumer = new SpoolConsumer( $quorumQueueFactory, $this->messageHandler, - new SpoolConsumerConfiguration(2000, 2) + new SpoolConsumerConfiguration(2000, 2), + new LoopConsumeStrategy(), ); $this->runConsumer($consumer); @@ -483,7 +493,13 @@ private function runConsumer(SpoolConsumer $consumer, bool $changeReadTimeout = $consumer->getQueue()->getChannel()->getConnection()->setReadTimeout(0.2); } - $consumer->run(); + try { + $consumer->run(); + } finally { + \usleep(10_000); + $consumer->getQueue()->getChannel()->getConnection()->disconnect(); + \usleep(10_000); + } } private function registerTimeoutListenerForStop(SpoolConsumer $consumer): void diff --git a/tests/Functional/Signals/ConsumerSignalsInConsumeStrategyTest.php b/tests/Functional/Signals/ConsumerSignalsInConsumeStrategyTest.php index 7d0a249..49938c3 100644 --- a/tests/Functional/Signals/ConsumerSignalsInConsumeStrategyTest.php +++ b/tests/Functional/Signals/ConsumerSignalsInConsumeStrategyTest.php @@ -93,7 +93,7 @@ public function shouldSuccessCatchSignalsInSpoolConsumer(): void { $consumerFile = \realpath(__DIR__.'/../../Consumers/spool-with-signals.php'); - $process = new PhpSubprocess([$consumerFile, 'consume']); + $process = new PhpSubprocess([$consumerFile, 'loop']); $process->setTimeout(10); $sendSignal = false; @@ -108,8 +108,9 @@ public function shouldSuccessCatchSignalsInSpoolConsumer(): void $output = $process->getOutput(); $expectedOutput = << Date: Mon, 25 May 2026 19:38:11 +0300 Subject: [PATCH 2/2] Fix CI --- src/Adapter/Amqp/Queue/AmqpQueueFactory.php | 2 +- src/Adapter/AmqpLib/Queue/AmqpQueue.php | 6 +++++- src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php | 1 - src/Consumer/Spool/SpoolConsumer.php | 6 ------ 4 files changed, 6 insertions(+), 9 deletions(-) diff --git a/src/Adapter/Amqp/Queue/AmqpQueueFactory.php b/src/Adapter/Amqp/Queue/AmqpQueueFactory.php index 3705048..fe3437d 100644 --- a/src/Adapter/Amqp/Queue/AmqpQueueFactory.php +++ b/src/Adapter/Amqp/Queue/AmqpQueueFactory.php @@ -45,6 +45,7 @@ public function create(): QueueInterface } $queue = new \AMQPQueue($channel->getChannel()); + $channel->getConnection()->attach($this); $flags = $this->calculateFlagsForQueue(); @@ -55,7 +56,6 @@ public function create(): QueueInterface $queue->declareQueue(); $this->queue = new AmqpQueue($channel, $queue); - $channel->getConnection()->attach($this); foreach ($this->definition->bindings as $binding) { $queue->bind($binding->exchangeName, $binding->routingKey); diff --git a/src/Adapter/AmqpLib/Queue/AmqpQueue.php b/src/Adapter/AmqpLib/Queue/AmqpQueue.php index d506960..69db94d 100644 --- a/src/Adapter/AmqpLib/Queue/AmqpQueue.php +++ b/src/Adapter/AmqpLib/Queue/AmqpQueue.php @@ -53,13 +53,17 @@ public function consume(\Closure $handler, string $tag = ''): void $this->definition->exclusive, false, function (AMQPMessage $message) use ($handler, $queueName, &$stopConsuming, &$amqplibChannel): void { + if ($stopConsuming) { + $amqplibChannel->basic_nack($message->getDeliveryTag(), false, true); + } + $receivedMessage = new AmqpReceivedMessage($message, $queueName); $result = $handler($receivedMessage); if (false === $result) { $stopConsuming = true; - $amqplibChannel->basic_cancel($message->getConsumerTag(), false, true); + $amqplibChannel->basic_cancel((string) $message->getConsumerTag(), false, true); } } ); diff --git a/src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php b/src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php index 64cbcf3..7acd775 100644 --- a/src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php +++ b/src/Adapter/AmqpLib/Queue/AmqpQueueFactory.php @@ -14,7 +14,6 @@ namespace FiveLab\Component\Amqp\Adapter\AmqpLib\Queue; use FiveLab\Component\Amqp\Adapter\AmqpLib\Channel\AmqpChannel; -use FiveLab\Component\Amqp\Adapter\AmqpLib\Connection\AmqpConnection; use FiveLab\Component\Amqp\Channel\ChannelFactoryInterface; use FiveLab\Component\Amqp\Queue\Definition\QueueDefinition; use FiveLab\Component\Amqp\Queue\QueueFactoryInterface; diff --git a/src/Consumer/Spool/SpoolConsumer.php b/src/Consumer/Spool/SpoolConsumer.php index bc9a599..e34912b 100644 --- a/src/Consumer/Spool/SpoolConsumer.php +++ b/src/Consumer/Spool/SpoolConsumer.php @@ -16,19 +16,13 @@ use FiveLab\Component\Amqp\AmqpEvents; use FiveLab\Component\Amqp\Channel\ChannelInterface; use FiveLab\Component\Amqp\Consumer\AbstractConsumer; -use FiveLab\Component\Amqp\Consumer\ConsumerConfiguration; use FiveLab\Component\Amqp\Consumer\ConsumerStoppedReason; -use FiveLab\Component\Amqp\Consumer\Handler\MessageHandlerInterface; -use FiveLab\Component\Amqp\Consumer\Strategy\ConsumeStrategyInterface; -use FiveLab\Component\Amqp\Consumer\Strategy\DefaultConsumeStrategy; use FiveLab\Component\Amqp\Consumer\Strategy\LoopConsumeStrategy; use FiveLab\Component\Amqp\Event\ConsumerStartedEvent; use FiveLab\Component\Amqp\Event\ConsumerStoppedEvent; use FiveLab\Component\Amqp\Exception\ConsumerTimeoutExceedException; use FiveLab\Component\Amqp\Message\MutableReceivedMessages; use FiveLab\Component\Amqp\Message\ReceivedMessage; -use FiveLab\Component\Amqp\Queue\QueueFactoryInterface; -use FiveLab\Component\Amqp\Queue\QueueInterface; /** * The consumer for buffer all received messages by configuration and flush by configuration.