Adding ConsumedByWorkerStamp as way to mark a message in a "worker context"

Author: weaverryan

Messenger/DoctrineClearEntityManagerMiddleware.php

@@ -14,7 +14,7 @@
 use Doctrine\ORM\EntityManagerInterface;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Middleware\StackInterface;
-use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
 
 /**
  * Clears entity manager after calling all handlers.
@@ -28,7 +28,7 @@ protected function handleForManager(EntityManagerInterface $entityManager, Envel
         try {
             return $stack->next()->handle($envelope, $stack);
         } finally {
-            if (null !== $envelope->last(ReceivedStamp::class)) {
+            if (null !== $envelope->last(ConsumedByWorkerStamp::class)) {
                 $entityManager->clear();
             }
         }

Messenger/DoctrineCloseConnectionMiddleware.php

@@ -14,7 +14,7 @@
 use Doctrine\ORM\EntityManagerInterface;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Middleware\StackInterface;
-use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
 
 /**
  * Closes connection and therefore saves number of connections.
@@ -30,7 +30,7 @@ protected function handleForManager(EntityManagerInterface $entityManager, Envel
 
             return $stack->next()->handle($envelope, $stack);
         } finally {
-            if (null !== $envelope->last(ReceivedStamp::class)) {
+            if (null !== $envelope->last(ConsumedByWorkerStamp::class)) {
                 $connection->close();
             }
         }

Messenger/DoctrinePingConnectionMiddleware.php

@@ -14,7 +14,7 @@
 use Doctrine\ORM\EntityManagerInterface;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Middleware\StackInterface;
-use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
 
 /**
  * Checks whether the connection is still open or reconnects otherwise.
@@ -25,7 +25,7 @@ class DoctrinePingConnectionMiddleware extends AbstractDoctrineMiddleware
 {
     protected function handleForManager(EntityManagerInterface $entityManager, Envelope $envelope, StackInterface $stack): Envelope
     {
-        if (null !== $envelope->last(ReceivedStamp::class)) {
+        if (null !== $envelope->last(ConsumedByWorkerStamp::class)) {
             $this->pingConnection($entityManager);
         }
 

Tests/Messenger/DoctrineClearEntityManagerMiddlewareTest.php

@@ -16,7 +16,7 @@
 use Symfony\Bridge\Doctrine\Messenger\DoctrineClearEntityManagerMiddleware;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
-use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
 use Symfony\Component\Messenger\Test\Middleware\MiddlewareTestCase;
 
 class DoctrineClearEntityManagerMiddlewareTest extends MiddlewareTestCase
@@ -36,7 +36,7 @@ public function testMiddlewareClearEntityManager()
         $middleware = new DoctrineClearEntityManagerMiddleware($managerRegistry, 'default');
 
         $envelope = new Envelope(new \stdClass(), [
-            new ReceivedStamp('async'),
+            new ConsumedByWorkerStamp(),
         ]);
         $middleware->handle($envelope, $this->getStackMock());
     }

Tests/Messenger/DoctrineCloseConnectionMiddlewareTest.php

@@ -17,7 +17,7 @@
 use Symfony\Bridge\Doctrine\Messenger\DoctrineCloseConnectionMiddleware;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
-use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
 use Symfony\Component\Messenger\Test\Middleware\MiddlewareTestCase;
 
 class DoctrineCloseConnectionMiddlewareTest extends MiddlewareTestCase
@@ -51,7 +51,7 @@ public function testMiddlewareCloseConnection()
         ;
 
         $envelope = new Envelope(new \stdClass(), [
-            new ReceivedStamp('async'),
+            new ConsumedByWorkerStamp(),
         ]);
         $this->middleware->handle($envelope, $this->getStackMock());
     }

Tests/Messenger/DoctrinePingConnectionMiddlewareTest.php

@@ -17,7 +17,7 @@
 use Symfony\Bridge\Doctrine\Messenger\DoctrinePingConnectionMiddleware;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
-use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
 use Symfony\Component\Messenger\Test\Middleware\MiddlewareTestCase;
 
 class DoctrinePingConnectionMiddlewareTest extends MiddlewareTestCase
@@ -58,7 +58,7 @@ public function testMiddlewarePingOk()
         ;
 
         $envelope = new Envelope(new \stdClass(), [
-            new ReceivedStamp('async'),
+            new ConsumedByWorkerStamp(),
         ]);
         $this->middleware->handle($envelope, $this->getStackMock());
     }
@@ -75,7 +75,7 @@ public function testMiddlewarePingResetEntityManager()
         ;
 
         $envelope = new Envelope(new \stdClass(), [
-            new ReceivedStamp('async'),
+            new ConsumedByWorkerStamp(),
         ]);
         $this->middleware->handle($envelope, $this->getStackMock());
     }

composer.json

@@ -29,7 +29,7 @@
         "symfony/dependency-injection": "^3.4|^4.0|^5.0",
         "symfony/form": "^4.4|^5.0",
         "symfony/http-kernel": "^3.4|^4.0|^5.0",
-        "symfony/messenger": "^4.3|^5.0",
+        "symfony/messenger": "^4.4|^5.0",
         "symfony/property-access": "^3.4|^4.0|^5.0",
         "symfony/property-info": "^3.4|^4.0|^5.0",
         "symfony/proxy-manager-bridge": "^3.4|^4.0|^5.0",