feature #51805 [Scheduler] pre_run and post_run events (alli83)

This PR was merged into the 6.4 branch.

Discussion
----------

[Scheduler] pre_run and post_run events

| Q             | A
| ------------- | ---
| Branch?       | 6.4
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| Tickets       | #49803 (comment)
| License       | MIT

Based on #49803 `@kbond`  and taking into account #51553

The aim of this PR is to be able to act on the accumulated messages 'if something happens' and to be able to recalculate the heap via events (currently pre_run and post_run).
The aim is to have access to
- the  the schedule and therefore add/cancel a certain type of message.
- MessageContexte (e.g. access the id)
- The message itself

This PR would complement `@Jeroeny` #51553 PR by enabling action via events.

Commits
-------

20fd21a [Scheduler] add PRE_RUN and POST_RUN events

src/Symfony/Bundle/FrameworkBundle/Resources/config/scheduler.php

@@ -11,6 +11,7 @@
 
 namespace Symfony\Component\DependencyInjection\Loader\Configurator;
 
+use Symfony\Component\Scheduler\EventListener\DispatchSchedulerEventListener;
 use Symfony\Component\Scheduler\Messenger\SchedulerTransportFactory;
 use Symfony\Component\Scheduler\Messenger\ServiceCallMessageHandler;
 
@@ -27,5 +28,11 @@
                 service('clock'),
             ])
             ->tag('messenger.transport_factory')
+        ->set('scheduler.event_listener', DispatchSchedulerEventListener::class)
+            ->args([
+                tagged_locator('scheduler.schedule_provider', 'name'),
+                service('event_dispatcher'),
+            ])
+            ->tag('kernel.event_subscriber')
     ;
 };

src/Symfony/Component/Scheduler/CHANGELOG.md

@@ -13,6 +13,8 @@ CHANGELOG
  * Add `ScheduledStamp` to `RedispatchMessage`
  * Allow modifying Schedule instances at runtime
  * Add `MessageProviderInterface` to trigger unique messages at runtime
+ * Add `PreRunEvent` and `PostRunEvent` events
+ * Add `DispatchSchedulerEventListener`
 
 6.3
 ---

src/Symfony/Component/Scheduler/DependencyInjection/AddScheduleMessengerPass.php

@@ -30,6 +30,10 @@ class AddScheduleMessengerPass implements CompilerPassInterface
 {
     public function process(ContainerBuilder $container): void
     {
+        if (!$container->has('event_dispatcher')) {
+            $container->removeDefinition('scheduler.event_listener');
+        }
+
         $receivers = [];
         foreach ($container->findTaggedServiceIds('messenger.receiver') as $tags) {
             $receivers[$tags[0]['alias']] = true;

src/Symfony/Component/Scheduler/Event/PostRunEvent.php

@@ -0,0 +1,40 @@
+<?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\Event;
+
+use Symfony\Component\Scheduler\Generator\MessageContext;
+use Symfony\Component\Scheduler\ScheduleProviderInterface;
+
+class PostRunEvent
+{
+    public function __construct(
+        private readonly ScheduleProviderInterface $schedule,
+        private readonly MessageContext $messageContext,
+        private readonly object $message,
+    ) {
+    }
+
+    public function getMessageContext(): MessageContext
+    {
+        return $this->messageContext;
+    }
+
+    public function getSchedule(): ScheduleProviderInterface
+    {
+        return $this->schedule;
+    }
+
+    public function getMessage(): object
+    {
+        return $this->message;
+    }
+}

src/Symfony/Component/Scheduler/Event/PreRunEvent.php

@@ -0,0 +1,51 @@
+<?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\Event;
+
+use Symfony\Component\Scheduler\Generator\MessageContext;
+use Symfony\Component\Scheduler\ScheduleProviderInterface;
+
+class PreRunEvent
+{
+    private bool $shouldCancel = false;
+
+    public function __construct(
+        private readonly ScheduleProviderInterface $schedule,
+        private readonly MessageContext $messageContext,
+        private readonly object $message,
+    ) {
+    }
+
+    public function getMessageContext(): MessageContext
+    {
+        return $this->messageContext;
+    }
+
+    public function getSchedule(): ScheduleProviderInterface
+    {
+        return $this->schedule;
+    }
+
+    public function getMessage(): object
+    {
+        return $this->message;
+    }
+
+    public function shouldCancel(bool $shouldCancel = null): bool
+    {
+        if (null !== $shouldCancel) {
+            $this->shouldCancel = $shouldCancel;
+        }
+
+        return $this->shouldCancel;
+    }
+}

src/Symfony/Component/Scheduler/EventListener/DispatchSchedulerEventListener.php

@@ -0,0 +1,73 @@
+<?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\EventListener;
+
+use Psr\Container\ContainerInterface;
+use Psr\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
+use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
+use Symfony\Component\Scheduler\Event\PostRunEvent;
+use Symfony\Component\Scheduler\Event\PreRunEvent;
+use Symfony\Component\Scheduler\Messenger\ScheduledStamp;
+
+class DispatchSchedulerEventListener implements EventSubscriberInterface
+{
+    public function __construct(
+        private readonly ContainerInterface $scheduleProviderLocator,
+        private readonly EventDispatcherInterface $eventDispatcher,
+    ) {
+    }
+
+    public function onMessageHandled(WorkerMessageHandledEvent $event): void
+    {
+        $envelope = $event->getEnvelope();
+        if (!$scheduledStamp = $envelope->last(ScheduledStamp::class)) {
+            return;
+        }
+
+        if (!$this->scheduleProviderLocator->has($scheduledStamp->messageContext->name)) {
+            return;
+        }
+
+        $this->eventDispatcher->dispatch(new PostRunEvent($this->scheduleProviderLocator->get($scheduledStamp->messageContext->name), $scheduledStamp->messageContext, $envelope->getMessage()));
+    }
+
+    public function onMessageReceived(WorkerMessageReceivedEvent $event): void
+    {
+        $envelope = $event->getEnvelope();
+
+        if (!$scheduledStamp = $envelope->last(ScheduledStamp::class)) {
+            return;
+        }
+
+        if (!$this->scheduleProviderLocator->has($scheduledStamp->messageContext->name)) {
+            return;
+        }
+
+        $preRunEvent = new PreRunEvent($this->scheduleProviderLocator->get($scheduledStamp->messageContext->name), $scheduledStamp->messageContext, $envelope->getMessage());
+
+        $this->eventDispatcher->dispatch($preRunEvent);
+
+        if ($preRunEvent->shouldCancel()) {
+            $event->shouldHandle(false);
+        }
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            WorkerMessageReceivedEvent::class => ['onMessageReceived'],
+            WorkerMessageHandledEvent::class => ['onMessageHandled'],
+        ];
+    }
+}

src/Symfony/Component/Scheduler/Generator/MessageGenerator.php

@@ -93,6 +93,11 @@ public function getMessages(): \Generator
         $checkpoint->release($now, $this->waitUntil);
     }
 
+    public function getSchedule(): Schedule
+    {
+        return $this->schedule ??= $this->scheduleProvider->getSchedule();
+    }
+
     private function heap(\DateTimeImmutable $time, \DateTimeImmutable $startTime): TriggerHeap
     {
         if (isset($this->triggerHeap) && $this->triggerHeap->time <= $time) {
@@ -101,7 +106,7 @@ private function heap(\DateTimeImmutable $time, \DateTimeImmutable $startTime):
 
         $heap = new TriggerHeap($time);
 
-        foreach ($this->schedule()->getRecurringMessages() as $index => $recurringMessage) {
+        foreach ($this->getSchedule()->getRecurringMessages() as $index => $recurringMessage) {
             $trigger = $recurringMessage->getTrigger();
 
             if ($trigger instanceof StatefulTriggerInterface) {
@@ -118,13 +123,8 @@ private function heap(\DateTimeImmutable $time, \DateTimeImmutable $startTime):
         return $this->triggerHeap = $heap;
     }
 
-    private function schedule(): Schedule
-    {
-        return $this->schedule ??= $this->scheduleProvider->getSchedule();
-    }
-
     private function checkpoint(): Checkpoint
     {
-        return $this->checkpoint ??= new Checkpoint('scheduler_checkpoint_'.$this->name, $this->schedule()->getLock(), $this->schedule()->getState());
+        return $this->checkpoint ??= new Checkpoint('scheduler_checkpoint_'.$this->name, $this->getSchedule()->getLock(), $this->getSchedule()->getState());
     }
 }

src/Symfony/Component/Scheduler/Messenger/SchedulerTransport.php

@@ -54,4 +54,9 @@ public function send(Envelope $envelope): Envelope
     {
         throw new LogicException(sprintf('"%s" cannot send messages.', __CLASS__));
     }
+
+    public function getMessageGenerator(): MessageGeneratorInterface
+    {
+        return $this->messageGenerator;
+    }
 }

src/Symfony/Component/Scheduler/Schedule.php

@@ -11,21 +11,29 @@
 
 namespace Symfony\Component\Scheduler;
 
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 use Symfony\Component\Lock\LockInterface;
+use Symfony\Component\Scheduler\Event\PostRunEvent;
+use Symfony\Component\Scheduler\Event\PreRunEvent;
 use Symfony\Component\Scheduler\Exception\LogicException;
 use Symfony\Contracts\Cache\CacheInterface;
 
 final class Schedule implements ScheduleProviderInterface
 {
+    public function __construct(
+        private readonly ?EventDispatcherInterface $dispatcher = null,
+    ) {
+    }
+
     /** @var array<string,RecurringMessage> */
     private array $messages = [];
     private ?LockInterface $lock = null;
     private ?CacheInterface $state = null;
     private bool $shouldRestart = false;
 
-    public static function with(RecurringMessage $message, RecurringMessage ...$messages): static
+    public function with(RecurringMessage $message, RecurringMessage ...$messages): static
     {
-        return static::doAdd(new self(), $message, ...$messages);
+        return static::doAdd(new self($this->dispatcher), $message, ...$messages);
     }
 
     /**
@@ -62,6 +70,17 @@ public function remove(RecurringMessage $message): static
         return $this;
     }
 
+    /**
+     * @return $this
+     */
+    public function removeById(string $id): static
+    {
+        unset($this->messages[$id]);
+        $this->setRestart(true);
+
+        return $this;
+    }
+
     /**
      * @return $this
      */
@@ -119,6 +138,20 @@ public function getSchedule(): static
         return $this;
     }
 
+    public function before(callable $listener, int $priority = 0): static
+    {
+        $this->dispatcher->addListener(PreRunEvent::class, $listener, $priority);
+
+        return $this;
+    }
+
+    public function after(callable $listener, int $priority = 0): static
+    {
+        $this->dispatcher->addListener(PostRunEvent::class, $listener, $priority);
+
+        return $this;
+    }
+
     public function shouldRestart(): bool
     {
         return $this->shouldRestart;

src/Symfony/Component/Scheduler/Scheduler.php

@@ -11,8 +11,11 @@
 
 namespace Symfony\Component\Scheduler;
 
+use Psr\EventDispatcher\EventDispatcherInterface;
 use Symfony\Component\Clock\Clock;
 use Symfony\Component\Clock\ClockInterface;
+use Symfony\Component\Scheduler\Event\PostRunEvent;
+use Symfony\Component\Scheduler\Event\PreRunEvent;
 use Symfony\Component\Scheduler\Generator\MessageGenerator;
 
 final class Scheduler
@@ -31,6 +34,7 @@ public function __construct(
         private readonly array $handlers,
         array $schedules,
         private readonly ClockInterface $clock = new Clock(),
+        private readonly ?EventDispatcherInterface $dispatcher = null,
     ) {
         foreach ($schedules as $schedule) {
             $this->addSchedule($schedule);
@@ -62,9 +66,25 @@ public function run(array $options = []): void
 
             $ran = false;
             foreach ($this->generators as $generator) {
-                foreach ($generator->getMessages() as $message) {
+                foreach ($generator->getMessages() as $context => $message) {
+                    if (!$this->dispatcher) {
+                        $this->handlers[$message::class]($message);
+                        $ran = true;
+
+                        continue;
+                    }
+
+                    $preRunEvent = new PreRunEvent($generator->getSchedule(), $context, $message);
+                    $this->dispatcher->dispatch($preRunEvent);
+
+                    if ($preRunEvent->shouldCancel()) {
+                        continue;
+                    }
+
                     $this->handlers[$message::class]($message);
                     $ran = true;
+
+                    $this->dispatcher->dispatch(new PostRunEvent($generator->getSchedule(), $context, $message));
                 }
             }