feature #47112 [Messenger] Add a scheduler component (upyx, fabpot)

This PR was merged into the 6.3 branch.

Discussion
----------

[Messenger] Add a scheduler component

| Q             | A
| ------------- | ---
| Branch?       | 6.3
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| Tickets       | no
| License       | MIT
| Doc PR        | TBD

### Introdution

There is no easy way to schedule periodical tasks. There are few useful tools which I touched:

- https://github.com/Guikingone/SchedulerBundle
- https://github.com/Cron/Symfony-Bundle
- https://github.com/zenstruck/schedule-bundle
- https://github.com/lavary/crunz

They are complicated. They doesn't allow to set time with precision in seconds (at least it isn't easy). They require difficult tricks to set not linear periods (like on sunset in Tokyo). ~They are~ Part of them inefficient with infrequent tasks because resources are needed on every run (e.g. Kubernetes [CronJob](https://kubernetes.io/docs/concepts/workloads/controllers/cron-jobs/) creates and destroes an entire container).

### Proposal

I use a custom transport of Messenger that **generates** messages periodically. It's a simple solution to run periodical jobs without external software. It's especially helpful in environments where additional runners are impossible or very painful (like Kubernetes). Configuration is very flexible and accurate, it's possible to configure any rules when to run or not to run.

Compared to `crond`: there is no need to install something external, precision of schedule in microseconds instead of *minutes*, it's possible to set any periods.

#### Simple example

```yaml
# messenger.yaml
framework:
  messenger:
    transports:
      schedule_default: 'schedule://default'
```

Few types of messages:
```php
class SomeJob {}
class OtherJob {}
```

A handlers:
```php
#[AsMessageHandler]
class SomeJobHandler {
    public function __invoke(SomeJob $job) {
        // do job or delegate it to other service
    }
}

#[AsMessageHandler]
class OtherJobHandler {
    public function __invoke(OtherJob $job) {
        // do job or delegate it to other service
    }
}
```

A schedules are provided by locators. It's possible to create many locators and/or provide many schedules by the same locator:

```php
class ExampleLocator implements ScheduleLocatorInterface
{
    public function get(string $id): ScheduleConfig
    {
        // once after an hour from now
        $deferForAnHour = new \DatePeriod(
            new \DateTimeImmutable('now'),
            new \DateInterval('PT1H'),
            1,
            \DatePeriod::EXCLUDE_START_DATE
        );

        return (new ScheduleConfig())
            ->add(new OnceTrigger(new \DateTimeImmutable()), new WarmUpJob()) // once on every worker's start
            ->add(PeriodicalTrigger::create('P1D', '2022-01-01T03:00:00Z'), new SomeJob()) // every night at 3 a.m. (UTC)
            ->add(PeriodicalTrigger::fromPeriod($deferForAnHour), new OtherJob())
        ;
    }

    public function has(string $id): bool
    {
        return 'default' === $id;
    }
}
```

To run schedule:

```sh
bin/console messenger:consume schedule_default
```

It's easy to run jobs manually:

```php
#[AsCommand(name: 'some', description: 'Manually runs SomeJob')]
class SomeCommand extends Command {
    public function __construct(private MessageBusInterface $bus) {
    }

    protected function execute(InputInterface $input, OutputInterface $output): int {
        $this->bus->dispatch(new SomeJob());
    }
}
```

#### Example with returning a result

```php
class GoodJob {
    public function __construct(public readonly ?LoggerInterface $logger) {
    }
}
```

```php
#[AsMessageHandler]
class GoodHandler {
    public function __construct(private readonly LoggerInterface $logger) {
    }

    public function __invoke(GoodJob $job){
        // compute $result
        ($job->logger ?? $this->logger)->info('The result is: {result}', ['result' => $result])
    }
}
```

```php
#[AsCommand(name: 'job', description: 'Manually runs job')]
class SomeCommand extends Command {
    public function __construct(private MessageBusInterface $bus) {
    }

    protected function execute(InputInterface $input, OutputInterface $output): int {
        $this->bus->dispatch(new GoodJob(new ConsoleLogger($output))); // result will be printed in console
    }
}
```

#### Configuring

A minimal configuration:

```yaml
# messenger.yaml
framework:
  messenger:
    transports:
      schedule_default: 'schedule://default'
```

More complex example:

```yaml
# messenger.yaml
framework:
  messenger:
    transports:
      schedule_default:
        dsn: 'schedule://default'
        options:
          cache: 'app.cache'
          lock: 'default'
```

Example HA configuration with redis:

```yaml
framework:
  cache:
    default_redis_provider: '%env(REDIS_DSN)%'
  lock:
    redis: '%env(REDIS_DSN)%'
  messenger:
    transports:
      schedule_default:
        dsn: 'schedule://default'
        options:
          cache: 'cache.redis'
          lock:
            resource: 'redis'
            ttl: 60
            auto_release: true
```

### Deprecations

None

### Implementation

This PR contains an implementation.

### ToDo

- [x] Remove obsolete code
- [x] Add a configuration to the Framework
- [x] Specialize exceptions and improve messages
- [x] Cover with tests
- [ ] Add acceptance tests for HA
- [x] Fix `CHANGELOG`s & `README`s
- [ ] Add documentation

Commits
-------

a18127b789 [Scheduler] Rework the component
6d9311f087 Add the Scheduler component

Author: fabpot

DependencyInjection/Compiler/UnusedTagsPass.php

@@ -80,6 +80,7 @@ class UnusedTagsPass implements CompilerPassInterface
         'routing.expression_language_provider',
         'routing.loader',
         'routing.route_loader',
+        'scheduler.schedule_provider',
         'security.authenticator.login_linker',
         'security.expression_language_provider',
         'security.remember_me_aware',

DependencyInjection/Configuration.php

@@ -37,6 +37,7 @@
 use Symfony\Component\PropertyInfo\PropertyInfoExtractorInterface;
 use Symfony\Component\RateLimiter\Policy\TokenBucketLimiter;
 use Symfony\Component\RemoteEvent\RemoteEvent;
+use Symfony\Component\Scheduler\Messenger\SchedulerTransportFactory;
 use Symfony\Component\Semaphore\Semaphore;
 use Symfony\Component\Serializer\Serializer;
 use Symfony\Component\Translation\Translator;
@@ -173,6 +174,7 @@ public function getConfigTreeBuilder(): TreeBuilder
         $this->addLockSection($rootNode, $enableIfStandalone);
         $this->addSemaphoreSection($rootNode, $enableIfStandalone);
         $this->addMessengerSection($rootNode, $enableIfStandalone);
+        $this->addSchedulerSection($rootNode, $enableIfStandalone);
         $this->addRobotsIndexSection($rootNode);
         $this->addHttpClientSection($rootNode, $enableIfStandalone);
         $this->addMailerSection($rootNode, $enableIfStandalone);
@@ -1606,6 +1608,18 @@ function ($a) {
         ;
     }
 
+    private function addSchedulerSection(ArrayNodeDefinition $rootNode, callable $enableIfStandalone): void
+    {
+        $rootNode
+            ->children()
+                ->arrayNode('scheduler')
+                    ->info('Scheduler configuration')
+                    ->{$enableIfStandalone('symfony/scheduler', SchedulerTransportFactory::class)}()
+                ->end()
+            ->end()
+        ;
+    }
+
     private function addRobotsIndexSection(ArrayNodeDefinition $rootNode): void
     {
         $rootNode

DependencyInjection/FrameworkExtension.php

@@ -216,6 +216,8 @@
 use Symfony\Component\RemoteEvent\Attribute\AsRemoteEventConsumer;
 use Symfony\Component\RemoteEvent\RemoteEvent;
 use Symfony\Component\Routing\Loader\Psr4DirectoryLoader;
+use Symfony\Component\Scheduler\Attribute\AsSchedule;
+use Symfony\Component\Scheduler\Messenger\SchedulerTransportFactory;
 use Symfony\Component\Security\Core\AuthenticationEvents;
 use Symfony\Component\Security\Core\Exception\AuthenticationException;
 use Symfony\Component\Security\Csrf\CsrfTokenManagerInterface;
@@ -527,9 +529,18 @@ public function load(array $configs, ContainerBuilder $container)
         // validation depends on form, annotations being registered
         $this->registerValidationConfiguration($config['validation'], $container, $loader, $propertyInfoEnabled);
 
+        $messengerEnabled = $this->readConfigEnabled('messenger', $container, $config['messenger']);
+
+        if ($this->readConfigEnabled('scheduler', $container, $config['scheduler'])) {
+            if (!$messengerEnabled) {
+                throw new LogicException('Scheduler support cannot be enabled as the Messenger component is not '.(interface_exists(MessageBusInterface::class) ? 'enabled.' : 'installed. Try running "composer require symfony/messenger".'));
+            }
+            $this->registerSchedulerConfiguration($config['scheduler'], $container, $loader);
+        }
+
         // messenger depends on validation being registered
-        if ($this->readConfigEnabled('messenger', $container, $config['messenger'])) {
-            $this->registerMessengerConfiguration($config['messenger'], $container, $loader, $config['validation']);
+        if ($messengerEnabled) {
+            $this->registerMessengerConfiguration($config['messenger'], $container, $loader, $this->readConfigEnabled('validation', $container, $config['validation']));
         } else {
             $container->removeDefinition('console.command.messenger_consume_messages');
             $container->removeDefinition('console.command.messenger_stats');
@@ -707,10 +718,12 @@ public function load(array $configs, ContainerBuilder $container)
             }
             $definition->addTag('messenger.message_handler', $tagAttributes);
         });
-
         $container->registerAttributeForAutoconfiguration(AsTargetedValueResolver::class, static function (ChildDefinition $definition, AsTargetedValueResolver $attribute): void {
             $definition->addTag('controller.targeted_value_resolver', $attribute->name ? ['name' => $attribute->name] : []);
         });
+        $container->registerAttributeForAutoconfiguration(AsSchedule::class, static function (ChildDefinition $definition, AsSchedule $attribute): void {
+            $definition->addTag('scheduler.schedule_provider', ['name' => $attribute->name]);
+        });
 
         if (!$container->getParameter('kernel.debug')) {
             // remove tagged iterator argument for resource checkers
@@ -1996,7 +2009,20 @@ private function registerSemaphoreConfiguration(array $config, ContainerBuilder
         }
     }
 
-    private function registerMessengerConfiguration(array $config, ContainerBuilder $container, PhpFileLoader $loader, array $validationConfig): void
+    private function registerSchedulerConfiguration(array $config, ContainerBuilder $container, PhpFileLoader $loader): void
+    {
+        if (!class_exists(SchedulerTransportFactory::class)) {
+            throw new LogicException('Scheduler support cannot be enabled as the Scheduler component is not installed. Try running "composer require symfony/scheduler".');
+        }
+
+        if (!interface_exists(MessageBusInterface::class)) {
+            throw new LogicException('Scheduler support cannot be enabled as the Messenger component is not installed. Try running "composer require symfony/messenger".');
+        }
+
+        $loader->load('scheduler.php');
+    }
+
+    private function registerMessengerConfiguration(array $config, ContainerBuilder $container, PhpFileLoader $loader, bool $validationEnabled): void
     {
         if (!interface_exists(MessageBusInterface::class)) {
             throw new LogicException('Messenger support cannot be enabled as the Messenger component is not installed. Try running "composer require symfony/messenger".');
@@ -2058,7 +2084,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             }
 
             foreach ($middleware as $middlewareItem) {
-                if (!$validationConfig['enabled'] && \in_array($middlewareItem['id'], ['validation', 'messenger.middleware.validation'], true)) {
+                if (!$validationEnabled && \in_array($middlewareItem['id'], ['validation', 'messenger.middleware.validation'], true)) {
                     throw new LogicException('The Validation middleware is only available when the Validator component is installed and enabled. Try running "composer require symfony/validator".');
                 }
             }

FrameworkBundle.php

@@ -60,6 +60,7 @@
 use Symfony\Component\Mime\DependencyInjection\AddMimeTypeGuesserPass;
 use Symfony\Component\PropertyInfo\DependencyInjection\PropertyInfoPass;
 use Symfony\Component\Routing\DependencyInjection\RoutingResolverPass;
+use Symfony\Component\Scheduler\DependencyInjection\AddScheduleMessengerPass;
 use Symfony\Component\Serializer\DependencyInjection\SerializerPass;
 use Symfony\Component\Translation\DependencyInjection\TranslationDumperPass;
 use Symfony\Component\Translation\DependencyInjection\TranslationExtractorPass;
@@ -165,6 +166,7 @@ public function build(ContainerBuilder $container)
         $container->addCompilerPass(new TestServiceContainerWeakRefPass(), PassConfig::TYPE_BEFORE_REMOVING, -32);
         $container->addCompilerPass(new TestServiceContainerRealRefPass(), PassConfig::TYPE_AFTER_REMOVING);
         $this->addCompilerPassIfExists($container, AddMimeTypeGuesserPass::class);
+        $this->addCompilerPassIfExists($container, AddScheduleMessengerPass::class);
         $this->addCompilerPassIfExists($container, MessengerPass::class);
         $this->addCompilerPassIfExists($container, HttpClientPass::class);
         $this->addCompilerPassIfExists($container, AddAutoMappingConfigurationPass::class);

Resources/config/cache.php

@@ -71,6 +71,11 @@
             ->private()
             ->tag('cache.pool')
 
+        ->set('cache.scheduler')
+            ->parent('cache.app')
+            ->private()
+            ->tag('cache.pool')
+
         ->set('cache.adapter.system', AdapterInterface::class)
             ->abstract()
             ->factory([AbstractAdapter::class, 'createSystemCache'])

Resources/config/messenger.php

@@ -75,6 +75,7 @@
             ->tag('serializer.normalizer', ['priority' => -880])
 
         ->set('messenger.transport.native_php_serializer', PhpSerializer::class)
+        ->alias('messenger.default_serializer', 'messenger.transport.native_php_serializer')
 
         // Middleware
         ->set('messenger.middleware.handle_message', HandleMessageMiddleware::class)

Resources/config/scheduler.php

@@ -0,0 +1,25 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\DependencyInjection\Loader\Configurator;
+
+use Symfony\Component\Scheduler\Messenger\SchedulerTransportFactory;
+
+return static function (ContainerConfigurator $container) {
+    $container->services()
+        ->set('scheduler.messenger_transport_factory', SchedulerTransportFactory::class)
+            ->args([
+                tagged_locator('scheduler.schedule_provider', 'name'),
+                service('clock'),
+            ])
+            ->tag('messenger.transport_factory')
+    ;
+};

Resources/config/schema/symfony-1.0.xsd

@@ -24,6 +24,7 @@
             <xsd:element name="validation" type="validation" minOccurs="0" maxOccurs="1" />
             <xsd:element name="annotations" type="annotations" minOccurs="0" maxOccurs="1" />
             <xsd:element name="property-access" type="property_access" minOccurs="0" maxOccurs="1" />
+            <xsd:element name="scheduler" type="scheduler" minOccurs="0" maxOccurs="1" />
             <xsd:element name="serializer" type="serializer" minOccurs="0" maxOccurs="1" />
             <xsd:element name="property-info" type="property_info" minOccurs="0" maxOccurs="1" />
             <xsd:element name="cache" type="cache" minOccurs="0" maxOccurs="1" />
@@ -271,6 +272,10 @@
         <xsd:attribute name="throw-exception-on-invalid-property-path" type="xsd:boolean" />
     </xsd:complexType>
 
+    <xsd:complexType name="scheduler">
+        <xsd:attribute name="enabled" type="xsd:boolean" />
+    </xsd:complexType>
+
     <xsd:complexType name="serializer">
         <xsd:choice minOccurs="0" maxOccurs="unbounded">
             <xsd:element name="mapping" type="file_mapping" />

Tests/DependencyInjection/ConfigurationTest.php

@@ -26,6 +26,7 @@
 use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Notifier\Notifier;
 use Symfony\Component\RateLimiter\Policy\TokenBucketLimiter;
+use Symfony\Component\Scheduler\Messenger\SchedulerTransportFactory;
 use Symfony\Component\Uid\Factory\UuidFactory;
 
 class ConfigurationTest extends TestCase
@@ -687,6 +688,9 @@ class_exists(SemaphoreStore::class) && SemaphoreStore::isSupported() ? 'semaphor
                 'enabled' => !class_exists(FullStack::class) && class_exists(HtmlSanitizer::class),
                 'sanitizers' => [],
             ],
+            'scheduler' => [
+                'enabled' => !class_exists(FullStack::class) && class_exists(SchedulerTransportFactory::class),
+            ],
             'exceptions' => [],
             'webhook' => [
                 'enabled' => false,

Tests/DependencyInjection/Fixtures/php/messenger.php

@@ -5,6 +5,7 @@
 
 $container->loadFromExtension('framework', [
     'http_method_override' => false,
+    'scheduler' => true,
     'messenger' => [
         'routing' => [
             FooMessage::class => ['sender.bar', 'sender.biz'],