symfony · fabpot · May 10, 2023 · Mar 27, 2023
@@ -22,6 +22,7 @@ final class MessageContext
 {
     public function __construct(
         public readonly string $name,
+        public readonly string $id,
         public readonly TriggerInterface $trigger,
         public readonly \DateTimeImmutable $triggeredAt,
         public readonly ?\DateTimeImmutable $nextTriggerAt = null,

@@ -13,8 +13,8 @@
 
 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;
 
 /**
  * @experimental
@@ -49,11 +49,13 @@ public function getMessages(): \Generator
         $heap = $this->heap($lastTime);
 
         while (!$heap->isEmpty() && $heap->top()[0] <= $now) {
-            /** @var TriggerInterface $trigger */
-            /** @var int $index */
             /** @var \DateTimeImmutable $time */
-            /** @var object $message */
-            [$time, $index, $trigger, $message] = $heap->extract();
+            /** @var int $index */
+            /** @var RecurringMessage $recurringMessage */
+            [$time, $index, $recurringMessage] = $heap->extract();
+            $id = $recurringMessage->getId();
+            $message = $recurringMessage->getMessage();
+            $trigger = $recurringMessage->getTrigger();
             $yield = true;
 
             if ($time < $lastTime) {
@@ -64,11 +66,11 @@ public function getMessages(): \Generator
             }
 
             if ($nextTime = $trigger->getNextRunDate($time)) {
-                $heap->insert([$nextTime, $index, $trigger, $message]);
+                $heap->insert([$nextTime, $index, $recurringMessage]);
             }
 
             if ($yield) {
-                yield (new MessageContext($this->name, $trigger, $time, $nextTime)) => $message;
+                yield (new MessageContext($this->name, $id, $trigger, $time, $nextTime)) => $message;
                 $this->checkpoint->save($time, $index);
             }
         }
@@ -91,7 +93,7 @@ private function heap(\DateTimeImmutable $time): TriggerHeap
                 continue;
             }
 
-            $heap->insert([$nextTime, $index, $recurringMessage->getTrigger(), $recurringMessage->getMessage()]);
+            $heap->insert([$nextTime, $index, $recurringMessage]);
         }
 
         return $this->triggerHeap = $heap;

@@ -11,12 +11,12 @@
 
 namespace Symfony\Component\Scheduler\Generator;
 
-use Symfony\Component\Scheduler\Trigger\TriggerInterface;
+use Symfony\Component\Scheduler\RecurringMessage;
 
 /**
  * @internal
  *
- * @extends \SplHeap<array{\DateTimeImmutable, int, TriggerInterface, object}>
+ * @extends \SplHeap<array{\DateTimeImmutable, int, RecurringMessage}>
  *
  * @experimental
  */
@@ -28,8 +28,8 @@ public function __construct(
     }
 
     /**
-     * @param array{\DateTimeImmutable, int, TriggerInterface, object} $value1
-     * @param array{\DateTimeImmutable, int, TriggerInterface, object} $value2
+     * @param array{\DateTimeImmutable, int, RecurringMessage} $value1
+     * @param array{\DateTimeImmutable, int, RecurringMessage} $value2
      */
     protected function compare(mixed $value1, mixed $value2): int
     {

@@ -22,6 +22,8 @@
  */
 final class RecurringMessage
 {
+    private string $id;
+
     private function __construct(
         private readonly TriggerInterface $trigger,
         private readonly object $message,
@@ -65,6 +67,29 @@ public function withJitter(int $maxSeconds = 60): self
         return new self(new JitterTrigger($this->trigger, $maxSeconds), $this->message);
     }
 
+    /**
+     * Unique identifier for this message's context.
+     */
+    public function getId(): string
+    {
+        if (isset($this->id)) {
+            return $this->id;
+        }
+
+        try {
+            $message = $this->message instanceof \Stringable ? (string) $this->message : serialize($this->message);
+        } catch (\Exception) {
+            $message = '';
+        }
+
+        return $this->id = hash('crc32c', implode('', [
+            $this->message::class,
+            $message,
+            $this->trigger::class,
+            (string) $this->trigger,
+        ]));
+    }
+
     public function getMessage(): object
     {
         return $this->message;

@@ -12,14 +12,15 @@
 namespace Symfony\Component\Scheduler;
 
 use Symfony\Component\Lock\LockInterface;
+use Symfony\Component\Scheduler\Exception\LogicException;
 use Symfony\Contracts\Cache\CacheInterface;
 
 /**
  * @experimental
  */
 final class Schedule implements ScheduleProviderInterface
 {
-    /** @var array<RecurringMessage> */
+    /** @var array<string,RecurringMessage> */
     private array $messages = [];
     private ?LockInterface $lock = null;
     private ?CacheInterface $state = null;
@@ -29,8 +30,13 @@ final class Schedule implements ScheduleProviderInterface
      */
     public function add(RecurringMessage $message, RecurringMessage ...$messages): static
     {
-        $this->messages[] = $message;
-        $this->messages = array_merge($this->messages, $messages);
+        foreach ([$message, ...$messages] as $m) {
+            if (isset($this->messages[$m->getId()])) {
+                throw new LogicException('Duplicated schedule message.');
+            }
+
+            $this->messages[$m->getId()] = $m;
+        }
 
         return $this;
     }
@@ -70,7 +76,7 @@ public function getState(): ?CacheInterface
      */
     public function getRecurringMessages(): array
     {
-        return $this->messages;
+        return array_values($this->messages);
     }
 
     /**

@@ -34,16 +34,17 @@ public function testGetFromIterator()
         $generator->method('getMessages')->willReturnCallback(function () use ($messages): \Generator {
             $trigger = $this->createMock(TriggerInterface::class);
             $triggerAt = new \DateTimeImmutable('2020-02-20T02:00:00', new \DateTimeZone('UTC'));
-            yield (new MessageContext('default', $trigger, $triggerAt)) => $messages[0];
-            yield (new MessageContext('default', $trigger, $triggerAt)) => $messages[1];
+            yield (new MessageContext('default', 'id1', $trigger, $triggerAt)) => $messages[0];
+            yield (new MessageContext('default', 'id2', $trigger, $triggerAt)) => $messages[1];
         });
         $transport = new SchedulerTransport($generator);
 
-        foreach ($transport->get() as $envelope) {
+        foreach ($transport->get() as $i => $envelope) {
             $this->assertInstanceOf(Envelope::class, $envelope);
             $this->assertNotNull($stamp = $envelope->last(ScheduledStamp::class));
             $this->assertSame(array_shift($messages), $envelope->getMessage());
             $this->assertSame('default', $stamp->messageContext->name);
+            $this->assertSame('id'.$i + 1, $stamp->messageContext->id);
         }
 
         $this->assertEmpty($messages);

@@ -42,4 +42,13 @@ public function testHashedCronContextIsRequiredIfMessageIsNotStringable()
 
         RecurringMessage::cron('#midnight', new \stdClass());
     }
+
+    public function testUniqueId()
+    {
+        $message1 = RecurringMessage::cron('* * * * *', new \stdClass());
+        $message2 = RecurringMessage::cron('* 5 * * *', new \stdClass());
+
+        $this->assertSame($message1->getId(), (clone $message1)->getId());
+        $this->assertNotSame($message1->getId(), $message2->getId());
+    }
 }
@@ -0,0 +1,30 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Scheduler\Exception\LogicException;
+use Symfony\Component\Scheduler\RecurringMessage;
+use Symfony\Component\Scheduler\Schedule;
+
+class ScheduleTest extends TestCase
+{
+    public function testCannotAddDuplicateMessage()
+    {
+        $schedule = new Schedule();
+        $schedule->add(RecurringMessage::cron('* * * * *', new \stdClass()));
+
+        $this->expectException(LogicException::class);
+
+        $schedule->add(RecurringMessage::cron('* * * * *', new \stdClass()));
+    }
+}