8000 [Scheduler] add PRE_RUN and POST_RUN events · symfony/symfony@ffd4881 · GitHub
[go: up one dir, main page]
Skip to content

Commit ffd4881

Browse files
alli83alli83
committed
[Scheduler] add PRE_RUN and POST_RUN events
1 parent cdf30c5 commit ffd4881

File tree

11 files changed

+316
-11
lines changed

11 files changed

+316
-11
lines changed

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

+7
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111

1212
namespace Symfony\Component\DependencyInjection\Loader\Configurator;
1313

14+
use Symfony\Component\Scheduler\EventListener\DispatchSchedulerEventListener;
1415
use Symfony\Component\Scheduler\Messenger\SchedulerTransportFactory;
1516
use Symfony\Component\Scheduler\Messenger\ServiceCallMessageHandler;
1617

@@ -27,5 +28,11 @@
2728
service('clock'),
2829
])
2930
->tag('messenger.transport_factory')
31+
->set('scheduler.event_listener', DispatchSchedulerEventListener::class)
32+
->args([
33+
tagged_locator('scheduler.schedule_provider', 'name'),
34+
service('event_dispatcher'),
35+
])
36+
->tag('kernel.event_subscriber')
3037
;
3138
};

‎src/Symfony/Component/Messenger/Tests/EventListener/DispatchSchedulerEventListenerTest.php

+73
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Tests\EventListener;
13+
14+
use PHPUnit\Framework\TestCase;
15+
use Psr\Container\ContainerInterface;
16+
use Symfony\Component\EventDispatcher\EventDispatcher;
17+
use Symfony\Component\Messenger\Envelope;
18+
use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
19+
use Symfony\Component\Scheduler\Event\PreRunEvent;
20+
use Symfony\Component\Scheduler\EventListener\DispatchSchedulerEventListener;
21+
use Symfony\Component\Scheduler\Generator\MessageContext;
22+
use Symfony\Component\Scheduler\Messenger\ScheduledStamp;
23+
use Symfony\Component\Scheduler\RecurringMessage;
24+
use Symfony\Component\Scheduler\Tests\Messenger\SomeScheduleProvider;
25+
use Symfony\Component\Scheduler\Trigger\TriggerInterface;
26+
27+
class DispatchSchedulerEventListenerTest extends TestCase
28+
{
29+
public function testDispatchSchedulerEvents()
30+
{
31+
$trigger = $this->createMock(TriggerInterface::class);
32+
$defaultRecurringMessage = RecurringMessage::trigger($trigger, (object) ['id' => 'default']);
33+
34+
$schedulerProvider = new SomeScheduleProvider([$defaultRecurringMessage]);
35+
$scheduleProviderLocator = $this->createMock(ContainerInterface::class);
36+
$scheduleProviderLocator->expects($this->once())->method('has')->willReturn(true);
37+
$scheduleProviderLocator->expects($this->once())->method('get')->willReturn($schedulerProvider);
38+
39+
$context = new MessageContext('default', 'default', $trigger, $this->createMock(\DateTimeImmutable::class));
40+
$envelope = (new Envelope(new \stdClass()))->with(new ScheduledStamp($context));
41+
42+
/** @var ContainerInterface $scheduleProviderLocator */
43+
$listener = new DispatchSchedulerEventListener($scheduleProviderLocator, $eventDispatcher = new EventDispatcher());
44+
$workerReceivedEvent = new WorkerMessageReceivedEvent($envelope, 'default');
45+
$secondListener = new TestEventListener();
46+
47+
$eventDispatcher->addListener(PreRunEvent::class, [$secondListener, 'preRun']);
48+
$eventDispatcher->addListener(PreRunEvent::class, [$secondListener, 'postRun']);
49+
$listener->onMessageReceived($workerReceivedEvent);
50+
51+
$this->assertTrue($secondListener->preInvoked);
52+
$this->assertTrue($secondListener->postInvoked);
53+
}
54+
}
55+
56+
class TestEventListener
57+
{
58+
public string $name;
59+
public bool $preInvoked = false;
60+
public bool $postInvoked = false;
61+
62+
/* Listener methods */
63+
64+
public function preRun($e)
65+
{
66+
$this->preInvoked = true;
67+
}
68+
69+
public function postRun($e)
70+
{
71+
$this->postInvoked = true;
72+
}
73+
}

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

+4
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,10 @@ class AddScheduleMessengerPass implements CompilerPassInterface
3030
{
3131
public function process(ContainerBuilder $container): void
3232
{
33+
if (!$container->has('event_dispatcher')) {
34+
$container->removeDefinition('scheduler.event_listener');
35+
}
36+
3337
$receivers = [];
3438
foreach ($container->findTaggedServiceIds('messenger.receiver') as $tags) {
3539
$receivers[$tags[0]['alias']] = true;

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

+40
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Scheduler\Event;
13+
14+
use Symfony\Component\Scheduler\Generator\MessageContext;
15+
use Symfony\Component\Scheduler\ScheduleProviderInterface;
16+
17+
class PostRunEvent
18+
{
19+
public function __construct(
20+
private readonly ScheduleProviderInterface $schedule,
21+
private readonly MessageContext $messageContext,
22+
private readonly object $message,
23+
) {
24+
}
25+
26+
public function getMessageContext(): MessageContext
27+
{
28+
return $this->messageContext;
29+
}
30+
31+
public function getSchedule(): ScheduleProviderInterface
32+
{
33+
return $this->schedule;
34+
}
35+
36+
public function getMessage(): object
37+
{
38+
return $this->message;
39+
}
40+
}

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

+51
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Scheduler\Event;
13+
14+
use Symfony\Component\Scheduler\Generator\MessageContext;
15+
use Symfony\Component\Scheduler\ScheduleProviderInterface;
16+
17+
class PreRunEvent
18+
{
19+
private bool $shouldCancel = false;
20+
21+
public function __construct(
22+
private readonly ScheduleProviderInterface $schedule,
23+
private readonly MessageContext $messageContext,
24+
private readonly object $message,
25+
) {
26+
}
27+
28+
public function getMessageContext(): MessageContext
29+
{
30+
return $this->messageContext;
31+
}
32+
33+
public function getSchedule(): ScheduleProviderInterface
34+
{
35+
return $this->schedule;
36+
}
37+
38+
public function getMessage(): object
39+
{
40+
return $this->message;
41+
}
42+
43+
public function shouldCancel(bool $shouldCancel = null): bool
44+
{
45+
if (null !== $shouldCancel) {
46+
$this->shouldCancel = $shouldCancel;
47+
}
48+
49+
return $this->shouldCancel;
50+
}
51+
}

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

+73
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Scheduler\EventListener;
13+
14+
use Psr\Container\ContainerInterface;
15+
use Psr\EventDispatcher\EventDispatcherInterface;
16+
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
17+
use Symfony\Component\Messenger\Event\Worker 10000 MessageHandledEvent;
18+
use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
19+
use Symfony\Component\Scheduler\Event\PostRunEvent;
20+
use Symfony\Component\Scheduler\Event\PreRunEvent;
21+
use Symfony\Component\Scheduler\Messenger\ScheduledStamp;
22+
23+
class DispatchSchedulerEventListener implements EventSubscriberInterface
24+
{
25+
public function __construct(
26+
private readonly ContainerInterface $scheduleProviderLocator,
27+
private readonly EventDispatcherInterface $eventDispatcher,
28+
) {
29+
}
30+
31+
public function onMessageHandled(WorkerMessageHandledEvent $event): void
32+
{
33+
$envelope = $event->getEnvelope();
34+
if (!$scheduledStamp = $envelope->last(ScheduledStamp::class)) {
35+
return;
36+
}
37+
38+
if (!$this->scheduleProviderLocator->has($scheduledStamp->messageContext->name)) {
39+
return;
40+
}
41+
42+
$this->eventDispatcher->dispatch(new PostRunEvent($this->scheduleProviderLocator->get($scheduledStamp->messageContext->name), $scheduledStamp->messageContext, $envelope->getMessage()));
43+
}
44+
45+
public function onMessageReceived(WorkerMessageReceivedEvent $event): void
46+
{
47+
$envelope = $event->getEnvelope();
48+
49+
if (!$scheduledStamp = $envelope->last(ScheduledStamp::class)) {
50+
return;
51+
}
52+
53+
if (!$this->scheduleProviderLocator->has($scheduledStamp->messageContext->name)) {
54+
return;
55+
}
56+
57+
$preRunEvent = new PreRunEvent($this->scheduleProviderLocator->get($scheduledStamp->messageContext->name), $scheduledStamp->messageContext, $envelope->getMessage());
58+
59+
$this->eventDispatcher->dispatch($preRunEvent);
60+
61+
if ($preRunEvent->shouldCancel()) {
62+
$event->shouldHandle(false);
63+
}
64+
}
65+
66+
public static function getSubscribedEvents(): array
67+
{
68+
return [
69+
WorkerMessageReceivedEvent::class => ['onMessageReceived'],
70+
WorkerMessageHandledEvent::class => ['onMessageHandled'],
71+
];
72+
}
73+
}

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

+7-7
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,11 @@ public function getMessages(): \Generator
9393
$checkpoint->release($now, $this->waitUntil);
9494
}
9595

96+
public function getSchedule(): Schedule
97+
{
98+
return $this->schedule ??= $this->scheduleProvider->getSchedule();
99+
}
100+
96101
private function heap(\DateTimeImmutable $time, \DateTimeImmutable $startTime): TriggerHeap
97102
{
98103
if (isset($this->triggerHeap) && $this->triggerHeap->time <= $time) {
@@ -101,7 +106,7 @@ private function heap(\DateTimeImmutable $time, \DateTimeImmutable $startTime):
101106

102107
$heap = new TriggerHeap($time);
103108

104-
foreach ($this->schedule()->getRecurringMessages() as $index => $recurringMessage) {
109+
foreach ($this->getSchedule()->getRecurringMessages() as $index => $recurringMessage) {
105110
$trigger = $recurringMessage->getTrigger();
106111

107112
if ($trigger instanceof StatefulTriggerInterface) {
@@ -118,13 +123,8 @@ private function heap(\DateTimeImmutable $time, \DateTimeImmutable $startTime):
118123
return $this->triggerHeap = $heap;
119124
}
120125

121-
private function schedule(): Schedule
122-
{
123-
return $this->schedule ??= $this->scheduleProvider->getSchedule();
124-
}
125-
126126
private function checkpoint(): Checkpoint
127127
{
128-
return $this->checkpoint ??= new Checkpoint('scheduler_checkpoint_'.$this->name, $this->schedule()->getLock(), $this->schedule()->getState());
128+
return $this->checkpoint ??= new Checkpoint('scheduler_checkpoint_'.$this->name, $this->getSchedule()->getLock(), $this->getSchedule()->getState());
129129
}
130130
}

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

+5
Original file line numberDiff line numberDiff line change
@@ -54,4 +54,9 @@ public function send(Envelope $envelope): Envelope
5454
{
5555
throw new LogicException(sprintf('"%s" cannot send messages.', __CLASS__));
5656
}
57+
58+
public function getMessageGenerator(): MessageGeneratorInterface
59+
{
60+
return $this->messageGenerator;
61+
}
5762
}

‎src/Symfony/Component/Scheduler/Schedule.php

+35-2
Original file line numberDiff line numberDiff line change
@@ -11,21 +11,29 @@
1111

1212
namespace Symfony\Component\Scheduler;
1313

14+
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
1415
use Symfony\Component\Lock\LockInterface;
16+
use Symfony\Component\Scheduler\Event\PostRunEvent;
17+
use Symfony\Component\Scheduler\Event\PreRunEvent;
1518
use Symfony\Component\Scheduler\Exception\LogicException;
1619
use Symfony\Contracts\Cache\CacheInterface;
1720

1821
final class Schedule implements ScheduleProviderInterface
1922
{
23+
public function __construct(
24+
private readonly ?EventDispatcherInterface $dispatcher = null
25+
) {
26+
}
27+
2028
/** @var array<string,RecurringMessage> */
2129
private array $messages = [];
2230
private ?LockInterface $lock = null;
2331
private ?CacheInterface $state = null;
2432
private bool $shouldRestart = false;
2533

26-
public static function with(RecurringMessage $message, RecurringMessage ...$messages): static
34+
public function with(RecurringMessage $message, RecurringMessage ...$messages): static
2735
{
28-
return static::doAdd(new self(), $message, ...$messages);
36+
return static::doAdd(new self($this->dispatcher), $message, ...$messages);
2937
}
3038

3139
/**
@@ -62,6 +70,17 @@ public function remove(RecurringMessage $message): static
6270
return $this;
6371
}
6472

73+
/**
74+
* @return $this
75+
*/
76+
public function removeById(string $id): static
77+
{
78+
unset($this->messages[$id]);
79+
$this->setRestart(true);
80+
81+
return $this;
82+
}
83+
6584
/**
6685
* @return $this
6786
*/
@@ -119,6 +138,20 @@ public function getSchedule(): static
119138
return $this;
120139
}
121140

141+
public function before(callable $listener, int $priority = 0): static
142+
{
143+
$this->dispatcher->addListener(PreRunEvent::class, $listener, $priority);
144+
145+
return $this;
146+
}
147+
148+
public function after(callable $listener, int $priority = 0): static
149+
{
150+
$this->dispatcher->addListener(PostRunEvent::class, $listener, $priority);
151+
152+
return $this;
153+
}
154+
122155
public function shouldRestart(): bool
123156
{
124157
return $this->shouldRestart;

0 commit comments

Comments
 (0)
0