Automatic declare exchange when produce message (#6696)

src/DeclaredExchanges.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  [email protected]
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\Amqp;
+
+class DeclaredExchanges
+{
+    private static array $exchanges = [];
+
+    public static function add(string $exchange): void
+    {
+        self::$exchanges[$exchange] = true;
+    }
+
+    public static function remove(string $exchange): void
+    {
+        unset(self::$exchanges[$exchange]);
+    }
+
+    public static function has(string $exchange): bool
+    {
+        return isset(self::$exchanges[$exchange]);
+    }
+}

src/Listener/MainWorkerStartListener.php

@@ -14,6 +14,7 @@
 
 use Doctrine\Instantiator\Instantiator;
 use Hyperf\Amqp\Annotation\Producer;
+use Hyperf\Amqp\DeclaredExchanges;
 use Hyperf\Amqp\Message\ProducerMessageInterface;
 use Hyperf\Contract\ConfigInterface;
 use Hyperf\Contract\StdoutLoggerInterface;
@@ -70,13 +71,15 @@ public function process(object $event): void
                 $annotation->routingKey && $instance->setRoutingKey($annotation->routingKey);
                 try {
                     $producer->declare($instance);
+                    DeclaredExchanges::add($instance->getExchange());
                     $routingKey = $instance->getRoutingKey();
                     if (is_array($routingKey)) {
                         $routingKey = implode(',', $routingKey);
                     }
                     $this->logger->debug(sprintf('AMQP exchange[%s] and routingKey[%s] were created successfully.', $instance->getExchange(), $routingKey));
                 } catch (AMQPProtocolChannelException $e) {
                     $this->logger->debug('AMQPProtocolChannelException: ' . $e->getMessage());
+                    DeclaredExchanges::remove($instance->getExchange());
                     // Do nothing.
                 } catch (Throwable $exception) {
                     $this->logger->error((string) $exception);

src/Producer.php

@@ -42,10 +42,22 @@ private function produceMessage(ProducerMessageInterface $producerMessage, bool
                 $channel = $connection->getChannel();
             }
 
+            $exchange = $producerMessage->getExchange();
+
+            if (! DeclaredExchanges::has($exchange)) {
+                try {
+                    DeclaredExchanges::add($exchange);
+                    $this->declare($producerMessage, $channel);
+                } catch (Throwable $exception) {
+                    DeclaredExchanges::remove($exchange);
+                    throw $exception;
+                }
+            }
+
             $channel->set_ack_handler(function () use (&$result) {
                 $result = true;
             });
-            $channel->basic_publish($message, $producerMessage->getExchange(), $producerMessage->getRoutingKey());
+            $channel->basic_publish($message, $exchange, $producerMessage->getRoutingKey());
             $channel->wait_for_pending_acks_returns($timeout);
         } catch (Throwable $exception) {
             isset($channel) && $channel->close();