[consumption] Do not close context.

pkg/enqueue/Consumption/QueueConsumer.php

@@ -149,7 +149,6 @@ public function consume(ExtensionInterface $runtimeExtension = null)
                 $context->setExecutionInterrupted(true);
 
                 $extension->onInterrupted($context);
-                $this->psrContext->close();
 
                 return;
             } catch (\Exception $exception) {
@@ -158,10 +157,8 @@ public function consume(ExtensionInterface $runtimeExtension = null)
 
                 try {
                     $this->onInterruptionByException($extension, $context);
-                    $this->psrContext->close();
                 } catch (\Exception $e) {
                     // for some reason finally does not work here on php5.5
-                    $this->psrContext->close();
 
                     throw $e;
                 }

pkg/enqueue/Symfony/Client/ConsumeMessagesCommand.php

@@ -97,11 +97,7 @@ protected function execute(InputInterface $input, OutputInterface $output)
             $this->consumer->bind($queue, $this->processor);
         }
 
-        try {
-            $this->consumer->consume($this->getRuntimeExtensions($input, $output));
-        } finally {
-            $this->consumer->getPsrContext()->close();
-        }
+        $this->consumer->consume($this->getRuntimeExtensions($input, $output));
     }
 
     /**

pkg/enqueue/Symfony/Consumption/ConsumeMessagesCommand.php

@@ -56,10 +56,6 @@ protected function execute(InputInterface $input, OutputInterface $output)
 
         $runtimeExtensions = new ChainExtension($extensions);
 
-        try {
-            $this->consumer->consume($runtimeExtensions);
-        } finally {
-            $this->consumer->getPsrContext()->close();
-        }
+        $this->consumer->consume($runtimeExtensions);
     }
 }

pkg/enqueue/Symfony/Consumption/ContainerAwareConsumeMessagesCommand.php

@@ -87,14 +87,10 @@ protected function execute(InputInterface $input, OutputInterface $output)
 
         $runtimeExtensions = new ChainExtension($extensions);
 
-        try {
-            foreach ($queues as $queue) {
-                $this->consumer->bind($queue, $processor);
-            }
-
-            $this->consumer->consume($runtimeExtensions);
-        } finally {
-            $this->consumer->getPsrContext()->close();
+        foreach ($queues as $queue) {
+            $this->consumer->bind($queue, $processor);
         }
+
+        $this->consumer->consume($runtimeExtensions);
     }
 }

pkg/enqueue/Tests/Consumption/QueueConsumerTest.php

@@ -652,13 +652,13 @@ public function testShouldAllowInterruptConsumingOnIdle()
         $queueConsumer->consume();
     }
 
-    public function testShouldCloseSessionWhenConsumptionInterrupted()
+    public function testShouldNotCloseContextWhenConsumptionInterrupted()
     {
         $messageConsumerStub = $this->createMessageConsumerStub($message = null);
 
         $contextStub = $this->createPsrContextStub($messageConsumerStub);
         $contextStub
-            ->expects($this->once())
+            ->expects($this->never())
             ->method('close')
         ;
 
@@ -681,15 +681,15 @@ public function testShouldCloseSessionWhenConsumptionInterrupted()
         $queueConsumer->consume();
     }
 
-    public function testShouldCloseSessionWhenConsumptionInterruptedByException()
+    public function testShouldNotCloseContextWhenConsumptionInterruptedByException()
     {
         $expectedException = new \Exception();
 
         $messageConsumerStub = $this->createMessageConsumerStub($message = $this->createMessageMock());
 
         $contextStub = $this->createPsrContextStub($messageConsumerStub);
         $contextStub
-            ->expects($this->once())
+            ->expects($this->never())
             ->method('close')
         ;
 

pkg/enqueue/Tests/Symfony/Client/ConsumeMessagesCommandTest.php

@@ -91,7 +91,7 @@ public function testShouldExecuteConsumptionAndUseDefaultQueueName()
 
         $context = $this->createPsrContextMock();
         $context
-            ->expects($this->once())
+            ->expects($this->never())
             ->method('close')
         ;
 
@@ -106,11 +106,6 @@ public function testShouldExecuteConsumptionAndUseDefaultQueueName()
             ->method('consume')
             ->with($this->isInstanceOf(ChainExtension::class))
         ;
-        $consumer
-            ->expects($this->once())
-            ->method('getPsrContext')
-            ->will($this->returnValue($context))
-        ;
 
         $queueMetaRegistry = $this->createQueueMetaRegistry([
             'default' => [],
@@ -138,7 +133,7 @@ public function testShouldExecuteConsumptionAndUseCustomClientDestinationName()
 
         $context = $this->createPsrContextMock();
         $context
-            ->expects($this->once())
+            ->expects($this->never())
             ->method('close')
         ;
 
@@ -153,11 +148,6 @@ public function testShouldExecuteConsumptionAndUseCustomClientDestinationName()
             ->method('consume')
             ->with($this->isInstanceOf(ChainExtension::class))
         ;
-        $consumer
-            ->expects($this->once())
-            ->method('getPsrContext')
-            ->will($this->returnValue($context))
-        ;
 
         $queueMetaRegistry = $this->createQueueMetaRegistry([
             'non-default-queue' => [],

pkg/enqueue/Tests/Symfony/Consumption/ConsumeMessagesCommandTest.php

@@ -48,7 +48,7 @@ public function testShouldExecuteConsumption()
     {
         $context = $this->createContextMock();
         $context
-            ->expects($this->once())
+            ->expects($this->never())
             ->method('close')
         ;
 
@@ -58,11 +58,6 @@ public function testShouldExecuteConsumption()
             ->method('consume')
             ->with($this->isInstanceOf(ChainExtension::class))
         ;
-        $consumer
-            ->expects($this->once())
-            ->method('getPsrContext')
-            ->will($this->returnValue($context))
-        ;
 
         $command = new ConsumeMessagesCommand($consumer);
 

pkg/enqueue/Tests/Symfony/Consumption/ContainerAwareConsumeMessagesCommandTest.php

@@ -97,13 +97,13 @@ public function testThrowIfNeitherQueueOptionNorProcessorImplementsQueueSubscrib
         ]);
     }
 
-    public function testShouldExecuteConsumptionWithExplisitlySetQueueViaQueueOption()
+    public function testShouldExecuteConsumptionWithExplicitlySetQueueViaQueueOption()
     {
         $processor = $this->createProcessor();
 
         $context = $this->createContextMock();
         $context
-            ->expects($this->once())
+            ->expects($this->never())
             ->method('close')
         ;
 
@@ -118,11 +118,6 @@ public function testShouldExecuteConsumptionWithExplisitlySetQueueViaQueueOption
             ->method('consume')
             ->with($this->isInstanceOf(ChainExtension::class))
         ;
-        $consumer
-            ->expects($this->exactly(1))
-            ->method('getPsrContext')
-            ->will($this->returnValue($context))
-        ;
 
         $container = new Container();
         $container->set('processor-service', $processor);
@@ -143,7 +138,7 @@ public function testShouldExecuteConsumptionWhenProcessorImplementsQueueSubscrib
 
         $context = $this->createContextMock();
         $context
-            ->expects($this->once())
+            ->expects($this->never())
             ->method('close')
         ;
 
@@ -163,11 +158,6 @@ public function testShouldExecuteConsumptionWhenProcessorImplementsQueueSubscrib
             ->method('consume')
             ->with($this->isInstanceOf(ChainExtension::class))
         ;
-        $consumer
-            ->expects($this->at(3))
-            ->method('getPsrContext')
-            ->will($this->returnValue($context))
-        ;
 
         $container = new Container();
         $container->set('processor-service', $processor);