Merge pull request #234 from php-enqueue/queue-consumer-amqp-and-befo…

…re-receive-issue [consumption][amqp] move beforeReceive call at the end of the cycle f…
php-enqueue · Oct 17, 2017 · 380108f · 380108f
2 parents b2287c3 + 071f8ca
commit 380108f
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/pkg/enqueue/Consumption/QueueConsumer.php b/pkg/enqueue/Consumption/QueueConsumer.php
@@ -263,7 +263,9 @@ protected function doConsume(ExtensionInterface $extension, Context $context)
         $consumer = $context->getPsrConsumer();
         $logger = $context->getLogger();
 
-        $extension->onBeforeReceive($context);
+        if (false == $context->getPsrMessage() instanceof AmqpContext) {
+            $extension->onBeforeReceive($context);
+        }
 
         if ($context->isExecutionInterrupted()) {
             throw new ConsumptionInterruptedException();
@@ -307,6 +309,10 @@ protected function doConsume(ExtensionInterface $extension, Context $context)
             $logger->info(sprintf('Message processed: %s', $context->getResult()));
 
             $extension->onPostReceived($context);
+
+            if ($context->getPsrMessage() instanceof AmqpContext) {
+                $extension->onBeforeReceive($context);
+            }
         } else {
             usleep($this->idleTimeout * 1000);
             $extension->onIdle($context);