Allow only loop consume strategy for Spool

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) {
@@ -50,6 +45,7 @@ public function create(): QueueInterface
         }
 
         $queue = new \AMQPQueue($channel->getChannel());
+        $channel->getConnection()->attach($this);
 
         $flags = $this->calculateFlagsForQueue();
 
@@ -59,15 +55,7 @@ public function create(): QueueInterface
 
         $queue->declareQueue();
 
-        $amqpQueue = new AmqpQueue($channel, $queue);
-
-        if ($this->definition->passive) {
-            return $amqpQueue;
-        }
-
-        $this->queue = $amqpQueue;
-
-        $channel->getConnection()->attach($this);
+        $this->queue = new AmqpQueue($channel, $queue);
 
         foreach ($this->definition->bindings as $binding) {
             $queue->bind($binding->exchangeName, $binding->routingKey);

src/Adapter/AmqpLib/Queue/AmqpQueue.php

@@ -52,13 +52,18 @@ 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 {
+                    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((string) $message->getConsumerTag(), false, true);
                     }
                 }
             );

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;
@@ -30,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) {
@@ -55,10 +49,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) {

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();
-    }
 }

src/Consumer/Spool/SpoolConsumer.php

@@ -17,12 +17,12 @@
 use FiveLab\Component\Amqp\Channel\ChannelInterface;
 use FiveLab\Component\Amqp\Consumer\AbstractConsumer;
 use FiveLab\Component\Amqp\Consumer\ConsumerStoppedReason;
+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\QueueInterface;
 
 /**
  * The consumer for buffer all received messages by configuration and flush by configuration.
@@ -35,6 +35,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 +56,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 +81,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 +93,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 +120,6 @@ private function flushMessages(MutableReceivedMessages $messages): void
 
             $messages->clear();
 
-            $this->gracefulDisconnect();
-
             throw $e;
         }
 

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;
 }

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

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 = <<<OUTPUT
-bla 0
+tick 1
 handle signal: 15
+bla 0
 flush messages
 
 OUTPUT;