Skip to content

Use connection factory instead of context everywhere. Be lazy #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 30 additions & 12 deletions pkg/amqp-ext/Client/AmqpDriver.php
Original file line number Diff line number Diff line change
@@ -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;
}
}
22 changes: 8 additions & 14 deletions pkg/amqp-ext/Client/RabbitMqDriver.php
Original file line number Diff line number Diff line change
@@ -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);
2 changes: 1 addition & 1 deletion pkg/amqp-ext/Symfony/AmqpTransportFactory.php
Original file line number Diff line number Diff line change
@@ -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'),
]);
2 changes: 1 addition & 1 deletion pkg/enqueue-bundle/Resources/config/client.yml
Original file line number Diff line number Diff line change
@@ -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:
2 changes: 1 addition & 1 deletion pkg/enqueue-bundle/Resources/config/services.yml
Original file line number Diff line number Diff line change
@@ -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:
19 changes: 13 additions & 6 deletions pkg/enqueue/Client/NullDriver.php
Original file line number Diff line number Diff line change
@@ -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');
}
}
Loading