Merge pull request #53 from ale10257/master

Mikhail Bakulin · web-flow · commit 2c5552a1c9e9 · 2020-10-13T14:14:18.000+02:00
Fix ssl connection bug
diff --git a/Configuration.php b/Configuration.php
@@ -8,7 +8,10 @@
 use mikemadisonweb\rabbitmq\exceptions\InvalidConfigException;
 use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Connection\AMQPLazyConnection;
+use PhpAmqpLib\Connection\AMQPSSLConnection;
+use Yii;
 use yii\base\Component;
+use yii\di\NotInstantiableException;
 use yii\helpers\ArrayHelper;
 
 class Configuration extends Component
@@ -145,44 +148,53 @@ public function getConfig() : Configuration
     /**
      * Get connection service
      * @param string $connectionName
-     * @return AbstractConnection
+     * @return object|AbstractConnection
+     * @throws NotInstantiableException
+     * @throws \yii\base\InvalidConfigException
      */
     public function getConnection(string $connectionName = '') : AbstractConnection
     {
         if ('' === $connectionName) {
             $connectionName = self::DEFAULT_CONNECTION_NAME;
         }
 
-        return \Yii::$container->get(sprintf(self::CONNECTION_SERVICE_NAME, $connectionName));
+        return Yii::$container->get(sprintf(self::CONNECTION_SERVICE_NAME, $connectionName));
     }
 
     /**
      * Get producer service
      * @param string $producerName
-     * @return Producer
+     * @return Producer|object
+     * @throws \yii\base\InvalidConfigException
+     * @throws NotInstantiableException
      */
-    public function getProducer(string $producerName) : Producer
+    public function getProducer(string $producerName)
     {
-        return \Yii::$container->get(sprintf(self::PRODUCER_SERVICE_NAME, $producerName));
+        return Yii::$container->get(sprintf(self::PRODUCER_SERVICE_NAME, $producerName));
     }
 
     /**
      * Get consumer service
      * @param string $consumerName
-     * @return Consumer
+     * @return Consumer|object
+     * @throws NotInstantiableException
+     * @throws \yii\base\InvalidConfigException
      */
-    public function getConsumer(string $consumerName) : Consumer
+    public function getConsumer(string $consumerName)
     {
-        return \Yii::$container->get(sprintf(self::CONSUMER_SERVICE_NAME, $consumerName));
+        return Yii::$container->get(sprintf(self::CONSUMER_SERVICE_NAME, $consumerName));
     }
 
     /**
      * Get routing service
-     * @return Routing
+     * @param AbstractConnection $connection
+     * @return Routing|object|string
+     * @throws NotInstantiableException
+     * @throws \yii\base\InvalidConfigException
      */
-    public function getRouting(AbstractConnection $connection) : Routing
+    public function getRouting(AbstractConnection $connection)
     {
-        return \Yii::$container->get(Configuration::ROUTING_SERVICE_NAME, ['conn' => $connection]);
+        return Yii::$container->get(Configuration::ROUTING_SERVICE_NAME, ['conn' => $connection]);
     }
 
     /**
@@ -261,6 +273,12 @@ protected function validateRequired()
             if (isset($connection['type']) && !is_subclass_of($connection['type'], AbstractConnection::class)) {
                 throw new InvalidConfigException('Connection type should be a subclass of PhpAmqpLib\Connection\AbstractConnection.');
             }
+            if (!empty($connection['ssl_context']) && empty($connection['type'])) {
+                throw new InvalidConfigException('If you are using a ssl connection, the connection type must be AMQPSSLConnection::class');
+            }
+            if (!empty($connection['ssl_context']) && $connection['type'] !== AMQPSSLConnection::class) {
+                throw new InvalidConfigException('If you are using a ssl connection, the connection type must be AMQPSSLConnection::class');
+            }
         }
 
         foreach ($this->exchanges as $exchange) {
diff --git a/components/AbstractConnectionFactory.php b/components/AbstractConnectionFactory.php
@@ -29,6 +29,23 @@ public function __construct($class, array $parameters)
      */
     public function createConnection() : AbstractConnection
     {
+        if ($this->_parameters['ssl_context'] !== null) {
+            return new $this->_class(
+                $this->_parameters['host'],
+                $this->_parameters['port'],
+                $this->_parameters['user'],
+                $this->_parameters['password'],
+                $this->_parameters['vhost'],
+                $this->_parameters['ssl_context'],
+                [
+                    'connection_timeout' => $this->_parameters['connection_timeout'],
+                    'read_write_timeout' => $this->_parameters['read_write_timeout'],
+                    'keepalive' => $this->_parameters['keepalive'],
+                    'heartbeat' => $this->_parameters['heartbeat'],
+                    'channel_rpc_timeout' => $this->_parameters['channel_rpc_timeout'],
+                ]
+            );
+        }
         return new $this->_class(
             $this->_parameters['host'],
             $this->_parameters['port'],
diff --git a/composer.json b/composer.json
@@ -12,7 +12,7 @@
   ],
   "require": {
     "php": "^7.0",
-    "yiisoft/yii2": "^2.0",
+    "yiisoft/yii2": "^2.0.13",
     "php-amqplib/php-amqplib": "^2.9"
   },
   "require-dev": {
diff --git a/controllers/RabbitMQController.php b/controllers/RabbitMQController.php
@@ -10,6 +10,7 @@
 use ReflectionException;
 use yii\base\Action;
 use yii\console\Controller;
+use yii\console\ExitCode;
 use yii\helpers\Console;
 
 /**
@@ -101,7 +102,7 @@ public function actionConsume(string $name): int
         {
             $this->stderr(Console::ansiFormat("Consumer `{$name}` doesn't exist: {$e->getMessage()}\n", [Console::FG_RED]));
 
-            return self::EXIT_CODE_ERROR;
+            return ExitCode::UNSPECIFIED_ERROR;
         }
 
         $this->validateConsumerOptions($consumer);
@@ -111,7 +112,7 @@ public function actionConsume(string $name): int
         }
         $consumer->consume($this->messagesLimit);
 
-        return self::EXIT_CODE_NORMAL;
+        return ExitCode::OK;
     }
 
     /**
@@ -136,15 +137,15 @@ public function actionPublish(string $producerName, string $exchangeName, string
         {
             $this->stderr(Console::ansiFormat("Producer `{$producerName}` doesn't exist: {$e->getMessage()}\n", [Console::FG_RED]));
 
-            return self::EXIT_CODE_ERROR;
+            return ExitCode::UNSPECIFIED_ERROR;
         }
 
         $data = '';
         if (posix_isatty(STDIN))
         {
             $this->stderr(Console::ansiFormat("Please pipe in some data in order to send it.\n", [Console::FG_RED]));
 
-            return self::EXIT_CODE_ERROR;
+            return ExitCode::UNSPECIFIED_ERROR;
         }
         while (!feof(STDIN))
         {
@@ -153,7 +154,7 @@ public function actionPublish(string $producerName, string $exchangeName, string
         $producer->publish($data, $exchangeName, $routingKey);
         $this->stdout("Message was successfully published.\n", Console::FG_GREEN);
 
-        return self::EXIT_CODE_NORMAL;
+        return ExitCode::OK;
     }
 
     /**
@@ -175,11 +176,11 @@ public function actionDeclareAll(string $connectionName = Configuration::DEFAULT
                 Console::ansiFormat("All configured entries was successfully declared.\n", [Console::FG_GREEN])
             );
 
-            return self::EXIT_CODE_NORMAL;
+            return ExitCode::OK;
         }
         $this->stderr(Console::ansiFormat("No queues, exchanges or bindings configured.\n", [Console::FG_RED]));
 
-        return self::EXIT_CODE_ERROR;
+        return ExitCode::UNSPECIFIED_ERROR;
     }
 
     /**
@@ -201,12 +202,12 @@ public function actionDeclareExchange(
         {
             $this->stderr(Console::ansiFormat("Exchange `{$exchangeName}` is already exists.\n", [Console::FG_RED]));
 
-            return self::EXIT_CODE_ERROR;
+            return ExitCode::UNSPECIFIED_ERROR;
         }
         $routing->declareExchange($exchangeName);
         $this->stdout(Console::ansiFormat("Exchange `{$exchangeName}` was declared.\n", [Console::FG_GREEN]));
 
-        return self::EXIT_CODE_NORMAL;
+        return ExitCode::OK;
     }
 
     /**
@@ -228,12 +229,12 @@ public function actionDeclareQueue(
         {
             $this->stderr(Console::ansiFormat("Queue `{$queueName}` is already exists.\n", [Console::FG_RED]));
 
-            return self::EXIT_CODE_ERROR;
+            return ExitCode::UNSPECIFIED_ERROR;
         }
         $routing->declareQueue($queueName);
         $this->stdout(Console::ansiFormat("Queue `{$queueName}` was declared.\n", [Console::FG_GREEN]));
 
-        return self::EXIT_CODE_NORMAL;
+        return ExitCode::OK;
     }
 
     /**
@@ -249,19 +250,19 @@ public function actionDeleteAll(string $connection = Configuration::DEFAULT_CONN
         if ($this->interactive)
         {
             $input = Console::prompt('Are you sure you want to delete all queues and exchanges?', ['default' => 'yes']);
-            if ($input !== 'yes')
+            if ($input !== 'yes' && $input !== 'y')
             {
                 $this->stderr(Console::ansiFormat("Aborted.\n", [Console::FG_RED]));
 
-                return self::EXIT_CODE_ERROR;
+                return ExitCode::UNSPECIFIED_ERROR;
             }
         }
         $conn    = \Yii::$app->rabbitmq->getConnection($connection);
         $routing = \Yii::$app->rabbitmq->getRouting($conn);
         $routing->deleteAll();
         $this->stdout(Console::ansiFormat("All configured entries was deleted.\n", [Console::FG_GREEN]));
 
-        return self::EXIT_CODE_NORMAL;
+        return ExitCode::OK;
     }
 
     /**
@@ -284,15 +285,15 @@ public function actionDeleteExchange(
             {
                 $this->stderr(Console::ansiFormat("Aborted.\n", [Console::FG_RED]));
 
-                return self::EXIT_CODE_ERROR;
+                return ExitCode::UNSPECIFIED_ERROR;
             }
         }
         $conn    = \Yii::$app->rabbitmq->getConnection($connectionName);
         $routing = \Yii::$app->rabbitmq->getRouting($conn);
         $routing->deleteExchange($exchangeName);
         $this->stdout(Console::ansiFormat("Exchange `{$exchangeName}` was deleted.\n", [Console::FG_GREEN]));
 
-        return self::EXIT_CODE_NORMAL;
+        return ExitCode::OK;
     }
 
     /**
@@ -315,7 +316,7 @@ public function actionDeleteQueue(
             {
                 $this->stderr(Console::ansiFormat("Aborted.\n", [Console::FG_RED]));
 
-                return self::EXIT_CODE_ERROR;
+                return ExitCode::UNSPECIFIED_ERROR;
             }
         }
 
@@ -324,7 +325,7 @@ public function actionDeleteQueue(
         $routing->deleteQueue($queueName);
         $this->stdout(Console::ansiFormat("Queue `{$queueName}` was deleted.\n", [Console::FG_GREEN]));
 
-        return self::EXIT_CODE_NORMAL;
+        return ExitCode::OK;
     }
 
     /**
@@ -350,7 +351,7 @@ public function actionPurgeQueue(
             {
                 $this->stderr(Console::ansiFormat("Aborted.\n", [Console::FG_RED]));
 
-                return self::EXIT_CODE_ERROR;
+                return ExitCode::UNSPECIFIED_ERROR;
             }
         }
 
@@ -359,7 +360,7 @@ public function actionPurgeQueue(
         $routing->purgeQueue($queueName);
         $this->stdout(Console::ansiFormat("Queue `{$queueName}` was purged.\n", [Console::FG_GREEN]));
 
-        return self::EXIT_CODE_NORMAL;
+        return ExitCode::OK;
     }
 
     /**
diff --git a/events/RabbitMQConsumerEvent.php b/events/RabbitMQConsumerEvent.php
@@ -2,6 +2,8 @@
 
 namespace mikemadisonweb\rabbitmq\events;
 
+use mikemadisonweb\rabbitmq\components\Consumer;
+use PhpAmqpLib\Message\AMQPMessage;
 use yii\base\Event;
 
 class RabbitMQConsumerEvent extends Event
diff --git a/events/RabbitMQPublisherEvent.php b/events/RabbitMQPublisherEvent.php
@@ -2,6 +2,8 @@
 
 namespace mikemadisonweb\rabbitmq\events;
 
+use mikemadisonweb\rabbitmq\components\Consumer;
+use PhpAmqpLib\Message\AMQPMessage;
 use yii\base\Event;
 
 class RabbitMQPublisherEvent extends Event

Original file line number	Diff line number	Diff line change
`@@ -10,6 +10,7 @@`
`10`	`10`	`use ReflectionException;`
`11`	`11`	`use yii\base\Action;`
`12`	`12`	`use yii\console\Controller;`
	`13`	`+use yii\console\ExitCode;`
`13`	`14`	`use yii\helpers\Console;`
`14`	`15`
`15`	`16`	`/**`
`@@ -101,7 +102,7 @@ public function actionConsume(string $name): int`
`101`	`102`	`{`
`102`	`103`	$this->stderr(Console::ansiFormat("Consumer `{$name}` doesn't exist: {$e->getMessage()}\n", [Console::FG_RED]));
`103`	`104`
`104`		`- return self::EXIT_CODE_ERROR;`
	`105`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`105`	`106`	`}`
`106`	`107`
`107`	`108`	`$this->validateConsumerOptions($consumer);`
`@@ -111,7 +112,7 @@ public function actionConsume(string $name): int`
`111`	`112`	`}`
`112`	`113`	`$consumer->consume($this->messagesLimit);`
`113`	`114`
`114`		`- return self::EXIT_CODE_NORMAL;`
	`115`	`+ return ExitCode::OK;`
`115`	`116`	`}`
`116`	`117`
`117`	`118`	`/**`
`@@ -136,15 +137,15 @@ public function actionPublish(string $producerName, string $exchangeName, string`
`136`	`137`	`{`
`137`	`138`	$this->stderr(Console::ansiFormat("Producer `{$producerName}` doesn't exist: {$e->getMessage()}\n", [Console::FG_RED]));
`138`	`139`
`139`		`- return self::EXIT_CODE_ERROR;`
	`140`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`140`	`141`	`}`
`141`	`142`
`142`	`143`	`$data = '';`
`143`	`144`	`if (posix_isatty(STDIN))`
`144`	`145`	`{`
`145`	`146`	`$this->stderr(Console::ansiFormat("Please pipe in some data in order to send it.\n", [Console::FG_RED]));`
`146`	`147`
`147`		`- return self::EXIT_CODE_ERROR;`
	`148`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`148`	`149`	`}`
`149`	`150`	`while (!feof(STDIN))`
`150`	`151`	`{`
`@@ -153,7 +154,7 @@ public function actionPublish(string $producerName, string $exchangeName, string`
`153`	`154`	`$producer->publish($data, $exchangeName, $routingKey);`
`154`	`155`	`$this->stdout("Message was successfully published.\n", Console::FG_GREEN);`
`155`	`156`
`156`		`- return self::EXIT_CODE_NORMAL;`
	`157`	`+ return ExitCode::OK;`
`157`	`158`	`}`
`158`	`159`
`159`	`160`	`/**`
`@@ -175,11 +176,11 @@ public function actionDeclareAll(string $connectionName = Configuration::DEFAULT`
`175`	`176`	`Console::ansiFormat("All configured entries was successfully declared.\n", [Console::FG_GREEN])`
`176`	`177`	`);`
`177`	`178`
`178`		`- return self::EXIT_CODE_NORMAL;`
	`179`	`+ return ExitCode::OK;`
`179`	`180`	`}`
`180`	`181`	`$this->stderr(Console::ansiFormat("No queues, exchanges or bindings configured.\n", [Console::FG_RED]));`
`181`	`182`
`182`		`- return self::EXIT_CODE_ERROR;`
	`183`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`183`	`184`	`}`
`184`	`185`
`185`	`186`	`/**`
`@@ -201,12 +202,12 @@ public function actionDeclareExchange(`
`201`	`202`	`{`
`202`	`203`	$this->stderr(Console::ansiFormat("Exchange `{$exchangeName}` is already exists.\n", [Console::FG_RED]));
`203`	`204`
`204`		`- return self::EXIT_CODE_ERROR;`
	`205`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`205`	`206`	`}`
`206`	`207`	`$routing->declareExchange($exchangeName);`
`207`	`208`	$this->stdout(Console::ansiFormat("Exchange `{$exchangeName}` was declared.\n", [Console::FG_GREEN]));
`208`	`209`
`209`		`- return self::EXIT_CODE_NORMAL;`
	`210`	`+ return ExitCode::OK;`
`210`	`211`	`}`
`211`	`212`
`212`	`213`	`/**`
`@@ -228,12 +229,12 @@ public function actionDeclareQueue(`
`228`	`229`	`{`
`229`	`230`	$this->stderr(Console::ansiFormat("Queue `{$queueName}` is already exists.\n", [Console::FG_RED]));
`230`	`231`
`231`		`- return self::EXIT_CODE_ERROR;`
	`232`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`232`	`233`	`}`
`233`	`234`	`$routing->declareQueue($queueName);`
`234`	`235`	$this->stdout(Console::ansiFormat("Queue `{$queueName}` was declared.\n", [Console::FG_GREEN]));
`235`	`236`
`236`		`- return self::EXIT_CODE_NORMAL;`
	`237`	`+ return ExitCode::OK;`
`237`	`238`	`}`
`238`	`239`
`239`	`240`	`/**`
`@@ -249,19 +250,19 @@ public function actionDeleteAll(string $connection = Configuration::DEFAULT_CONN`
`249`	`250`	`if ($this->interactive)`
`250`	`251`	`{`
`251`	`252`	`$input = Console::prompt('Are you sure you want to delete all queues and exchanges?', ['default' => 'yes']);`
`252`		`- if ($input !== 'yes')`
	`253`	`+ if ($input !== 'yes' && $input !== 'y')`
`253`	`254`	`{`
`254`	`255`	`$this->stderr(Console::ansiFormat("Aborted.\n", [Console::FG_RED]));`
`255`	`256`
`256`		`- return self::EXIT_CODE_ERROR;`
	`257`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`257`	`258`	`}`
`258`	`259`	`}`
`259`	`260`	`$conn = \Yii::$app->rabbitmq->getConnection($connection);`
`260`	`261`	`$routing = \Yii::$app->rabbitmq->getRouting($conn);`
`261`	`262`	`$routing->deleteAll();`
`262`	`263`	`$this->stdout(Console::ansiFormat("All configured entries was deleted.\n", [Console::FG_GREEN]));`
`263`	`264`
`264`		`- return self::EXIT_CODE_NORMAL;`
	`265`	`+ return ExitCode::OK;`
`265`	`266`	`}`
`266`	`267`
`267`	`268`	`/**`
`@@ -284,15 +285,15 @@ public function actionDeleteExchange(`
`284`	`285`	`{`
`285`	`286`	`$this->stderr(Console::ansiFormat("Aborted.\n", [Console::FG_RED]));`
`286`	`287`
`287`		`- return self::EXIT_CODE_ERROR;`
	`288`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`288`	`289`	`}`
`289`	`290`	`}`
`290`	`291`	`$conn = \Yii::$app->rabbitmq->getConnection($connectionName);`
`291`	`292`	`$routing = \Yii::$app->rabbitmq->getRouting($conn);`
`292`	`293`	`$routing->deleteExchange($exchangeName);`
`293`	`294`	$this->stdout(Console::ansiFormat("Exchange `{$exchangeName}` was deleted.\n", [Console::FG_GREEN]));
`294`	`295`
`295`		`- return self::EXIT_CODE_NORMAL;`
	`296`	`+ return ExitCode::OK;`
`296`	`297`	`}`
`297`	`298`
`298`	`299`	`/**`
`@@ -315,7 +316,7 @@ public function actionDeleteQueue(`
`315`	`316`	`{`
`316`	`317`	`$this->stderr(Console::ansiFormat("Aborted.\n", [Console::FG_RED]));`
`317`	`318`
`318`		`- return self::EXIT_CODE_ERROR;`
	`319`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`319`	`320`	`}`
`320`	`321`	`}`
`321`	`322`
`@@ -324,7 +325,7 @@ public function actionDeleteQueue(`
`324`	`325`	`$routing->deleteQueue($queueName);`
`325`	`326`	$this->stdout(Console::ansiFormat("Queue `{$queueName}` was deleted.\n", [Console::FG_GREEN]));
`326`	`327`
`327`		`- return self::EXIT_CODE_NORMAL;`
	`328`	`+ return ExitCode::OK;`
`328`	`329`	`}`
`329`	`330`
`330`	`331`	`/**`
`@@ -350,7 +351,7 @@ public function actionPurgeQueue(`
`350`	`351`	`{`
`351`	`352`	`$this->stderr(Console::ansiFormat("Aborted.\n", [Console::FG_RED]));`
`352`	`353`
`353`		`- return self::EXIT_CODE_ERROR;`
	`354`	`+ return ExitCode::UNSPECIFIED_ERROR;`
`354`	`355`	`}`
`355`	`356`	`}`
`356`	`357`
`@@ -359,7 +360,7 @@ public function actionPurgeQueue(`
`359`	`360`	`$routing->purgeQueue($queueName);`
`360`	`361`	$this->stdout(Console::ansiFormat("Queue `{$queueName}` was purged.\n", [Console::FG_GREEN]));
`361`	`362`
`362`		`- return self::EXIT_CODE_NORMAL;`
	`363`	`+ return ExitCode::OK;`
`363`	`364`	`}`
`364`	`365`
`365`	`366`	`/**`