BC-XX Prevent spamming the queues by ignoring own messages.

src/Contract/MonitoringMessageInterface.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BytesCommerce\ZabbixApi\Contract;
+
+interface MonitoringMessageInterface
+{
+}

src/Message/PushEventMessage.php

@@ -4,10 +4,11 @@
 
 namespace BytesCommerce\ZabbixApi\Message;
 
+use BytesCommerce\ZabbixApi\Contract\MonitoringMessageInterface;
 use Symfony\Component\Messenger\Attribute\AsMessage;
 
 #[AsMessage('async')]
-final readonly class PushEventMessage
+final readonly class PushEventMessage implements MonitoringMessageInterface
 {
     public function __construct(
         public string $key,

src/Message/PushMetricMessage.php

@@ -4,10 +4,11 @@
 
 namespace BytesCommerce\ZabbixApi\Message;
 
+use BytesCommerce\ZabbixApi\Contract\MonitoringMessageInterface;
 use Symfony\Component\Messenger\Attribute\AsMessage;
 
 #[AsMessage('async')]
-final readonly class PushMetricMessage
+final readonly class PushMetricMessage implements MonitoringMessageInterface
 {
     public function __construct(
         public string $key,

src/Subscriber/MessengerMonitoringSubscriber.php

@@ -4,6 +4,7 @@
 
 namespace BytesCommerce\ZabbixApi\Subscriber;
 
+use BytesCommerce\ZabbixApi\Contract\MonitoringMessageInterface;
 use BytesCommerce\ZabbixApi\Contract\ZabbixNamingProviderInterface;
 use BytesCommerce\ZabbixApi\Message\PushMetricMessage;
 use Symfony\Component\DependencyInjection\Attribute\Autowire;
@@ -34,6 +35,10 @@ public static function getSubscribedEvents(): array
 
     public function onMessageReceived(WorkerMessageReceivedEvent $event): void
     {
+        if ($this->isMonitoringMessage($event)) {
+            return;
+        }
+
         $messageClass = $this->getShortClassName($event->getEnvelope()->getMessage()::class);
 
         $this->bus->dispatch(new PushMetricMessage(
@@ -49,6 +54,10 @@ public function onMessageReceived(WorkerMessageReceivedEvent $event): void
 
     public function onMessageHandled(WorkerMessageHandledEvent $event): void
     {
+        if ($this->isMonitoringMessage($event)) {
+            return;
+        }
+
         $messageClass = $this->getShortClassName($event->getEnvelope()->getMessage()::class);
 
         $this->bus->dispatch(new PushMetricMessage(
@@ -63,6 +72,10 @@ public function onMessageHandled(WorkerMessageHandledEvent $event): void
 
     public function onMessageFailed(WorkerMessageFailedEvent $event): void
     {
+        if ($this->isMonitoringMessage($event)) {
+            return;
+        }
+
         $messageClass = $this->getShortClassName($event->getEnvelope()->getMessage()::class);
         $errorClass = $this->getShortClassName($event->getThrowable()::class);
 
@@ -77,6 +90,11 @@ public function onMessageFailed(WorkerMessageFailedEvent $event): void
         ));
     }
 
+    private function isMonitoringMessage(WorkerMessageReceivedEvent|WorkerMessageHandledEvent|WorkerMessageFailedEvent $event): bool
+    {
+        return $event->getEnvelope()->getMessage() instanceof MonitoringMessageInterface;
+    }
+
     private function getShortClassName(string $fqcn): string
     {
         $parts = explode('\\', $fqcn);

tests/Subscriber/MessengerMonitoringSubscriberTest.php

@@ -0,0 +1,143 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BytesCommerce\ZabbixApi\Tests\Subscriber;
+
+use BytesCommerce\ZabbixApi\Contract\MonitoringMessageInterface;
+use BytesCommerce\ZabbixApi\Contract\ZabbixNamingProviderInterface;
+use BytesCommerce\ZabbixApi\Message\PushEventMessage;
+use BytesCommerce\ZabbixApi\Message\PushMetricMessage;
+use BytesCommerce\ZabbixApi\Subscriber\MessengerMonitoringSubscriber;
+use PHPUnit\Framework\MockObject\MockObject;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
+use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+
+final class MessengerMonitoringSubscriberTest extends TestCase
+{
+    private MessageBusInterface&MockObject $bus;
+
+    private ZabbixNamingProviderInterface&MockObject $naming;
+
+    private MessengerMonitoringSubscriber $subscriber;
+
+    protected function setUp(): void
+    {
+        $this->bus = $this->createMock(MessageBusInterface::class);
+        $this->naming = $this->createMock(ZabbixNamingProviderInterface::class);
+        $this->naming->method('getItemKey')->willReturnCallback(static fn (string $key) => 'app.' . $key);
+
+        $this->subscriber = new MessengerMonitoringSubscriber(
+            $this->bus,
+            $this->naming,
+            'test',
+        );
+    }
+
+    public function testPushMetricMessageImplementsMonitoringMessageInterface(): void
+    {
+        $message = new PushMetricMessage('test.key', 1);
+        $this->assertInstanceOf(MonitoringMessageInterface::class, $message);
+    }
+
+    public function testPushEventMessageImplementsMonitoringMessageInterface(): void
+    {
+        $message = new PushEventMessage('test.key', ['foo' => 'bar']);
+        $this->assertInstanceOf(MonitoringMessageInterface::class, $message);
+    }
+
+    public function testOnMessageReceivedDoesNotDispatchForPushMetricMessage(): void
+    {
+        $monitoringMessage = new PushMetricMessage('test.key', 42);
+        $envelope = new Envelope($monitoringMessage, [new ReceivedStamp('async')]);
+        $event = new WorkerMessageReceivedEvent($envelope, 'async');
+
+        $this->bus->expects($this->never())->method('dispatch');
+
+        $this->subscriber->onMessageReceived($event);
+    }
+
+    public function testOnMessageHandledDoesNotDispatchForPushMetricMessage(): void
+    {
+        $monitoringMessage = new PushMetricMessage('test.key', 42);
+        $envelope = new Envelope($monitoringMessage);
+        $event = new WorkerMessageHandledEvent($envelope, 'async');
+
+        $this->bus->expects($this->never())->method('dispatch');
+
+        $this->subscriber->onMessageHandled($event);
+    }
+
+    public function testOnMessageFailedDoesNotDispatchForPushMetricMessage(): void
+    {
+        $monitoringMessage = new PushMetricMessage('test.key', 42);
+        $envelope = new Envelope($monitoringMessage);
+        $event = new WorkerMessageFailedEvent($envelope, 'async', new \RuntimeException('Test error'));
+
+        $this->bus->expects($this->never())->method('dispatch');
+
+        $this->subscriber->onMessageFailed($event);
+    }
+
+    public function testOnMessageReceivedDoesNotDispatchForPushEventMessage(): void
+    {
+        $monitoringMessage = new PushEventMessage('test.key', ['data' => 'value']);
+        $envelope = new Envelope($monitoringMessage, [new ReceivedStamp('async')]);
+        $event = new WorkerMessageReceivedEvent($envelope, 'async');
+
+        $this->bus->expects($this->never())->method('dispatch');
+
+        $this->subscriber->onMessageReceived($event);
+    }
+
+    public function testOnMessageHandledDoesNotDispatchForPushEventMessage(): void
+    {
+        $monitoringMessage = new PushEventMessage('test.key', ['data' => 'value']);
+        $envelope = new Envelope($monitoringMessage);
+        $event = new WorkerMessageHandledEvent($envelope, 'async');
+
+        $this->bus->expects($this->never())->method('dispatch');
+
+        $this->subscriber->onMessageHandled($event);
+    }
+
+    public function testOnMessageFailedDoesNotDispatchForPushEventMessage(): void
+    {
+        $monitoringMessage = new PushEventMessage('test.key', ['data' => 'value']);
+        $envelope = new Envelope($monitoringMessage);
+        $event = new WorkerMessageFailedEvent($envelope, 'async', new \RuntimeException('Test error'));
+
+        $this->bus->expects($this->never())->method('dispatch');
+
+        $this->subscriber->onMessageFailed($event);
+    }
+
+    public function testNoRecursiveMessageLoopOccursForPushMetricMessage(): void
+    {
+        $monitoringMessage = new PushMetricMessage('test.metric', 100);
+        $envelope = new Envelope($monitoringMessage);
+
+        $this->bus->expects($this->never())->method('dispatch');
+
+        $this->subscriber->onMessageReceived(new WorkerMessageReceivedEvent($envelope, 'async'));
+        $this->subscriber->onMessageHandled(new WorkerMessageHandledEvent($envelope, 'async'));
+        $this->subscriber->onMessageFailed(new WorkerMessageFailedEvent($envelope, 'async', new \RuntimeException()));
+    }
+
+    public function testNoRecursiveMessageLoopOccursForPushEventMessage(): void
+    {
+        $monitoringMessage = new PushEventMessage('test.event', ['data' => 'value']);
+        $envelope = new Envelope($monitoringMessage);
+
+        $this->bus->expects($this->never())->method('dispatch');
+
+        $this->subscriber->onMessageReceived(new WorkerMessageReceivedEvent($envelope, 'async'));
+        $this->subscriber->onMessageHandled(new WorkerMessageHandledEvent($envelope, 'async'));
+        $this->subscriber->onMessageFailed(new WorkerMessageFailedEvent($envelope, 'async', new \RuntimeException()));
+    }
+}