Renamed Rabbit instances to use RabbitMq

2026-03-09 16:53:11 +08:00 · 2021-12-12 10:32:57 +01:00
parent cb1705b6e8
commit 8e5730f374
6 changed files with 21 additions and 20 deletions
--- a/module/Core/src/EventDispatcher/NotifyVisitToRabbitMq.php
+++ b/module/Core/src/EventDispatcher/NotifyVisitToRabbitMq.php
@@ -0,0 +1,102 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Shlinkio\Shlink\Core\EventDispatcher;
+
+use Doctrine\ORM\EntityManagerInterface;
+use PhpAmqpLib\Connection\AMQPStreamConnection;
+use PhpAmqpLib\Exchange\AMQPExchangeType;
+use PhpAmqpLib\Message\AMQPMessage;
+use Psr\Log\LoggerInterface;
+use Shlinkio\Shlink\Common\Rest\DataTransformerInterface;
+use Shlinkio\Shlink\Core\Entity\Visit;
+use Shlinkio\Shlink\Core\EventDispatcher\Event\VisitLocated;
+use Throwable;
+
+use function Shlinkio\Shlink\Common\json_encode;
+use function sprintf;
+
+class NotifyVisitToRabbitMq
+{
+    private const NEW_VISIT_QUEUE = 'https://shlink.io/new-visit';
+    private const NEW_ORPHAN_VISIT_QUEUE = 'https://shlink.io/new-orphan-visit';
+
+    public function __construct(
+        private AMQPStreamConnection $connection,
+        private EntityManagerInterface $em,
+        private LoggerInterface $logger,
+        private DataTransformerInterface $orphanVisitTransformer,
+        private bool $isEnabled,
+    ) {
+    }
+
+    public function __invoke(VisitLocated $shortUrlLocated): void
+    {
+        if (! $this->isEnabled) {
+            return;
+        }
+
+        $visitId = $shortUrlLocated->visitId();
+        $visit = $this->em->find(Visit::class, $visitId);
+
+        if ($visit === null) {
+            $this->logger->warning('Tried to notify RabbitMQ for visit with id "{visitId}", but it does not exist.', [
+                'visitId' => $visitId,
+            ]);
+            return;
+        }
+
+        if (! $this->connection->isConnected()) {
+            $this->connection->reconnect();
+        }
+
+        $queues = $this->determineQueuesToPublishTo($visit);
+        $message = $this->visitToMessage($visit);
+
+        try {
+            $channel = $this->connection->channel();
+
+            foreach ($queues as $queue) {
+                // Declare an exchange and a queue that will persist server restarts
+                $exchange = $queue; // We use the same name for the exchange and the queue
+                $channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, false, true, false);
+                $channel->queue_declare($queue, false, true, false, false);
+
+                // Bind the exchange and the queue together, and publish the message
+                $channel->queue_bind($queue, $exchange);
+                $channel->basic_publish($message, $exchange);
+            }
+
+            $channel->close();
+        } catch (Throwable $e) {
+            $this->logger->debug('Error while trying to notify RabbitMQ with new visit. {e}', ['e' => $e]);
+        } finally {
+            $this->connection->close();
+        }
+    }
+
+    /**
+     * @return string[]
+     */
+    private function determineQueuesToPublishTo(Visit $visit): array
+    {
+        if ($visit->isOrphan()) {
+            return [self::NEW_ORPHAN_VISIT_QUEUE];
+        }
+
+        return [
+            self::NEW_VISIT_QUEUE,
+            sprintf('%s/%s', self::NEW_VISIT_QUEUE, $visit->getShortUrl()?->getShortCode()),
+        ];
+    }
+
+    private function visitToMessage(Visit $visit): AMQPMessage
+    {
+        $messageBody = json_encode(! $visit->isOrphan() ? $visit : $this->orphanVisitTransformer->transform($visit));
+        return new AMQPMessage($messageBody, [
+            'content_type' => 'application/json',
+            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
+        ]);
+    }
+}