vendor/gos/web-socket-bundle/src/Pusher/Amqp/AmqpConnectionFactory.php line 8

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace Gos\Bundle\WebSocketBundle\Pusher\Amqp;
  5. use Gos\Bundle\WebSocketBundle\Pusher\Exception\PusherUnsupportedException;
  6. use Symfony\Component\OptionsResolver\OptionsResolver;
  8. trigger_deprecation('gos/web-socket-bundle', '3.1', 'The "%s" class is deprecated and will be removed in 4.0, use the symfony/messenger component instead.', AmqpConnectionFactory::class);
  10. /**
  11.  * @deprecated to be removed in 4.0, use the symfony/messenger component instead
  12.  */
  13. final class AmqpConnectionFactory implements AmqpConnectionFactoryInterface
  14. {
  15.     private array $config;
  17.     public function __construct(array $config)
  18.     {
  19.         $this->config = $this->resolveConfig($config);
  20.     }
  22.     /**
  23.      * @throws PusherUnsupportedException if the AMQP pusher is not supported in the current environment
  24.      */
  25.     public function createConnection(): \AMQPConnection
  26.     {
  27.         if (!$this->isSupported()) {
  28.             throw new PusherUnsupportedException('The AMQP pusher requires the PHP amqp extension.');
  29.         }
  31.         return new \AMQPConnection($this->config);
  32.     }
  34.     public function createExchange(\AMQPConnection $connection): \AMQPExchange
  35.     {
  36.         $exchange = new \AMQPExchange(new \AMQPChannel($connection));
  37.         $exchange->setName($this->config['exchange_name']);
  38.         $exchange->setType(AMQP_EX_TYPE_DIRECT);
  39.         $exchange->setFlags(AMQP_DURABLE);
  40.         $exchange->declareExchange();
  42.         return $exchange;
  43.     }
  45.     public function createQueue(\AMQPConnection $connection): \AMQPQueue
  46.     {
  47.         $queue = new \AMQPQueue(new \AMQPChannel($connection));
  48.         $queue->setName($this->config['queue_name']);
  49.         $queue->setFlags(AMQP_DURABLE);
  50.         $queue->declareQueue();
  52.         return $queue;
  53.     }
  55.     public function isSupported(): bool
  56.     {
  57.         return \extension_loaded('amqp');
  58.     }
  60.     private function resolveConfig(array $config): array
  61.     {
  62.         $resolver = new OptionsResolver();
  64.         $resolver->setRequired(
  65.             [
  66.                 'host',
  67.                 'port',
  68.                 'login',
  69.                 'password',
  70.             ]
  71.         );
  73.         $resolver->setDefaults(
  74.             [
  75.                 'vhost' => '/',
  76.                 'read_timeout' => 0,
  77.                 'write_timeout' => 0,
  78.                 'connect_timeout' => 0,
  79.                 'queue_name' => 'gos_websocket',
  80.                 'exchange_name' => 'gos_websocket_exchange',
  81.             ]
  82.         );
  84.         $resolver->setAllowedTypes('host', 'string');
  85.         $resolver->setAllowedTypes('port', ['string', 'integer']);
  86.         $resolver->setAllowedTypes('login', 'string');
  87.         $resolver->setAllowedTypes('password', 'string');
  88.         $resolver->setAllowedTypes('vhost', 'string');
  89.         $resolver->setAllowedTypes('read_timeout', 'integer');
  90.         $resolver->setAllowedTypes('write_timeout', 'integer');
  91.         $resolver->setAllowedTypes('connect_timeout', 'integer');
  92.         $resolver->setAllowedTypes('queue_name', 'string');
  93.         $resolver->setAllowedTypes('exchange_name', 'string');
  95.         return $resolver->resolve($config);
  96.     }
  97. }