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');
     }
 }