8000 [AmqpMessenger] Allow using the default amqp exchange · Issue #45784 · symfony/symfony · GitHub
[go: up one dir, main page]
Skip to content

[AmqpMessenger] Allow using the default amqp exchange #45784

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

Jump to bottom
Closed
d-ph opened this issue Mar 18, 2022 · 11 comments · Fixed by #54275
Closed

[AmqpMessenger] Allow using the default amqp exchange #45784

d-ph opened this issue Mar 18, 2022 · 11 comments · Fixed by #54275
Labels
Feature Messenger

Comments

@d-ph
Copy link
d-ph commented Mar 18, 2022

Description

I would like to request that the AmqpMessenger allows using the default AMQP Exchange. I.e. a setup, where there isn't a pair of an Exchange and a Queue for each Messenger's Transport, but rather a single Exchange (the default one) and a separate Queue for each Transport.

This is already supported in AMQP, RabbitMQ and php/ext-amqp -- one just has to use the special, "well-known" Exchange name '' (an empty string) to indicate to the AMQP protocol that the default Exchange is desired to be used.

Adding this feature to AmqpMessenger requires that the AMQP Exchange is not attempted to be declared (or bound to any Queue), if its name is an empty string. In other words, the required change is minimal. Please see the before/after in the examples below.

Apart from the code change, it might be worthwhile to mention this use case in the online documentation, in the "exchange[name]" option description.

Thank you.

Example

The default AMQP Exchange is directing the queue messages to Queues based on the supplied routing key when dispatching a queue message. For this reason, it's best to use the following Messenger configuration at a minimum:

transports:
  my_queue:
    dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
    options:
      exchange:
        # Indicates that the default Exchange is to be used
        name: ''
        # Makes the default Exchange direct the queue messages to the desired Queue
        default_publish_routing_key: my_queue

      queues:
        my_queue: {}

The current code always attempts to declare the Exchange. The following 2 code changes are needed to make it not do that if the exchange name is an empty string:

  1. https://github.com/symfony/amqp-messenger/blob/3c6563c2a6d5c343281fc35922498afb909ea13a/Transport/Connection.php#L522
public function exchange(): \AMQPExchange
{
    if (!isset($this->amqpExchange)) {
        $this->amqpExchange = $this->amqpFactory->createExchange($this->channel());
        $this->amqpExchange->setName($this->exchangeOptions['name']);
-        $this->amqpExchange->setType($this->exchangeOptions['type'] ?? \AMQP_EX_TYPE_FANOUT);
-        $this->amqpExchange->setFlags($this->exchangeOptions['flags'] ?? \AMQP_DURABLE);
-
-        if (isset($this->exchangeOptions['arguments'])) {
-            $this->amqpExchange->setArguments($this->exchangeOptions['arguments']);
-        }
+
+        if ($this->exchangeOptions['name'] !== '') {
+            $this->amqpExchange->setType($this->exchangeOptions['type'] ?? \AMQP_EX_TYPE_FANOUT);
+            $this->amqpExchange->setFlags($this->exchangeOptions['flags'] ?? \AMQP_DURABLE);
+
+            if (isset($this->exchangeOptions['arguments'])) {
+                $this->amqpExchange->setArguments($this->exchangeOptions['arguments']);
+            }
+        }

    }

    return $this->amqpExchange;
}
  1. https://github.com/symfony/amqp-messenger/blob/3c6563c2a6d5c343281fc35922498afb909ea13a/Transport/Connection.php#L447
private function setupExchangeAndQueues(): void
{
-    $this->exchange()->declareExchange();
+    $this->exchange();
+    if ($this->exchangeOptions['name'] !== '') {
+        $this->exchange()->declareExchange();
+    }

    foreach ($this->queuesOptions as $queueName => $queueConfig) {
        $this->queue($queueName)->declareQueue();
-        foreach ($queueConfig['binding_keys'] ?? [null] as $bindingKey) {
-            $this->queue($queueName)->bind($this->exchangeOptions['name'], $bindingKey, $queueConfig['binding_arguments'] ?? []);
-        }
+        if ($this->exchangeOptions['name'] !== '') {
+           foreach ($queueConfig['binding_keys'] ?? [null] as $bindingKey) {
+               $this->queue($queueName)->bind($this->exchangeOptions['name'], $bindingKey, $queueConfig['binding_arguments'] ?? []);
+           }
+        }
    }
    $this->autoSetupExchange = false;
}
@carsonbot
Copy link

Thank you for this suggestion.
There has not been a lot of activity here for a while. Would you still like to see this feature?

@d-ph
Copy link
Author
d-ph commented Oct 31, 2022

👍

@carsonbot carsonbot removed the Stalled label Oct 31, 2022
@carsonbot
Copy link

Thank you for this suggestion.
There has not been a lot of activity here for a while. Would you still like to see this feature?

@ilya-levin-lokalise
Copy link

👍

@carsonbot carsonbot removed the Stalled label Jun 14, 2023
@carsonbot
Copy link

Thank you for this suggestion.
There has not been a lot of activity here for a while. Would you still like to see this feature?

@ilya-levin-lokalise
Copy link

Yes, this feature makes sense.

@carsonbot carsonbot removed the Stalled label Dec 18, 2023
@ilyachase
Copy link
Contributor

The default exchange, in RabbitMQ, does not allow bind/unbind operations. Binding operations to the default exchange will result in an error.

I'm wondering if we'll need to manage this part in Symfony code as well, or if amqp extension will throw an exception by itself

@ilyachase
Copy link
Contributor

@d-ph also, could you elaborate if you intentionally put this in the configuration example:

queues:
        my_queue: {}

I'm trying to understand if we still need to declare queues when using the default exchange or they are declared automagically when a message is sent there.
Going to check locally but asking because maybe we have to manage this logic in the code as well (e.g. not to declare queues automatically)

@d-ph
Copy link
Author
d-ph commented Mar 10, 2024

Hi. Yes, queues still have to be declared in the configuration as usual. RabbitMQ will not create those queues automatically. In fact, one "nuisance" of using the default exchange, is that if a queue with a given name does not exist, then all messages/tasks sent to that non-existent queue will get silently ignored. So yes, I left that piece of configuration in the example above to indicate that it's still needed (at least at the time the example was written).

@marvinhinz
Copy link

We spent a few hours trying to figure out how to use the default exchange :D +1 for this idea

@ilyachase
Copy link
Contributor

I have prepared a PR and it's working, now writing tests. Let's see if we can get this one through 👍

@ilyachase ilyachase mentioned this issue Mar 13, 2024
Merged
@fabpot fabpot closed this as completed in 23e2230 Mar 28, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
Feature Messenger
Projects
None yet
Milestone
No milestone
Development

Successfully merging a pull request may close this issue.

[Messenger] [Amqp] Add default exchange support ilyachase/symfony
6 participants
@xabbuh @ilyachase @d-ph @carsonbot @marvinhinz @ilya-levin-lokalise
0