true, 'connections' => [ [ 'name' => self::DEFAULT_CONNECTION_NAME, 'type' => AMQPLazyConnection::class, 'url' => null, 'host' => null, 'port' => 5672, 'user' => 'guest', 'password' => 'guest', 'vhost' => '/', 'connection_timeout' => 3, 'read_write_timeout' => 3, 'ssl_context' => null, 'keepalive' => false, 'heartbeat' => 0, 'channel_rpc_timeout' => 0.0, ], ], 'exchanges' => [ [ 'name' => null, 'type' => null, 'passive' => false, 'durable' => true, 'auto_delete' => false, 'internal' => false, 'nowait' => false, 'arguments' => null, 'ticket' => null, 'declare' => true, ], ], 'queues' => [ [ 'name' => '', 'passive' => false, 'durable' => true, 'exclusive' => false, 'auto_delete' => false, 'nowait' => false, 'arguments' => null, 'ticket' => null, 'declare' => true, ], ], 'bindings' => [ [ 'exchange' => null, 'queue' => null, 'to_exchange' => null, 'routing_keys' => [], ], ], 'producers' => [ [ 'name' => null, 'connection' => self::DEFAULT_CONNECTION_NAME, 'safe' => true, 'content_type' => 'text/plain', 'delivery_mode' => 2, 'serializer' => 'serialize', ], ], 'consumers' => [ [ 'name' => null, 'connection' => self::DEFAULT_CONNECTION_NAME, 'callbacks' => [], 'qos' => [ 'prefetch_size' => 0, 'prefetch_count' => 0, 'global' => false, ], 'idle_timeout' => 0, 'idle_timeout_exit_code' => null, 'proceed_on_exception' => false, 'deserializer' => 'unserialize', 'systemd' => [ 'memory_limit' => 0, 'workers' => 1 ], ], ], 'logger' => [ 'log' => false, 'category' => 'application', 'print_console' => true, 'system_memory' => false, ], ]; public $auto_declare = null; public $connections = []; public $producers = []; public $consumers = []; public $queues = []; public $exchanges = []; public $bindings = []; public $logger = []; protected $isLoaded = false; /** * Get passed configuration * @return Configuration * @throws InvalidConfigException */ public function getConfig() : Configuration { if(!$this->isLoaded) { $this->normalizeConnections(); $this->validate(); $this->completeWithDefaults(); $this->isLoaded = true; } return $this; } /** * Get connection service * @param string $connectionName * @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)); } /** * Get producer service * @param string $producerName * @return Producer|object * @throws \yii\base\InvalidConfigException * @throws NotInstantiableException */ public function getProducer(string $producerName) { return Yii::$container->get(sprintf(self::PRODUCER_SERVICE_NAME, $producerName)); } /** * Get consumer service * @param string $consumerName * @return Consumer|object * @throws NotInstantiableException * @throws \yii\base\InvalidConfigException */ public function getConsumer(string $consumerName) { return Yii::$container->get(sprintf(self::CONSUMER_SERVICE_NAME, $consumerName)); } /** * Get routing service * @param AbstractConnection $connection * @return Routing|object|string * @throws NotInstantiableException * @throws \yii\base\InvalidConfigException */ public function getRouting(AbstractConnection $connection) { return Yii::$container->get(Configuration::ROUTING_SERVICE_NAME, ['conn' => $connection]); } /** * Config validation * @throws InvalidConfigException */ protected function validate() { $this->validateTopLevel(); $this->validateMultidimensional(); $this->validateRequired(); $this->validateDuplicateNames(['connections', 'exchanges', 'queues', 'producers', 'consumers']); } /** * Validate multidimensional entries names * @throws InvalidConfigException */ protected function validateMultidimensional() { $multidimensional = [ 'connection' => $this->connections, 'exchange' => $this->exchanges, 'queue' => $this->queues, 'binding' => $this->bindings, 'producer' => $this->producers, 'consumer' => $this->consumers, ]; foreach ($multidimensional as $configName => $configItem) { if (!is_array($configItem)) { throw new InvalidConfigException("Every {$configName} entry should be of type array."); } foreach ($configItem as $key => $value) { if (!is_int($key)) { throw new InvalidConfigException("Invalid key: `{$key}`. There should be a list of {$configName}s in the array."); } } } } /** * Validate top level options * @throws InvalidConfigException */ protected function validateTopLevel() { if (($this->auto_declare !== null) && !is_bool($this->auto_declare)) { throw new InvalidConfigException("Option `auto_declare` should be of type boolean."); } if (!is_array($this->logger)) { throw new InvalidConfigException("Option `logger` should be of type array."); } $this->validateArrayFields($this->logger, self::DEFAULTS['logger']); } /** * Validate required options * @throws InvalidConfigException */ protected function validateRequired() { foreach ($this->connections as $connection) { $this->validateArrayFields($connection, self::DEFAULTS['connections'][0]); if (!isset($connection['url']) && !isset($connection['host'])) { throw new InvalidConfigException('Either `url` or `host` options required for configuring connection.'); } if (isset($connection['url']) && (isset($connection['host']) || isset($connection['port']))) { throw new InvalidConfigException('Connection options `url` and `host:port` should not be both specified, configuration is ambigious.'); } if (!isset($connection['name'])) { throw new InvalidConfigException('Connection name is required when multiple connections is specified.'); } 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) { $this->validateArrayFields($exchange, self::DEFAULTS['exchanges'][0]); if (!isset($exchange['name'])) { throw new InvalidConfigException('Exchange name should be specified.'); } if (!isset($exchange['type'])) { throw new InvalidConfigException('Exchange type should be specified.'); } $allowed = ['direct', 'topic', 'fanout', 'headers', 'x-delayed-message']; if (!in_array($exchange['type'], $allowed, true)) { $allowed = implode(', ', $allowed); throw new InvalidConfigException("Unknown exchange type `{$exchange['type']}`. Allowed values are: {$allowed}"); } } foreach ($this->queues as $queue) { $this->validateArrayFields($queue, self::DEFAULTS['queues'][0]); } foreach ($this->bindings as $binding) { $this->validateArrayFields($binding, self::DEFAULTS['bindings'][0]); if (!isset($binding['exchange'])) { throw new InvalidConfigException('Exchange name is required for binding.'); } if (!$this->isNameExist($this->exchanges, $binding['exchange'])) { throw new InvalidConfigException("`{$binding['exchange']}` defined in binding doesn't configured in exchanges."); } if (isset($binding['routing_keys']) && !is_array($binding['routing_keys'])) { throw new InvalidConfigException('Option `routing_keys` should be an array.'); } if ((!isset($binding['queue']) && !isset($binding['to_exchange'])) || isset($binding['queue'], $binding['to_exchange'])) { throw new InvalidConfigException('Either `queue` or `to_exchange` options should be specified to create binding.'); } if (isset($binding['queue']) && !$this->isNameExist($this->queues, $binding['queue'])) { throw new InvalidConfigException("`{$binding['queue']}` defined in binding doesn't configured in queues."); } } foreach ($this->producers as $producer) { $this->validateArrayFields($producer, self::DEFAULTS['producers'][0]); if (!isset($producer['name'])) { throw new InvalidConfigException('Producer name is required.'); } if (isset($producer['connection']) && !$this->isNameExist($this->connections, $producer['connection'])) { throw new InvalidConfigException("Connection `{$producer['connection']}` defined in producer doesn't configured in connections."); } if (isset($producer['safe']) && !is_bool($producer['safe'])) { throw new InvalidConfigException('Producer option safe should be of type boolean.'); } if (!isset($producer['connection']) && !$this->isNameExist($this->connections, self::DEFAULT_CONNECTION_NAME)) { throw new InvalidConfigException("Connection for producer `{$producer['name']}` is required."); } if (isset($producer['serializer']) && !is_callable($producer['serializer'])) { throw new InvalidConfigException('Producer `serializer` option should be a callable.'); } } foreach ($this->consumers as $consumer) { $this->validateArrayFields($consumer, self::DEFAULTS['consumers'][0]); if (!isset($consumer['name'])) { throw new InvalidConfigException('Consumer name is required.'); } if (isset($consumer['connection']) && !$this->isNameExist($this->connections, $consumer['connection'])) { throw new InvalidConfigException("Connection `{$consumer['connection']}` defined in consumer doesn't configured in connections."); } if (!isset($consumer['connection']) && !$this->isNameExist($this->connections, self::DEFAULT_CONNECTION_NAME)) { throw new InvalidConfigException("Connection for consumer `{$consumer['name']}` is required."); } if (!isset($consumer['callbacks']) || empty($consumer['callbacks'])) { throw new InvalidConfigException("No callbacks specified for consumer `{$consumer['name']}`."); } if (isset($consumer['qos']) && !is_array($consumer['qos'])) { throw new InvalidConfigException('Consumer option `qos` should be of type array.'); } if (isset($consumer['proceed_on_exception']) && !is_bool($consumer['proceed_on_exception'])) { throw new InvalidConfigException('Consumer option `proceed_on_exception` should be of type boolean.'); } foreach ($consumer['callbacks'] as $queue => $callback) { if (!$this->isNameExist($this->queues, $queue)) { throw new InvalidConfigException("Queue `{$queue}` from {$consumer['name']} is not defined in queues."); } if (!is_string($callback)) { throw new InvalidConfigException('Consumer `callback` parameter value should be a class name or service name in DI container.'); } } if (isset($consumer['deserializer']) && !is_callable($consumer['deserializer'])) { throw new InvalidConfigException('Consumer `deserializer` option should be a callable.'); } } } /** * Validate config entry value * @param array $passed * @param array $required * @throws InvalidConfigException */ protected function validateArrayFields(array $passed, array $required) { $undeclaredFields = array_diff_key($passed, $required); if (!empty($undeclaredFields)) { $asString = json_encode($undeclaredFields); throw new InvalidConfigException("Unknown options: {$asString}"); } } /** * Check entrees for duplicate names * @param array $keys * @throws InvalidConfigException */ protected function validateDuplicateNames(array $keys) { foreach ($keys as $key) { $names = []; foreach ($this->$key as $item) { if (!isset($item['name'])) { $item['name'] = ''; } if (isset($names[$item['name']])) { throw new InvalidConfigException("Duplicate name `{$item['name']}` in {$key}"); } $names[$item['name']] = true; } } } /** * Allow certain flexibility on connection configuration * @throws InvalidConfigException */ protected function normalizeConnections() { if (empty($this->connections)) { throw new InvalidConfigException('Option `connections` should have at least one entry.'); } if (ArrayHelper::isAssociative($this->connections)) { $this->connections[0] = $this->connections; } if (count($this->connections) === 1) { if (!isset($this->connections[0]['name'])) { $this->connections[0]['name'] = self::DEFAULT_CONNECTION_NAME; } } } /** * Merge passed config with extension defaults */ protected function completeWithDefaults() { $defaults = self::DEFAULTS; if (null === $this->auto_declare) { $this->auto_declare = $defaults['auto_declare']; } if (empty($this->logger)) { $this->logger = $defaults['logger']; } else { foreach ($defaults['logger'] as $key => $option) { if (!isset($this->logger[$key])) { $this->logger[$key] = $option; } } } $multi = ['connections', 'bindings', 'exchanges', 'queues', 'producers', 'consumers']; foreach ($multi as $key) { foreach ($this->$key as &$item) { $item = array_replace_recursive($defaults[$key][0], $item); } } } /** * Check if an entry with specific name exists in array * @param array $multidimentional * @param string $name * @return bool */ private function isNameExist(array $multidimentional, string $name) { if($name == '') { foreach ($multidimentional as $item) { if (!isset($item['name'])) { return true; } } return false; } $key = array_search($name, array_column($multidimentional, 'name'), true); if (is_int($key)) { return true; } return false; } }