Skip to content

[Messenger] Setup queues once in AMQP #39608

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.

Sign up for GitHub

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

Merged
merged 1 commit into from
Jan 1, 2021
Merged

[Messenger] Setup queues once in AMQP #39608

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/Connection.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ public static function fromDsn(string $dsn, array $options = [], HttpClientInter
'wait_time' => (int) $options['wait_time'],
'poll_timeout' => $options['poll_timeout'],
'visibility_timeout' => $options['visibility_timeout'],
'auto_setup' => (bool) $options['auto_setup'],
'auto_setup' => filter_var($options['auto_setup'], \FILTER_VALIDATE_BOOLEAN),
'queue_name' => (string) $options['queue_name'],
];

Expand Down
18 changes: 18 additions & 0 deletions src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -421,6 +421,24 @@ public function testItCanDisableTheSetup()
$connection->publish('body');
}

public function testItSetupQueuesOnce()
{
$factory = new TestAmqpFactory(
$amqpConnection = $this->createMock(\AMQPConnection::class),
$amqpChannel = $this->createMock(\AMQPChannel::class),
$amqpQueue = $this->createMock(\AMQPQueue::class),
$amqpExchange = $this->createMock(\AMQPExchange::class)
);

$amqpExchange->expects($this->once())->method('declareExchange');
$amqpQueue->expects($this->once())->method('declareQueue');
$amqpQueue->expects($this->once())->method('bind');

$connection = Connection::fromDsn('amqp://localhost', ['auto_setup' => true], $factory);
$connection->publish('body');
$connection->publish('body');
}

public function testSetChannelPrefetchWhenSetup()
{
$factory = new TestAmqpFactory(
Expand Down
46 changes: 16 additions & 30 deletions src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,8 @@ class Connection
private $exchangeOptions;
private $queuesOptions;
private $amqpFactory;
private $autoSetupExchange;
private $autoSetup;

/**
* @var \AMQPChannel|null
Expand Down Expand Up @@ -112,6 +114,7 @@ public function __construct(array $connectionOptions, array $exchangeOptions, ar
'queue_name_pattern' => 'delay_%exchange_name%_%routing_key%_%delay%',
],
], $connectionOptions);
$this->autoSetupExchange = $this->autoSetup = $connectionOptions['auto_setup'] ?? true;
$this->exchangeOptions = $exchangeOptions;
$this->queuesOptions = $queuesOptions;
$this->amqpFactory = $amqpFactory ?: new AmqpFactory();
Expand Down Expand Up @@ -207,6 +210,9 @@ public static function fromDsn(string $dsn, array $options = [], AmqpFactory $am
$exchangeOptions = $amqpOptions['exchange'];
$queuesOptions = $amqpOptions['queues'];
unset($amqpOptions['queues'], $amqpOptions['exchange']);
if (isset($amqpOptions['auto_setup'])) {
$amqpOptions['auto_setup'] = filter_var($amqpOptions['auto_setup'], \FILTER_VALIDATE_BOOLEAN);
}

$queuesOptions = array_map(function ($queueOptions) {
if (!\is_array($queueOptions)) {
Expand Down Expand Up @@ -285,7 +291,7 @@ public function publish(string $body, array $headers = [], int $delayInMs = 0, A
return;
}

if ($this->shouldSetup()) {
if ($this->autoSetupExchange) {
$this->setupExchangeAndQueues();
}

Expand Down Expand Up @@ -347,7 +353,7 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string

private function setupDelay(int $delay, ?string $routingKey)
{
if ($this->shouldSetup()) {
if ($this->autoSetup) {
$this->setup(); // setup delay exchange and normal exchange for delay queue to DLX messages to
}

Expand Down Expand Up @@ -418,23 +424,12 @@ public function get(string $queueName): ?\AMQPEnvelope
{
$this->clearWhenDisconnected();

if ($this->shouldSetup()) {
if ($this->autoSetupExchange) {
$this->setupExchangeAndQueues();
}

try {
if (false !== $message = $this->queue($queueName)->get()) {
return $message;
}
} catch (\AMQPQueueException $e) {
if (404 === $e->getCode() && $this->shouldSetup()) {
// If we get a 404 for the queue, it means we need to set up the exchange & queue.
$this->setupExchangeAndQueues();

return $this->get($queueName);
}

throw $e;
if (false !== $message = $this->queue($queueName)->get()) {
return $message;
}

return null;
Expand All @@ -452,8 +447,11 @@ public function nack(\AMQPEnvelope $message, string $queueName, int $flags = \AM

public function setup(): void
{
$this->setupExchangeAndQueues();
if ($this->autoSetupExchange) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The setup seems broken now. The setup method is public because it's called by SetupableTransportInterface. Now if you disable auto_setup and want to setup the transports manually, then it does not actually setup the exchanges and queues anymore. But that is the whole point of the setup method: being able to setup manually when autosetup is not used.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in #40966

$this->setupExchangeAndQueues();
}
$this->getDelayExchange()->declareExchange();
$this->autoSetup = false;
}

private function setupExchangeAndQueues(): void
Expand All @@ -466,6 +464,7 @@ private function setupExchangeAndQueues(): void
$this->queue($queueName)->bind($this->exchangeOptions['name'], $bindingKey, $queueConfig['binding_arguments'] ?? []);
}
}
$this->autoSetupExchange = false;
}

/**
Expand Down Expand Up @@ -558,19 +557,6 @@ private function clearWhenDisconnected(): void
}
}

private function shouldSetup(): bool
{
if (!\array_key_exists('auto_setup', $this->connectionOptions)) {
return true;
}

if (\in_array($this->connectionOptions['auto_setup'], [false, 'false'], true)) {
return false;
}

return true;
}

private function getDefaultPublishRoutingKey(): ?string
{
return $this->exchangeOptions['default_publish_routing_key'] ?? null;
Expand Down