The receivers are returning a \Generator. This allows us to manage the exceptions without specific cases.

Author: sroze

src/Symfony/Component/Message/Asynchronous/Transport/WrapIntoReceivedMessage.php

@@ -25,10 +25,16 @@ public function __construct(ReceiverInterface $decoratedConsumer)
         $this->decoratedReceiver = $decoratedConsumer;
     }
 
-    public function receive(): iterable
+    public function receive(): \Generator
     {
-        foreach ($this->decoratedReceiver->receive() as $message) {
-            yield new ReceivedMessage($message);
+        $generator = $this->decoratedReceiver->receive();
+
+        foreach ($generator as $message) {
+            try {
+                yield new ReceivedMessage($message);
+            } catch (\Throwable $e) {
+                $generator->throw($e);
+            }
         }
     }
 }

src/Symfony/Component/Message/Transport/Enhancers/MaximumCountReceiver.php

@@ -24,12 +24,17 @@ public function __construct(ReceiverInterface $decoratedReceiver, int $maximumNu
         $this->maximumNumberOfMessages = $maximumNumberOfMessages;
     }
 
-    public function receive(): iterable
+    public function receive(): \Generator
     {
+        $generator = $this->decoratedReceiver->receive();
         $receivedMessages = 0;
 
-        foreach ($this->decoratedReceiver->receive() as $message) {
-            yield $message;
+        foreach ($generator as $message) {
+            try {
+                yield $message;
+            } catch (\Throwable $e) {
+                $generator->throw($e);
+            }
 
             if (++$receivedMessages > $this->maximumNumberOfMessages) {
                 break;

src/Symfony/Component/Message/Transport/ReceiverInterface.php

@@ -16,5 +16,5 @@
  */
 interface ReceiverInterface
 {
-    public function receive(): iterable;
+    public function receive(): \Generator;
 }

src/Symfony/Component/Message/Worker.php

@@ -33,12 +33,19 @@ public function __construct(ReceiverInterface $receiver, MessageBusInterface $bu
      */
     public function run()
     {
-        foreach ($this->receiver->receive() as $message) {
+        $messageGenerator = $this->receiver->receive();
+
+        foreach ($messageGenerator as $message) {
             if (!$message instanceof ReceivedMessage) {
                 $message = new ReceivedMessage($message);
             }
 
-            $this->bus->dispatch($message);
+            try {
+                $this->bus->dispatch($message);
+            } catch (\Throwable $e) {
+                // Return the exception to the message generator so it can properly handle it if needed.
+                $messageGenerator->throw($e);
+            }
         }
     }
 }