Added AMQP Module option to use only one channel (#4281)

Author: Tekill

src/Codeception/Module/AMQP.php

@@ -1,15 +1,15 @@
 <?php
 namespace Codeception\Module;
 
+use Codeception\Exception\ModuleException as ModuleException;
 use Codeception\Lib\Interfaces\RequiresPackage;
 use Codeception\Module as CodeceptionModule;
-use Codeception\Exception\ModuleException as ModuleException;
 use Codeception\TestInterface;
 use Exception;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
-use PhpAmqpLib\Message\AMQPMessage;
 use PhpAmqpLib\Exception\AMQPProtocolChannelException;
+use PhpAmqpLib\Message\AMQPMessage;
 
 /**
  * This module interacts with message broker software that implements
@@ -27,6 +27,7 @@
  * * vhost: '/' - vhost to connect
  * * cleanup: true - defined queues will be purged before running every test.
  * * queues: [mail, twitter] - queues to cleanup
+ * * single_channel - create and use only one channel during test execution
  *
  * ### Example
  *
@@ -39,6 +40,7 @@
  *                 password: 'guest'
  *                 vhost: '/'
  *                 queues: [queue1, queue2]
+ *                 single_channel: false
  *
  * ## Public Properties
  *
@@ -47,12 +49,13 @@
 class AMQP extends CodeceptionModule implements RequiresPackage
 {
     protected $config = [
-        'host'     => 'localhost',
-        'username' => 'guest',
-        'password' => 'guest',
-        'port'     => '5672',
-        'vhost'    => '/',
-        'cleanup'  => true,
+        'host'           => 'localhost',
+        'username'       => 'guest',
+        'password'       => 'guest',
+        'port'           => '5672',
+        'vhost'          => '/',
+        'cleanup'        => true,
+        'single_channel' => false
     ];
 
     /**
@@ -61,9 +64,9 @@ class AMQP extends CodeceptionModule implements RequiresPackage
     public $connection;
 
     /**
-     * @var AMQPChannel
+     * @var int
      */
-    protected $channel;
+    protected $channelId;
 
     protected $requiredFields = ['host', 'username', 'password', 'vhost'];
 
@@ -115,7 +118,7 @@ public function pushToExchange($exchange, $message, $routing_key = null)
         $message = $message instanceof AMQPMessage
             ? $message
             : new AMQPMessage($message);
-        $this->connection->channel()->basic_publish($message, $exchange, $routing_key);
+        $this->getChannel()->basic_publish($message, $exchange, $routing_key);
     }
 
     /**
@@ -137,8 +140,8 @@ public function pushToQueue($queue, $message)
             ? $message
             : new AMQPMessage($message);
 
-        $this->connection->channel()->queue_declare($queue);
-        $this->connection->channel()->basic_publish($message, '', $queue);
+        $this->getChannel()->queue_declare($queue);
+        $this->getChannel()->basic_publish($message, '', $queue);
     }
 
     /**
@@ -176,7 +179,7 @@ public function declareExchange(
         $arguments = null,
         $ticket = null
     ) {
-        return $this->connection->channel()->exchange_declare(
+        return $this->getChannel()->exchange_declare(
             $exchange,
             $type,
             $passive,
@@ -221,7 +224,7 @@ public function declareQueue(
         $arguments = null,
         $ticket = null
     ) {
-        return $this->connection->channel()->queue_declare(
+        return $this->getChannel()->queue_declare(
             $queue,
             $passive,
             $durable,
@@ -263,7 +266,7 @@ public function bindQueueToExchange(
         $arguments = null,
         $ticket = null
     ) {
-        return $this->connection->channel()->queue_bind(
+        return $this->getChannel()->queue_bind(
             $queue,
             $exchange,
             $routing_key,
@@ -291,7 +294,7 @@ public function bindQueueToExchange(
      */
     public function seeMessageInQueueContainsText($queue, $text)
     {
-        $msg = $this->connection->channel()->basic_get($queue);
+        $msg = $this->getChannel()->basic_get($queue);
         if (!$msg) {
             $this->fail("Message was not received");
         }
@@ -316,7 +319,7 @@ public function seeMessageInQueueContainsText($queue, $text)
      */
     public function grabMessageFromQueue($queue)
     {
-        $message = $this->connection->channel()->basic_get($queue);
+        $message = $this->getChannel()->basic_get($queue);
         return $message;
     }
 
@@ -337,7 +340,7 @@ public function purgeQueue($queueName = '')
             throw new ModuleException(__CLASS__, "'$queueName' doesn't exist in queues config list");
         }
 
-        $this->connection->channel()->queue_purge($queueName, true);
+        $this->getChannel()->queue_purge($queueName, true);
     }
 
     /**
@@ -354,6 +357,17 @@ public function purgeAllQueues()
         $this->cleanup();
     }
 
+    /**
+     * @return AMQPChannel
+     */
+    protected function getChannel()
+    {
+        if ($this->config['single_channel'] && $this->channelId === null) {
+            $this->channelId = $this->connection->get_free_channel_id();
+        }
+        return $this->connection->channel($this->channelId);
+    }
+
     protected function cleanup()
     {
         if (!isset($this->config['queues'])) {
@@ -364,7 +378,7 @@ protected function cleanup()
         }
         foreach ($this->config['queues'] as $queue) {
             try {
-                $this->connection->channel()->queue_purge($queue);
+                $this->getChannel()->queue_purge($queue);
             } catch (AMQPProtocolChannelException $e) {
                 // ignore if exchange/queue doesn't exist and rethrow exception if it's something else
                 if ($e->getCode() !== 404) {