vendor/gos/web-socket-bundle/src/Pusher/Amqp/AmqpPusher.php line 11

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace Gos\Bundle\WebSocketBundle\Pusher\Amqp;
  5. use Gos\Bundle\WebSocketBundle\Pusher\AbstractPusher;
  6. use Gos\Bundle\WebSocketBundle\Pusher\Message;
  7. use Gos\Bundle\WebSocketBundle\Router\WampRouter;
  8. use Symfony\Component\OptionsResolver\OptionsResolver;
  9. use Symfony\Component\Serializer\SerializerInterface;
  11. 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.', AmqpPusher::class);
  13. /**
  14.  * @deprecated to be removed in 4.0, use the symfony/messenger component instead
  15.  */
  16. final class AmqpPusher extends AbstractPusher
  17. {
  18.     private \AMQPConnection $connection;
  19.     private \AMQPExchange $exchange;
  20.     private AmqpConnectionFactoryInterface $connectionFactory;
  22.     public function __construct(
  23.         WampRouter $router,
  24.         SerializerInterface $serializer,
  25.         AmqpConnectionFactoryInterface $connectionFactory
  26.     ) {
  27.         parent::__construct($router, $serializer);
  29.         $this->connectionFactory = $connectionFactory;
  30.     }
  32.     protected function doPush(Message $message, array $context): void
  33.     {
  34.         if (false === $this->connected) {
  35.             $this->connection = $this->connectionFactory->createConnection();
  36.             $this->connection->connect();
  38.             $this->exchange = $this->connectionFactory->createExchange($this->connection);
  40.             $this->setConnected();
  41.         }
  43.         $resolver = new OptionsResolver();
  45.         $resolver->setDefaults(
  46.             [
  47.                 'routing_key' => '',
  48.                 'publish_flags' => AMQP_NOPARAM,
  49.                 'attributes' => [],
  50.             ]
  51.         );
  53.         $context = $resolver->resolve($context);
  55.         $this->exchange->publish(
  56.             $this->serializer->serialize($message, 'json'),
  57.             $context['routing_key'],
  58.             $context['publish_flags'],
  59.             $context['attributes']
  60.         );
  61.     }
  63.     public function close(): void
  64.     {
  65.         if (false === $this->isConnected()) {
  66.             return;
  67.         }
  69.         $this->connection->disconnect();
  70.     }
  71. }