php-enqueue · makasim · Jan 5, 2017
diff --git a/pkg/amqp-ext/Client/AmqpDriver.php b/pkg/amqp-ext/Client/AmqpDriver.php
@@ -2,6 +2,7 @@
 
 namespace  Enqueue\AmqpExt\Client;
 
+use Enqueue\AmqpExt\AmqpConnectionFactory;
 use Enqueue\AmqpExt\AmqpContext;
 use Enqueue\AmqpExt\AmqpMessage;
 use Enqueue\AmqpExt\AmqpQueue;
@@ -18,6 +19,11 @@
 
 class AmqpDriver implements DriverInterface
 {
+    /**
+     * @var AmqpConnectionFactory
+     */
+    private $connectionFactory;
+
     /**
      * @var AmqpContext
      */
@@ -39,13 +45,13 @@ class AmqpDriver implements DriverInterface
     private $priorityMap;
 
     /**
-     * @param AmqpContext       $context
+     * @param AmqpConnectionFactory       $connectionFactory
      * @param Config            $config
      * @param QueueMetaRegistry $queueMetaRegistry
      */
-    public function __construct(AmqpContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry)
+    public function __construct(AmqpConnectionFactory $connectionFactory, Config $config, QueueMetaRegistry $queueMetaRegistry)
     {
-        $this->context = $context;
+        $this->connectionFactory = $connectionFactory;
         $this->config = $config;
         $this->queueMetaRegistry = $queueMetaRegistry;
 
@@ -70,7 +76,7 @@ public function sendToRouter(Message $message)
         $topic = $this->createRouterTopic();
         $transportMessage = $this->createTransportMessage($message);
 
-        $this->context->createProducer()->send($topic, $transportMessage);
+        $this->getContext()->createProducer()->send($topic, $transportMessage);
     }
 
     /**
@@ -89,7 +95,7 @@ public function sendToProcessor(Message $message)
         $transportMessage = $this->createTransportMessage($message);
         $destination = $this->createQueue($queueName);
 
-        $this->context->createProducer()->send($destination, $transportMessage);
+        $this->getContext()->createProducer()->send($destination, $transportMessage);
     }
 
     /**
@@ -107,18 +113,18 @@ public function setupBroker(LoggerInterface $logger = null)
         $routerQueue = $this->createQueue($this->config->getRouterQueueName());
 
         $log('Declare router exchange: %s', $routerTopic->getTopicName());
-        $this->context->declareTopic($routerTopic);
+        $this->getContext()->declareTopic($routerTopic);
         $log('Declare router queue: %s', $routerQueue->getQueueName());
-        $this->context->declareQueue($routerQueue);
+        $this->getContext()->declareQueue($routerQueue);
         $log('Bind router queue to exchange: %s -> %s', $routerQueue->getQueueName(), $routerTopic->getTopicName());
-        $this->context->bind($routerTopic, $routerQueue);
+        $this->getContext()->bind($routerTopic, $routerQueue);
 
         // setup queues
         foreach ($this->queueMetaRegistry->getQueuesMeta() as $meta) {
             $queue = $this->createQueue($meta->getClientName());
 
             $log('Declare processor queue: %s', $queue->getQueueName());
-            $this->context->declareQueue($queue);
+            $this->getContext()->declareQueue($queue);
         }
     }
 
@@ -129,7 +135,7 @@ public function setupBroker(LoggerInterface $logger = null)
      */
     public function createQueue($queueName)
     {
-        $queue = $this->context->createQueue($this->config->createTransportQueueName($queueName));
+        $queue = $this->getContext()->createQueue($this->config->createTransportQueueName($queueName));
         $queue->addFlag(AMQP_DURABLE);
         $queue->setArguments(['x-max-priority' => 4]);
 
@@ -165,7 +171,7 @@ public function createTransportMessage(Message $message)
 
         $headers['delivery_mode'] = DeliveryMode::PERSISTENT;
 
-        $transportMessage = $this->context->createMessage();
+        $transportMessage = $this->getContext()->createMessage();
         $transportMessage->setBody($message->getBody());
         $transportMessage->setHeaders($headers);
         $transportMessage->setProperties($properties);
@@ -225,12 +231,24 @@ public function getConfig()
      */
     private function createRouterTopic()
     {
-        $topic = $this->context->createTopic(
+        $topic = $this->getContext()->createTopic(
             $this->config->createTransportRouterTopicName($this->config->getRouterTopicName())
         );
         $topic->setType(AMQP_EX_TYPE_FANOUT);
         $topic->addFlag(AMQP_DURABLE);
 
         return $topic;
     }
+
+    /**
+     * @return AmqpContext
+     */
+    protected function getContext()
+    {
+        if (false == $this->context) {
+            $this->context = $this->connectionFactory->createContext();
+        }
+
+        return $this->context;
+    }
 }
diff --git a/pkg/amqp-ext/Client/RabbitMqDriver.php b/pkg/amqp-ext/Client/RabbitMqDriver.php
@@ -2,7 +2,7 @@
 
 namespace  Enqueue\AmqpExt\Client;
 
-use Enqueue\AmqpExt\AmqpContext;
+use Enqueue\AmqpExt\AmqpConnectionFactory;
 use Enqueue\AmqpExt\AmqpMessage;
 use Enqueue\AmqpExt\AmqpQueue;
 use Enqueue\AmqpExt\AmqpTopic;
@@ -16,11 +16,6 @@
 
 class RabbitMqDriver extends AmqpDriver
 {
-    /**
-     * @var AmqpContext
-     */
-    private $context;
-
     /**
      * @var Config
      */
@@ -32,16 +27,15 @@ class RabbitMqDriver extends AmqpDriver
     private $queueMetaRegistry;
 
     /**
-     * @param AmqpContext       $context
+     * @param AmqpConnectionFactory $connectionFactory
      * @param Config            $config
      * @param QueueMetaRegistry $queueMetaRegistry
      */
-    public function __construct(AmqpContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry)
+    public function __construct(AmqpConnectionFactory $connectionFactory, Config $config, QueueMetaRegistry $queueMetaRegistry)
     {
-        parent::__construct($context, $config, $queueMetaRegistry);
+        parent::__construct($connectionFactory, $config, $queueMetaRegistry);
 
         $this->config = $config;
-        $this->context = $context;
         $this->queueMetaRegistry = $queueMetaRegistry;
     }
 
@@ -65,7 +59,7 @@ public function sendToProcessor(Message $message)
             $destination = $this->createDelayedTopic($destination);
         }
 
-        $this->context->createProducer()->send($destination, $transportMessage);
+        $this->getContext()->createProducer()->send($destination, $transportMessage);
     }
 
     /**
@@ -129,10 +123,10 @@ public function setupBroker(LoggerInterface $logger = null)
                 $delayTopic = $this->createDelayedTopic($queue);
 
                 $log('Declare delay exchange: %s', $delayTopic->getTopicName());
-                $this->context->declareTopic($delayTopic);
+                $this->getContext()->declareTopic($delayTopic);
 
                 $log('Bind processor queue to delay exchange: %s -> %s', $queue->getQueueName(), $delayTopic->getTopicName());
-                $this->context->bind($delayTopic, $queue);
+                $this->getContext()->bind($delayTopic, $queue);
             }
         }
     }
@@ -147,7 +141,7 @@ private function createDelayedTopic(AmqpQueue $queue)
         $queueName = $queue->getQueueName();
 
         // in order to use delay feature make sure the rabbitmq_delayed_message_exchange plugin is installed.
-        $delayTopic = $this->context->createTopic($queueName.'.delayed');
+        $delayTopic = $this->getContext()->createTopic($queueName.'.delayed');
         $delayTopic->setRoutingKey($queueName);
         $delayTopic->setType('x-delayed-message');
         $delayTopic->addFlag(AMQP_DURABLE);

diff --git a/pkg/amqp-ext/Symfony/AmqpTransportFactory.php b/pkg/amqp-ext/Symfony/AmqpTransportFactory.php
@@ -104,7 +104,7 @@ public function createDriver(ContainerBuilder $container, array $config)
     {
         $driver = new Definition(AmqpDriver::class);
         $driver->setArguments([
-            new Reference(sprintf('enqueue.transport.%s.context', $this->getName())),
+            new Reference(sprintf('enqueue.transport.%s.connection_factory', $this->getName())),
             new Reference('enqueue.client.config'),
             new Reference('enqueue.client.meta.queue_meta_registry'),
         ]);

diff --git a/pkg/enqueue-bundle/Resources/config/client.yml b/pkg/enqueue-bundle/Resources/config/client.yml
@@ -56,7 +56,7 @@ services:
         class: 'Enqueue\Consumption\QueueConsumer'
         public: false
         arguments:
-            - '@enqueue.transport.context'
+            - '@enqueue.transport.connection_factory'
             - '@enqueue.consumption.extensions'
 
     enqueue.client.consume_messages_command:

diff --git a/pkg/enqueue-bundle/Resources/config/services.yml b/pkg/enqueue-bundle/Resources/config/services.yml
@@ -8,7 +8,7 @@ services:
     enqueue.consumption.queue_consumer:
         class: 'Enqueue\Consumption\QueueConsumer'
         arguments:
-            - '@enqueue.transport.context'
+            - '@enqueue.transport.connection_factory'
             - '@enqueue.consumption.extensions'
 
     enqueue.command.consume_messages:

diff --git a/pkg/enqueue/Client/NullDriver.php b/pkg/enqueue/Client/NullDriver.php
@@ -3,30 +3,37 @@
 namespace Enqueue\Client;
 
 use Enqueue\Psr\Message as TransportMessage;
+use Enqueue\Transport\Null\NullConnectionFactory;
 use Enqueue\Transport\Null\NullContext;
 use Enqueue\Transport\Null\NullMessage;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 
 class NullDriver implements DriverInterface
 {
+    /**
+     * @var NullConnectionFactory
+     */
+    private $connectionFactory;
+
     /**
      * @var NullContext
      */
-    protected $context;
+    private $context;
 
     /**
      * @var Config
      */
-    protected $config;
+    private $config;
 
     /**
-     * @param NullContext $session
+     * @param NullConnectionFactory $connectionFactory
      * @param Config      $config
      */
-    public function __construct(NullContext $session, Config $config)
+    public function __construct(NullConnectionFactory $connectionFactory, Config $config)
     {
-        $this->context = $session;
+        $this->connectionFactory = $connectionFactory;
+        $this->context = $connectionFactory->createContext();
         $this->config = $config;
     }
 
@@ -138,6 +145,6 @@ public function sendToProcessor(Message $message)
     public function setupBroker(LoggerInterface $logger = null)
     {
         $logger ?: new NullLogger();
-        $logger->debug('[NullDriver] setup broker');
+        $logger->debug('[NullDriver] nothing to setup');
     }
 }
diff --git a/pkg/enqueue/Consumption/QueueConsumer.php b/pkg/enqueue/Consumption/QueueConsumer.php
@@ -6,6 +6,7 @@
 use Enqueue\Consumption\Exception\InvalidArgumentException;
 use Enqueue\Consumption\Exception\LogicException;
 use Enqueue\Psr\Consumer;
+use Enqueue\Psr\ConnectionFactory as PsrConnectionFactory;
 use Enqueue\Psr\Context as PsrContext;
 use Enqueue\Psr\Processor;
 use Enqueue\Psr\Queue;
@@ -14,6 +15,11 @@
 
 class QueueConsumer
 {
+    /**
+     * @var PsrConnectionFactory
+     */
+    private $psrConnectionFactory;
+
     /**
      * @var PsrContext
      */
@@ -39,30 +45,22 @@ class QueueConsumer
     private $idleMicroseconds;
 
     /**
-     * @param PsrContext                             $psrContext
+     * @param PsrConnectionFactory                   $psrConnectionFactory
      * @param ExtensionInterface|ChainExtension|null $extension
      * @param int                                    $idleMicroseconds 100ms by default
      */
     public function __construct(
-        PsrContext $psrContext,
+        PsrConnectionFactory $psrConnectionFactory,
         ExtensionInterface $extension = null,
         $idleMicroseconds = 100000
     ) {
-        $this->psrContext = $psrContext;
+        $this->psrConnectionFactory = $psrConnectionFactory;
         $this->extension = $extension;
         $this->idleMicroseconds = $idleMicroseconds;
 
         $this->boundProcessors = [];
     }
 
-    /**
-     * @return PsrContext
-     */
-    public function getPsrContext()
-    {
-        return $this->psrContext;
-    }
-
     /**
      * @param Queue|string       $queue
      * @param Processor|callable $processor
@@ -72,7 +70,7 @@ public function getPsrContext()
     public function bind($queue, $processor)
     {
         if (is_string($queue)) {
-            $queue = $this->psrContext->createQueue($queue);
+            $queue = $this->getPsrContext()->createQueue($queue);
         }
         if (is_callable($processor)) {
             $processor = new CallbackProcessor($processor);
@@ -103,19 +101,22 @@ public function bind($queue, $processor)
      */
     public function consume(ExtensionInterface $runtimeExtension = null)
     {
+        $psrContext = $this->getPsrContext();
+
         /** @var Consumer[] $messageConsumers */
         $messageConsumers = [];
         /** @var \Enqueue\Psr\Queue $queue */
         foreach ($this->boundProcessors as list($queue, $processor)) {
-            $messageConsumers[$queue->getQueueName()] = $this->psrContext->createConsumer($queue);
+            $messageConsumers[$queue->getQueueName()] = $psrContext->createConsumer($queue);
         }
 
         $extension = $this->extension ?: new ChainExtension([]);
         if ($runtimeExtension) {
             $extension = new ChainExtension([$extension, $runtimeExtension]);
         }
 
-        $context = new Context($this->psrContext);
+
+        $context = new Context($psrContext);
         $extension->onStart($context);
 
         $logger = $context->getLogger() ?: new NullLogger();
@@ -129,7 +130,7 @@ public function consume(ExtensionInterface $runtimeExtension = null)
 
                     $messageConsumer = $messageConsumers[$queue->getQueueName()];
 
-                    $context = new Context($this->psrContext);
+                    $context = new Context($psrContext);
                     $context->setLogger($logger);
                     $context->setPsrQueue($queue);
                     $context->setPsrConsumer($messageConsumer);
@@ -143,7 +144,7 @@ public function consume(ExtensionInterface $runtimeExtension = null)
                 $context->setExecutionInterrupted(true);
 
                 $extension->onInterrupted($context);
-                $this->psrContext->close();
+                $psrContext->close();
 
                 return;
             } catch (\Exception $exception) {
@@ -152,17 +153,29 @@ public function consume(ExtensionInterface $runtimeExtension = null)
 
                 try {
                     $this->onInterruptionByException($extension, $context);
-                    $this->psrContext->close();
+                    $psrContext->close();
                 } catch (\Exception $e) {
                     // for some reason finally does not work here on php5.5
-                    $this->psrContext->close();
+                    $psrContext->close();
 
                     throw $e;
                 }
             }
         }
     }
 
+    /**
+     * @return PsrContext
+     */
+    public function getPsrContext()
+    {
+        if (false == $this->psrContext) {
+            $this->psrContext = $this->psrConnectionFactory->createContext();
+        }
+
+        return $this->psrContext;
+    }
+
     /**
      * @param ExtensionInterface $extension
      * @param Context            $context
@@ -176,6 +189,7 @@ protected function doConsume(ExtensionInterface $extension, Context $context)
         $processor = $context->getPsrProcessor();
         $consumer = $context->getPsrConsumer();
         $logger = $context->getLogger();
+        $psrContext = $context->getPsrContext();
 
         $extension->onBeforeReceive($context);
 
@@ -193,7 +207,7 @@ protected function doConsume(ExtensionInterface $extension, Context $context)
 
             $extension->onPreReceived($context);
             if (!$context->getResult()) {
-                $result = $processor->process($message, $this->psrContext);
+                $result = $processor->process($message, $psrContext);
                 $context->setResult($result);
             }
 

diff --git a/pkg/enqueue/Symfony/DefaultTransportFactory.php b/pkg/enqueue/Symfony/DefaultTransportFactory.php
@@ -42,6 +42,12 @@ public function addConfiguration(ArrayNodeDefinition $builder)
      */
     public function createContext(ContainerBuilder $container, array $config)
     {
+        $factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
+        $aliasId = sprintf('enqueue.transport.%s.connection_factory', $config['alias']);
+
+        $container->setAlias($factoryId, $aliasId);
+        $container->setAlias('enqueue.transport.connection_factory', $factoryId);
+
         $contextId = sprintf('enqueue.transport.%s.context', $this->getName());
         $aliasId = sprintf('enqueue.transport.%s.context', $config['alias']);
 

diff --git a/pkg/enqueue/Symfony/NullTransportFactory.php b/pkg/enqueue/Symfony/NullTransportFactory.php
@@ -3,6 +3,7 @@
 namespace Enqueue\Symfony;
 
 use Enqueue\Client\NullDriver;
+use Enqueue\Transport\Null\NullConnectionFactory;
 use Enqueue\Transport\Null\NullContext;
 use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
@@ -36,8 +37,13 @@ public function addConfiguration(ArrayNodeDefinition $builder)
      */
     public function createContext(ContainerBuilder $container, array $config)
     {
+        $factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
+        $factory = new Definition(NullConnectionFactory::class);
+        $container->setDefinition($factoryId, $factory);
+
         $contextId = sprintf('enqueue.transport.%s.context', $this->getName());
         $context = new Definition(NullContext::class);
+        $context->setFactory([new Reference($factoryId), 'createContext']);
 
         $container->setDefinition($contextId, $context);
 
@@ -51,7 +57,7 @@ public function createDriver(ContainerBuilder $container, array $config)
     {
         $driver = new Definition(NullDriver::class);
         $driver->setArguments([
-            new Reference(sprintf('enqueue.transport.%s.context', $this->getName())),
+            new Reference(sprintf('enqueue.transport.%s.connection_factory', $this->getName())),
             new Reference('enqueue.client.config'),
         ]);
 

diff --git a/pkg/stomp/Client/RabbitMqStompDriver.php b/pkg/stomp/Client/RabbitMqStompDriver.php
@@ -7,6 +7,7 @@
 use Enqueue\Client\MessagePriority;
 use Enqueue\Client\Meta\QueueMetaRegistry;
 use Enqueue\Psr\Message as TransportMessage;
+use Enqueue\Stomp\StompConnectionFactory;
 use Enqueue\Stomp\StompContext;
 use Enqueue\Stomp\StompDestination;
 use Enqueue\Stomp\StompMessage;
@@ -15,11 +16,6 @@
 
 class RabbitMqStompDriver extends StompDriver
 {
-    /**
-     * @var StompContext
-     */
-    private $context;
-
     /**
      * @var Config
      */
@@ -41,16 +37,15 @@ class RabbitMqStompDriver extends StompDriver
     private $queueMetaRegistry;
 
     /**
-     * @param StompContext      $context
+     * @param StompConnectionFactory      $connectionFactory
      * @param Config            $config
      * @param QueueMetaRegistry $queueMetaRegistry
      * @param ManagementClient  $management
      */
-    public function __construct(StompContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry, ManagementClient $management)
+    public function __construct(StompConnectionFactory $connectionFactory, Config $config, QueueMetaRegistry $queueMetaRegistry, ManagementClient $management)
     {
-        parent::__construct($context, $config);
+        parent::__construct($connectionFactory, $config);
 
-        $this->context = $context;
         $this->config = $config;
         $this->queueMetaRegistry = $queueMetaRegistry;
         $this->management = $management;
@@ -160,7 +155,7 @@ public function sendToProcessor(Message $message)
             $destination = $this->createDelayedTopic($destination);
         }
 
-        $this->context->createProducer()->send($destination, $transportMessage);
+        $this->getContext()->createProducer()->send($destination, $transportMessage);
     }
 
     /**
@@ -258,7 +253,7 @@ public function setupBroker(LoggerInterface $logger = null)
     private function createDelayedTopic(StompDestination $queue)
     {
         // in order to use delay feature make sure the rabbitmq_delayed_message_exchange plugin is installed.
-        $destination = $this->context->createTopic($queue->getStompName().'.delayed');
+        $destination = $this->getContext()->createTopic($queue->getStompName().'.delayed');
         $destination->setType(StompDestination::TYPE_EXCHANGE);
         $destination->setDurable(true);
         $destination->setAutoDelete(false);

diff --git a/pkg/stomp/Client/StompDriver.php b/pkg/stomp/Client/StompDriver.php
@@ -6,6 +6,7 @@
 use Enqueue\Client\DriverInterface;
 use Enqueue\Client\Message;
 use Enqueue\Psr\Message as TransportMessage;
+use Enqueue\Stomp\StompConnectionFactory;
 use Enqueue\Stomp\StompContext;
 use Enqueue\Stomp\StompDestination;
 use Enqueue\Stomp\StompMessage;
@@ -14,6 +15,11 @@
 
 class StompDriver implements DriverInterface
 {
+    /**
+     * @var StompConnectionFactory
+     */
+    private $connectionFactory;
+
     /**
      * @var StompContext
      */
@@ -25,12 +31,12 @@ class StompDriver implements DriverInterface
     private $config;
 
     /**
-     * @param StompContext $context
-     * @param Config       $config
+     * @param StompConnectionFactory $connectionFactory
+     * @param Config $config
      */
-    public function __construct(StompContext $context, Config $config)
+    public function __construct(StompConnectionFactory $connectionFactory, Config $config)
     {
-        $this->context = $context;
+        $this->connectionFactory = $connectionFactory;
         $this->config = $config;
     }
 
@@ -46,7 +52,7 @@ public function sendToRouter(Message $message)
         $topic = $this->createRouterTopic();
         $transportMessage = $this->createTransportMessage($message);
 
-        $this->context->createProducer()->send($topic, $transportMessage);
+        $this->getContext()->createProducer()->send($topic, $transportMessage);
     }
 
     /**
@@ -65,7 +71,7 @@ public function sendToProcessor(Message $message)
         $transportMessage = $this->createTransportMessage($message);
         $destination = $this->createQueue($queueName);
 
-        $this->context->createProducer()->send($destination, $transportMessage);
+        $this->getContext()->createProducer()->send($destination, $transportMessage);
     }
 
     /**
@@ -87,7 +93,7 @@ public function createTransportMessage(Message $message)
         $headers = $message->getHeaders();
         $headers['content-type'] = $message->getContentType();
 
-        $transportMessage = $this->context->createMessage();
+        $transportMessage = $this->getContext()->createMessage();
         $transportMessage->setHeaders($headers);
         $transportMessage->setPersistent(true);
         $transportMessage->setBody($message->getBody());
@@ -137,7 +143,7 @@ public function createClientMessage(TransportMessage $message)
      */
     public function createQueue($queueName)
     {
-        $queue = $this->context->createQueue($this->config->createTransportQueueName($queueName));
+        $queue = $this->getContext()->createQueue($this->config->createTransportQueueName($queueName));
         $queue->setDurable(true);
         $queue->setAutoDelete(false);
         $queue->setExclusive(false);
@@ -158,12 +164,24 @@ public function getConfig()
      */
     private function createRouterTopic()
     {
-        $topic = $this->context->createTopic(
+        $topic = $this->getContext()->createTopic(
             $this->config->createTransportRouterTopicName($this->config->getRouterTopicName())
         );
         $topic->setDurable(true);
         $topic->setAutoDelete(false);
 
         return $topic;
     }
+
+    /**
+     * @return StompContext
+     */
+    protected function getContext()
+    {
+        if (false == $this->context) {
+            $this->context = $this->connectionFactory->createContext();
+        }
+
+        return $this->context;
+    }
 }
diff --git a/pkg/stomp/Symfony/RabbitMqStompTransportFactory.php b/pkg/stomp/Symfony/RabbitMqStompTransportFactory.php
@@ -60,7 +60,7 @@ public function createDriver(ContainerBuilder $container, array $config)
 
         $driver = new Definition(RabbitMqStompDriver::class);
         $driver->setArguments([
-            new Reference(sprintf('enqueue.transport.%s.context', $this->getName())),
+            new Reference(sprintf('enqueue.transport.%s.connection_factory', $this->getName())),
             new Reference('enqueue.client.config'),
             new Reference('enqueue.client.meta.queue_meta_registry'),
             new Reference($managementId),

diff --git a/pkg/stomp/Symfony/StompTransportFactory.php b/pkg/stomp/Symfony/StompTransportFactory.php
@@ -71,7 +71,7 @@ public function createDriver(ContainerBuilder $container, array $config)
     {
         $driver = new Definition(StompDriver::class);
         $driver->setArguments([
-            new Reference(sprintf('enqueue.transport.%s.context', $this->getName())),
+            new Reference(sprintf('enqueue.transport.%s.connection_factory', $this->getName())),
             new Reference('enqueue.client.config'),
         ]);