lockable middleware

Messenger/CommandMessage.php

@@ -96,4 +96,18 @@ abstract class CommandMessage
 
         return $options;
     }
+
+    /**
+     * For lockable message
+     *
+     * @return array
+     */
+    public function __serialize(): array
+    {
+        return [
+            'command' => $this->getCommandName(),
+            'arguments' => $this->getArguments(),
+            'options' => $this->getOptions()
+        ];
+    }
 }

Messenger/Middleware/LockableMessage.php

@@ -0,0 +1,13 @@
+<?php
+
+namespace RetailCrm\ServiceBundle\Messenger\Middleware;
+
+/**
+ * Interface LockableMessage
+ *
+ * @package RetailCrm\ServiceBundle\Messenger\Middleware
+ */
+interface LockableMessage
+{
+    public function __serialize(): array;
+}

Messenger/Middleware/LockableMessageMiddleware.php

@@ -0,0 +1,58 @@
+<?php
+
+namespace RetailCrm\ServiceBundle\Messenger\Middleware;
+
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackInterface;
+use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Throwable;
+
+/**
+ * Class LockableMessageMiddleware
+ *
+ * @package RetailCrm\ServiceBundle\Messenger\Middleware
+ */
+class LockableMessageMiddleware implements MiddlewareInterface
+{
+    /**
+     * @var LockFactory
+     */
+    private $lockFactory;
+
+    public function __construct(LockFactory $lockFactory)
+    {
+        $this->lockFactory = $lockFactory;
+    }
+
+    /**
+     * @param Envelope $envelope
+     * @param StackInterface $stack
+     *
+     * @return Envelope
+     *
+     * @throws Throwable
+     */
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        $message = $envelope->getMessage();
+
+        if ($envelope->all(ReceivedStamp::class) && $message instanceof LockableMessage) {
+            $lock = $this->lockFactory->createLock(md5(serialize($message)));
+            if (!$lock->acquire()) {
+                return $envelope;
+            }
+
+            try {
+                return $stack->next()->handle($envelope, $stack);
+            } catch (Throwable $exception) {
+                throw $exception;
+            } finally {
+                $lock->release();
+            }
+        }
+
+        return $stack->next()->handle($envelope, $stack);
+    }
+}

composer.json

@@ -20,7 +20,8 @@
         "symfony/console": "^5.2",
         "symfony/messenger": "^5.2",
         "symfony/process": "^5.2",
-        "symfony/event-dispatcher": "^5.2"
+        "symfony/event-dispatcher": "^5.2",
+        "symfony/lock": "^5.2"
     },
     "autoload": {
         "psr-4": {