symfony · kbond · Mar 29, 2023
@@ -13,6 +13,7 @@
 
 use Psr\Clock\ClockInterface;
 use Symfony\Component\Clock\Clock;
+use Symfony\Component\Scheduler\RecurringMessage;
 use Symfony\Component\Scheduler\Schedule;
 use Symfony\Component\Scheduler\Trigger\TriggerInterface;
 
@@ -37,6 +38,9 @@ public function __construct(
         $this->checkpoint = $checkpoint;
     }
 
+    /**
+     * @return \Generator<RecurringMessage>
+     */
     public function getMessages(): \Generator
     {
         if (!$this->waitUntil
@@ -54,8 +58,8 @@ public function getMessages(): \Generator
             /** @var TriggerInterface $trigger */
             /** @var int $index */
             /** @var \DateTimeImmutable $time */
-            /** @var object $message */
-            [$time, $index, $trigger, $message] = $heap->extract();
+            /** @var object $recurringMessage */
+            [$time, $index, $trigger, $recurringMessage] = $heap->extract();
             $yield = true;
 
             if ($time < $lastTime) {
@@ -66,11 +70,11 @@ public function getMessages(): \Generator
             }
 
             if ($nextTime = $trigger->getNextRunDate($time)) {
-                $heap->insert([$nextTime, $index, $trigger, $message]);
+                $heap->insert([$nextTime, $index, $trigger, $recurringMessage]);
             }
 
             if ($yield) {
-                yield $message;
+                yield $recurringMessage;
                 $this->checkpoint->save($time, $index);
             }
         }
@@ -93,7 +97,7 @@ private function heap(\DateTimeImmutable $time): TriggerHeap
                 continue;
             }
 
-            $heap->insert([$nextTime, $index, $recurringMessage->getTrigger(), $recurringMessage->getMessage()]);
+            $heap->insert([$nextTime, $index, $recurringMessage->getTrigger(), $recurringMessage]);
         }
 
         return $this->triggerHeap = $heap;

@@ -11,13 +11,15 @@
 
 namespace Symfony\Component\Scheduler\Generator;
 
+use Symfony\Component\Scheduler\RecurringMessage;
+
 /**
  * @experimental
  */
 interface MessageGeneratorInterface
 {
     /**
-     * @return iterable<object>
+     * @return iterable<object|RecurringMessage>
      */
     public function getMessages(): iterable;
 }
@@ -12,10 +12,14 @@
 namespace Symfony\Component\Scheduler\Messenger;
 
 use Symfony\Component\Messenger\Stamp\NonSendableStampInterface;
+use Symfony\Component\Scheduler\RecurringMessage;
 
 /**
  * @experimental
  */
 final class ScheduledStamp implements NonSendableStampInterface
 {
+    public function __construct(public readonly RecurringMessage $recurringMessage)
+    {
+    }
 }
@@ -15,6 +15,7 @@
 use Symfony\Component\Messenger\Transport\TransportInterface;
 use Symfony\Component\Scheduler\Exception\LogicException;
 use Symfony\Component\Scheduler\Generator\MessageGeneratorInterface;
+use Symfony\Component\Scheduler\RecurringMessage;
 
 /**
  * @experimental
@@ -29,7 +30,11 @@ public function __construct(
     public function get(): iterable
     {
         foreach ($this->messageGenerator->getMessages() as $message) {
-            yield Envelope::wrap($message, [new ScheduledStamp()]);
+            if (!$message instanceof RecurringMessage) {
+                throw new LogicException(sprintf('Messages from "%s" must be instances of "%s". Got "%s".', __CLASS__, RecurringMessage::class, get_debug_type($message)));
+            }
+
+            yield Envelope::wrap($message->getMessage(), [new ScheduledStamp($message)]);
         }
     }
 

@@ -66,6 +66,10 @@ public function run(array $options = []): void
             $ran = false;
             foreach ($this->generators as $generator) {
                 foreach ($generator->getMessages() as $message) {
+                    if ($message instanceof RecurringMessage) {
+                        $message = $message->getMessage();
+                    }
+
                     $this->handlers[$message::class]($message);
                     $ran = true;
                 }

@@ -47,7 +47,8 @@ public function testGetMessages(string $startTime, array $runs, array $schedule)
 
         foreach ($runs as $time => $expected) {
             $now = self::makeDateTime($time);
-            $this->assertSame($expected, iterator_to_array($scheduler->getMessages()));
+            $messages = array_map(fn (RecurringMessage $m) => $m->getMessage(), iterator_to_array($scheduler->getMessages()));
+            $this->assertSame($expected, $messages);
         }
     }
 

@@ -17,24 +17,28 @@
 use Symfony\Component\Scheduler\Generator\MessageGeneratorInterface;
 use Symfony\Component\Scheduler\Messenger\ScheduledStamp;
 use Symfony\Component\Scheduler\Messenger\SchedulerTransport;
+use Symfony\Component\Scheduler\RecurringMessage;
 
 class SchedulerTransportTest extends TestCase
 {
     public function testGetFromIterator()
     {
         $messages = [
-            (object) ['id' => 'first'],
-            (object) ['id' => 'second'],
+            RecurringMessage::cron('* * * * *', (object) ['id' => 'first']),
+            RecurringMessage::cron('* * * * *', (object) ['id' => 'second']),
         ];
         $generator = $this->createConfiguredMock(MessageGeneratorInterface::class, [
             'getMessages' => $messages,
         ]);
         $transport = new SchedulerTransport($generator);
 
         foreach ($transport->get() as $envelope) {
+            $message = array_shift($messages);
+
             $this->assertInstanceOf(Envelope::class, $envelope);
-            $this->assertNotNull($envelope->last(ScheduledStamp::class));
-            $this->assertSame(array_shift($messages), $envelope->getMessage());
+            $this->assertEquals($message, $envelope->last(ScheduledStamp::class)?->recurringMessage);
+            $this->assertSame($message->getMessage(), $envelope->getMessage());
+            $this->assertSame($message->getMessage(), $envelope->getMessage());
         }
 
         $this->assertEmpty($messages);