use RecoveryActionException to request Mesenger consumer restart

[issue #17] fix retry and faillure queue

.github/workflows/quality.yaml

@@ -0,0 +1,68 @@
+# This workflow uses actions that are not certified by GitHub.
+# They are provided by a third-party and are governed by
+# separate terms of service, privacy policy, and support
+# documentation.
+
+name: Quality
+
+on:
+  push:
+    branches: [ "2.x", "issue_17_failed_message_never_retried" ]
+  pull_request:
+    branches: [ "2.x" ]
+
+permissions:
+  contents: read
+
+jobs:
+  symfony-tests:
+    strategy:
+      fail-fast: false
+      matrix:
+        php: ['8.0', '8.1', '8.2', '8.3']
+    runs-on: ubuntu-latest
+
+    services:
+      mariadb:
+        image: mariadb:10.11
+        ports:
+          - 3306:3306
+        env:
+          MYSQL_USER: user
+          MYSQL_PASSWORD: nopassword
+          MYSQL_DATABASE: app_test
+          MYSQL_ROOT_PASSWORD: nopassword
+        options: --health-cmd="mysqladmin ping" --health-interval=5s --health-timeout=2s --health-retries=3
+
+
+    env:
+      DATABASE_URL: mysql://root:nopassword@127.0.0.1:3306/app?serverVersion=mariadb-10.11.2&charset=utf8mb4
+      APP_ENV: test
+    steps:
+      #  To automatically get bug fixes and new Php versions for shivammathur/setup-php,
+      # change this to (see https://github.com/shivammathur/setup-php#bookmark-versioning):
+      # uses: shivammathur/setup-php@v2
+      - uses: shivammathur/setup-php@v2
+        with:
+          php-version: ${{ matrix.php }}
+          coverage: xdebug
+
+      - uses: actions/checkout@v4
+#      - name: Copy .env.test.local
+#        run: php -r "file_exists('.env.test.local') || copy('.env.test', '.env.test.local');"
+      - name: Cache Composer packages
+        id: composer-cache
+        uses: actions/cache@v4
+        with:
+          path: vendor
+          key: ${{ runner.os }}-php${{ matrix.php }}-${{ hashFiles('**/composer.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-php${{ matrix.php }}-
+      - name: Install Dependencies
+        run: composer install -q --no-ansi --no-interaction --no-scripts --no-progress --prefer-dist
+      - name: Create Database
+        run: |
+          cd tests/Application
+          bin/console doctrine:migration:migrate -n
+      - name: Execute tests (Unit and Feature tests) via PHPUnit
+        run: vendor/bin/phpunit --process-isolation

.gitignore

@@ -1,2 +1,18 @@
-/vendor
 /composer.lock
+compose.override.yaml
+/.php-cs-fixer.cache
+
+###> symfony/framework-bundle ###
+/.env.local
+/tests/Application/.env.local
+/.env.local.php
+/tests/Application/.env.local.php
+/.env.*.local
+/tests/Application/.env.*.local
+/config/secrets/prod/prod.decrypt.private.php
+/public/bundles/
+/tests/Application/public/bundles/
+/var/
+/tests/Application/var
+/vendor/
+###< symfony/framework-bundle ###

.php-cs-fixer.dist.php

@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+$finder = PhpCsFixer\Finder::create()
+    ->in('lib')
+    ->in('tests')
+    ->files()->name('*.php');
+
+$config = new PhpCsFixer\Config();
+$config->setRules([
+    '@Symfony' => true,
+    '@Symfony:risky' => true,
+    '@PSR12' => true,
+    'array_syntax' => [
+        'syntax' => 'short',
+    ],
+    'declare_strict_types' => true,
+    'constant_case' => true,
+    'combine_consecutive_unsets' => true,
+    'native_function_invocation' => [
+        'include' => [
+            '@compiler_optimized',
+        ],
+    ],
+    'no_extra_blank_lines' => [
+        'tokens' => [
+            'break',
+            'continue',
+            'extra',
+            'return',
+            'throw',
+            'use',
+            'parenthesis_brace_block',
+            'square_brace_block',
+            'curly_brace_block',
+        ],
+    ],
+    'ordered_class_elements' => true,
+    'ordered_imports' => true,
+    'yoda_style' => [
+        'equal' => false,
+        'identical' => false,
+        'less_and_greater' => false,
+        'always_move_variable' => false,
+    ],
+])
+    ->setRiskyAllowed(true)
+    ->setFinder(
+        $finder
+    );
+
+return $config;

.php-version

@@ -0,0 +1 @@
+8.0

README.md

@@ -20,6 +20,7 @@ win32_service:
         channels: # The list of channels whom the processor add the thread number. If empty, the thread number is added for all channels.
          - ref1
     services:
+      # Prototype
       -
         service_id: "" # The service id. If `thread_count` > 1, you can use `%d` to define the place of the thread number
         machine: "" # the machine name for this service. 
@@ -47,6 +48,22 @@ win32_service:
           reset_period: 86400 # The period before reset the fail count (in minutes)
         dependencies: # The list of service depends
           - Netman # An example of dependency.
+ 
+    messenger: # This configuration allow to set on Windows service the Symfony Messenger Consumer.
+      # Prototype
+      -
+        user: # The User set on the service
+          account:            ~ # the account name
+          password:           ~ # the password
+        receivers:            [] # Symfony Messenger transport consumed by the service
+        machine:              '' # the machine name for this service. 
+        displayed_name:       ~ # Required, The friendly name of the service. If `thread_count` > 1, you can use `%d` to define the place of the thread number
+        description:          '' # the service description
+        thread_count:         1 # the number of this service need to register. Use `%d` into `service_id`, `displayed_name` and `script_params` for contains the service number.
+        delayed_start:        false # If true the starting for the service is delayed
+        limit:                0 # Reboot service after processed X messages
+        failure-limit:        0 # Reboot service after X messages failure
+        time-limit:           0 # Reboot service after X seconds
 ```
 
 # Define the runner

composer.json

@@ -20,19 +20,49 @@
   },
   "autoload-dev": {
     "psr-4": {
+      "Win32ServiceBundle\\Tests\\Application\\": "tests/Application/src/",
       "Win32ServiceBundle\\Tests\\": "tests/"
     }
   },
   "minimum-stability": "beta",
   "require": {
-    "php": "^7.1.0 <8.0.0",
-    "win32service/service-library": "^0.1.0",
-    "symfony/http-kernel": "^4.0||^5.0",
-    "symfony/console": "^4.0||^5.0",
-    "symfony/config": "^4.0||^5.0",
-    "symfony/dependency-injection": "^4.0||^5.0"
+    "php": "^8.0",
+    "win32service/service-library": "^1.0||1.x-dev",
+    "symfony/http-kernel": "^5.4|^6.0",
+    "symfony/console": "^5.4|^6.0",
+    "symfony/config": "^5.4|^6.0",
+    "symfony/dependency-injection": "^5.4|^6.0",
+    "symfony/event-dispatcher": "^5.4|^6.0",
+    "symfony/yaml": "^5.4|^6.0",
+    "symfony/framework-bundle": "^5.4|^6.0"
   },
   "require-dev": {
-    "atoum/atoum": "^3.3"
+    "rector/rector": "^0.14.2",
+    "symfony/dotenv": "^5.4|^6.0",
+    "symfony/runtime": "^5.4|^6.0",
+    "symfony/messenger": "^5.4|^6.0",
+    "symfony/doctrine-messenger":  "^5.4|^6.0",
+    "symfony/monolog-bundle": "^3.10",
+    "doctrine/doctrine-bundle": "^2.12",
+    "doctrine/doctrine-migrations-bundle": "^3.3",
+    "doctrine/orm": "^2.19",
+    "phpunit/phpunit": "^9.6"
+  },
+  "conflict": {
+    "win32service/service-library": "<1.0.2"
+  },
+  "suggest": {
+    "ext-win32service": "On Windows only, install this extension to run PHP Service on Windows Service Manager"
+  },
+  "extra": {
+    "symfony": {
+      "allow-contrib": false,
+      "require": "^5.4"
+    }
+  },
+  "config": {
+    "allow-plugins": {
+      "symfony/runtime": true
+    }
   }
 }

lib/Command/ActionServiceCommand.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copy Win32Service (c) 2019
  * Added by : macintoshplus at 19/02/19 13:59
@@ -6,113 +8,118 @@
 
 namespace Win32ServiceBundle\Command;
 
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use Win32Service\Model\ServiceIdentifier;
+use Win32Service\Model\ServiceInformations;
 use Win32Service\Service\ServiceStateManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
 
+#[AsCommand(name: 'win32service:action')]
 class ActionServiceCommand extends Command
 {
-    // the name of the command (the part after "bin/console")
-    protected static $defaultName = 'win32service:action';
+    public const ALL_SERVICE = 'All';
 
-    const ALL_SERVICE = 'All';
-
-    /**
-     * @var array
-     */
-    private $config;
-
-    protected function configure()
+    public function __construct(private ServiceConfigurationManager $serviceConfigurationManager)
     {
-        $this->setDescription("Send the action at all service");
-        $this->addArgument('control', InputArgument::REQUIRED, "The action you want");
-        $this->addOption('service-name', 's', InputOption::VALUE_REQUIRED, 'Send the controle to the service with service_id. The value must be equal to the configuration.', self::ALL_SERVICE);
-        $this->addOption('custom-action', 'c', InputOption::VALUE_REQUIRED, 'The custom control send to the service.', null);
+        parent::__construct();
     }
 
-    /**
-     * @param array $config
-     *
-     */
-    public function defineBundleConfig(array $config) {
-        $this->config = $config;
-
+    protected function configure(): void
+    {
+        $this->setDescription('Send the action at all service');
+        $this->addArgument('control', InputArgument::REQUIRED, 'The action you want');
+        $this->addOption(
+            'service-name',
+            's',
+            InputOption::VALUE_REQUIRED,
+            'Send the controle to the service with service_id. The value must be equal to the configuration.',
+            self::ALL_SERVICE
+        );
+        $this->addOption(
+            'custom-action',
+            'c',
+            InputOption::VALUE_REQUIRED,
+            'The custom control send to the service.',
+            null
+        );
     }
 
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        if ($this->config === null) {
-            throw new \Exception('The configuration of win32Service is not defined into command');
-        }
-
         $serviceToAction = $input->getOption('service-name');
         $customAction = $input->getOption('custom-action');
         $action = $input->getArgument('control');
 
+        $adminService = new ServiceStateManager();
+
         $actions = ['start', 'stop', 'pause', 'continue', 'custom'];
-        if (!in_array($action, $actions)) {
+        if (!\in_array($action, $actions)) {
             throw new \InvalidArgumentException('The value of action argument is invalid. Valid values : '.implode(', ', $actions));
         }
 
         if ($action === 'custom' && ($customAction < 128 || $customAction > 255)) {
-            throw new \InvalidArgumentException("The custom control value must be between 128 and 255");
+            throw new \InvalidArgumentException('The custom control value must be between 128 and 255');
         }
 
-        $services = $this->config['services'];
+        if ($serviceToAction !== self::ALL_SERVICE) {
+            $serviceInfos = $this->serviceConfigurationManager->getServiceInformations($serviceToAction);
 
-        $adminService = new ServiceStateManager();
+            $this->sendAction($adminService, $action, $serviceInfos, $customAction);
+            $output->writeln('Sending control to <info>'.$serviceInfos->serviceId().'</info> : OK');
+
+            return self::SUCCESS;
+        }
 
         $nbService = 0;
-        foreach ($services as $service) {
-
-            $threadNumber = $service['thread_count'];
-
-            for ($i = 0; $i < $threadNumber; $i++) {
-                $serviceThreadId = sprintf($service['service_id'], $i);
-                if ($serviceToAction !== self::ALL_SERVICE && $serviceToAction !== $service['service_id'] && $serviceThreadId !== $serviceToAction) {
-                    continue;
-                }
-
-                $nbService++;
-                //Init the service informations
-                $serviceInfos = ServiceIdentifier::identify($serviceThreadId, $service['machine']);
-
-                try {
-                    switch ($action) {
-                        case 'start':
-                            $adminService->startService($serviceInfos);
-                            break;
-                        case 'stop':
-                            $adminService->stopService($serviceInfos);
-                            break;
-                        case 'pause':
-                            $adminService->pauseService($serviceInfos);
-                            break;
-                        case 'continue':
-                            $adminService->continueService($serviceInfos);
-                            break;
-                        case 'custom':
-                            $adminService->sendCustomControl($serviceInfos, $customAction);
-                            break;
-                        default:
-                            break;
-                    }
-                    $output->writeln('Sending control to <info>' . $serviceInfos->serviceId() . '</info> : OK');
-                } catch (\Exception $e) {
-                    $output->writeln('<error> Error : ' . $serviceInfos->serviceId() . '(' . $e->getCode() . ') ' . $e->getMessage() . ' </error>');
-                }
+        foreach ($this->serviceConfigurationManager->getFullServiceList() as $serviceInfos) {
+            try {
+                ++$nbService;
+                $this->sendAction($adminService, $action, $serviceInfos, $customAction);
+                $output->writeln('Sending control to <info>'.$serviceInfos->serviceId().'</info> : OK');
+            } catch (\Exception $e) {
+                $output->writeln('<error> Error : '.$serviceInfos->serviceId().'('.$e->getCode().') '.$e->getMessage().' </error>');
             }
         }
 
         if ($nbService === 0) {
             $output->writeln('<info>No signal sent</info>');
-            return;
+
+            return self::FAILURE;
         }
 
         $output->writeln(sprintf('Signal sent to <info>%d</info> service(s)', $nbService));
+
+        return self::SUCCESS;
+    }
+
+    private function sendAction(
+        ServiceStateManager $adminService,
+        string $action,
+        ServiceInformations $serviceInfos,
+        ?int $customAction
+    ): void {
+        switch ($action) {
+            case 'start':
+                $adminService->startService($serviceInfos);
+                break;
+            case 'stop':
+                $adminService->stopService($serviceInfos);
+                break;
+            case 'pause':
+                $adminService->pauseService($serviceInfos);
+                break;
+            case 'continue':
+                $adminService->continueService($serviceInfos);
+                break;
+            case 'custom':
+                $adminService->sendCustomControl($serviceInfos, $customAction);
+                break;
+            default:
+                break;
+        }
     }
 }

lib/Command/ExecuteServiceCommand.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copyright Win32Service (c) 2019
  * Added by : macintoshplus at 19/02/19 21:18
@@ -6,6 +8,8 @@
 
 namespace Win32ServiceBundle\Command;
 
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
@@ -13,104 +17,62 @@ use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 use Win32Service\Model\ServiceIdentifier;
-use Win32Service\Model\RunnerServiceInterface;
 use Win32ServiceBundle\Logger\ThreadNumberEvent;
 use Win32ServiceBundle\Service\RunnerManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
 
+#[AsCommand(name: 'win32service:run')]
 class ExecuteServiceCommand extends Command
 {
-    // the name of the command (the part after "bin/console")
-    protected static $defaultName = 'win32service:run';
+    public function __construct(
+        private ServiceConfigurationManager $serviceConfigurationManager,
+        private RunnerManager $service,
+        private ?EventDispatcherInterface $eventDispatcher = null,
+        private ?LoggerInterface $logger = null
+    ) {
+        parent::__construct();
+    }
 
-    /**
-     * @var array
-     */
-    private $config;
-
-    /**
-     * @var RunnerManager
-     */
-    private $service;
-
-    /**
-     * @var EventDispatcherInterface
-     */
-    private $eventDispatcher;
-
-    protected function configure()
+    protected function configure(): void
     {
-        $this->setDescription("Run the service");
+        $this->setDescription('Run the service');
         $this->addArgument('service-name', InputArgument::REQUIRED, 'The service name.');
         $this->addArgument('thread', InputArgument::REQUIRED, 'Thread number');
         $this->addOption('max-run', 'r', InputOption::VALUE_REQUIRED, 'Set the max run');
     }
 
-    /**
-     * @param array $config
-     *
-     */
-    public function defineBundleConfig(array $config) {
-        $this->config = $config;
-
-    }
-
-    public function setService(RunnerManager $service) {
-        $this->service = $service;
-    }
-
-    /**
-     * @param EventDispatcherInterface $eventDispatcher
-     */
-    public function setEventDispatcher(EventDispatcherInterface $eventDispatcher) {
-        $this->eventDispatcher = $eventDispatcher;
-    }
-
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        if ($this->config === null) {
-            throw new \Exception('The configuration of win32Service is not defined into command');
-        }
-        if ($this->service === null) {
-            throw new \Exception('The service runner manager is not defined into command');
-        }
-
         $serviceName = $input->getArgument('service-name');
-        $threadNumber = $input->getArgument('thread');
+        $threadNumber = (int) $input->getArgument('thread') ?? 1;
         $maxRun = $input->getOption('max-run');
 
-        $infos=$this->getServiceInformation($serviceName, $threadNumber);
-        if ($infos === null) {
-            throw new \Exception(sprintf('The information for service %s is not found', $serviceName));
-        }
+        $infos = $this->serviceConfigurationManager->getServiceInformations($serviceName);
 
         if ($maxRun === null) {
-            $maxRun = $infos['run_max'];
+            $maxRun = -1;
         }
 
-        $runner = $this->service->getRunner($infos['service_id']);
+        $runner = $this->service->getRunner($this->serviceConfigurationManager->getRunnerAliasForServiceId($serviceName));
         if ($runner === null) {
-            throw new \Exception(sprintf('The runner for service "%1$s" is not found. Call method \'add\' on the RunnerManager with the runner instance and the alias "%1$s".', $infos['service_id']));
+            throw new \Win32ServiceException(sprintf('The runner for service "%1$s" is not found. Call method \'add\' on the RunnerManager with the runner instance and the alias "%1$s".', $infos['service_id']));
         }
 
         if ($this->eventDispatcher !== null) {
             $event = new ThreadNumberEvent($threadNumber);
-            $this->eventDispatcher->dispatch(ThreadNumberEvent::NAME, $event);
+            $this->eventDispatcher->dispatch($event, ThreadNumberEvent::NAME);
         }
 
-        $runner->setServiceId(ServiceIdentifier::identify($serviceName, $infos['machine']));
+        $runner->setServiceId(ServiceIdentifier::identify($serviceName, $infos->machine()));
+        $rawConfig = $this->serviceConfigurationManager->getServiceRawConfiguration($serviceName);
+        $this->logger?->info(
+            'Configure exit graceful and code',
+            ['exit_graceful' => $rawConfig['exit']['graceful'], 'exit_code' => $rawConfig['exit']['code']]
+        );
+        $runner->defineExitModeAndCode($rawConfig['exit']['graceful'], $rawConfig['exit']['code']);
 
-        $runner->defineExitModeAndCode($infos['exit']['graceful'], $infos['exit']['code']);
+        $runner->doRun((int) $maxRun, $threadNumber);
 
-        $runner->doRun(intval($maxRun), $threadNumber);
-
-    }
-
-    private function getServiceInformation(string $serviceToRun, $threadNumber) {
-        foreach ($this->config['services'] as $service) {
-            if ($serviceToRun === sprintf($service['service_id'], $threadNumber)) {
-                return $service;
-            }
-        }
-        return null;
+        return self::SUCCESS;
     }
 }

lib/Command/ListServiceCommand.php

@@ -0,0 +1,79 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Command;
+
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Win32Service\Model\RunnerServiceInterface;
+use Win32ServiceBundle\DependencyInjection\TagRunnerCompilerPass;
+use Win32ServiceBundle\Service\RunnerManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
+
+#[AsCommand(name: 'win32service:list', description: 'List all services')]
+final class ListServiceCommand extends Command
+{
+    public function __construct(
+        private RunnerManager $runnerManager,
+        private ServiceConfigurationManager $serviceConfigurationManager
+    ) {
+        parent::__construct();
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output);
+        $io->title('List all services');
+        $allAlias = array_keys($this->runnerManager->getRunners());
+        $data = [];
+
+        foreach ($this->serviceConfigurationManager->getFullServiceList() as $serviceInformations) {
+            try {
+                if (\function_exists('win32_query_service_status') === false) {
+                    throw new \Win32ServiceException('Win32Service extension is not installed.');
+                }
+                $status = win32_query_service_status(
+                    $serviceInformations->serviceId(),
+                    $serviceInformations->machine()
+                );
+                $status = match ($status['CurrentState']) {
+                    WIN32_SERVICE_CONTINUE_PENDING => 'continue pending',
+                    WIN32_SERVICE_PAUSE_PENDING => 'pause pending',
+                    WIN32_SERVICE_PAUSED => 'paused',
+                    WIN32_SERVICE_RUNNING => 'running',
+                    WIN32_SERVICE_START_PENDING => 'start pending',
+                    WIN32_SERVICE_STOP_PENDING => 'stop pending',
+                    WIN32_SERVICE_STOPPED => 'stopped',
+                    default => 'Unknow',
+                };
+            } catch (\Win32ServiceException $exception) {
+                $status = match ($exception->getCode()) {
+                    WIN32_ERROR_ACCESS_DENIED => 'Access denied',
+                    WIN32_ERROR_SERVICE_DOES_NOT_EXIST => 'Not registred',
+                    default => $exception->getMessage(),
+                };
+            }
+            $runnerTagAlias = $this->serviceConfigurationManager->getRunnerAliasForServiceId($serviceInformations->serviceId());
+            $data[] = [
+                empty($serviceInformations->machine()) ? 'localhost' : $serviceInformations->machine(),
+                $serviceInformations->serviceId(),
+                \in_array($runnerTagAlias, $allAlias) ? '<info>OK</info>' : sprintf(
+                    '<error>No Symfony service implements "%s" with tag "name: \'%s\', alias: \'%s\'"</error>',
+                    RunnerServiceInterface::class,
+                    TagRunnerCompilerPass::WIN32SERVICE_RUNNER_TAG,
+                    $runnerTagAlias,
+                ),
+                $status,
+                $serviceInformations[WIN32_INFO_DISPLAY],
+            ];
+        }
+
+        $io->table(['Machine', 'ServiceId', 'Runner config', 'State', 'Name'], $data);
+
+        return self::SUCCESS;
+    }
+}

lib/Command/RegisterServiceCommand.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copy Win32Service (c) 2019
  * Added by : macintoshplus at 19/02/19 13:59
@@ -6,130 +8,70 @@
 
 namespace Win32ServiceBundle\Command;
 
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use Win32Service\Model\ServiceIdentifier;
-use Win32Service\Model\ServiceInformations;
 use Win32Service\Service\ServiceAdminManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
 
+#[AsCommand(name: 'win32service:register')]
 class RegisterServiceCommand extends Command
 {
-    // the name of the command (the part after "bin/console")
-    protected static $defaultName = 'win32service:register';
+    public const ALL_SERVICE = 'All';
 
-    const ALL_SERVICE = 'All';
-
-    /**
-     * @var array
-     */
-    private $config;
-
-    /**
-     * @var string
-     */
-    private $projectRoot;
-
-    protected function configure()
+    public function __construct(private ServiceConfigurationManager $serviceConfigurationManager)
     {
-        $this->setDescription("Register all service into Windows Service Manager");
-        $this->addOption('service-name', 's', InputOption::VALUE_REQUIRED, 'Register the service with service_id. The value must be equal to the configuration.', self::ALL_SERVICE);
+        parent::__construct();
     }
 
-    /**
-     * @param array $config
-     *
-     */
-    public function defineBundleConfig(array $config) {
-        $this->config = $config;
-    }
-
-    /**
-     * @param string $projectRoot
-     * @required
-     */
-    public function defineProjectRoot(string $projectRoot) {
-        $this->projectRoot = $projectRoot;
-    }
-
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function configure(): void
     {
-        if ($this->config === null) {
-            throw new \Exception('The configuration of win32Service is not defined into command');
-        }
+        $this->setDescription('Register all service into Windows Service Manager');
+        $this->addOption(
+            'service-name',
+            's',
+            InputOption::VALUE_REQUIRED,
+            'Register the service with service_id. The value must be equal to the configuration.',
+            self::ALL_SERVICE
+        );
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $adminService = new ServiceAdminManager();
 
         $serviceToRegister = $input->getOption('service-name');
 
-        $services = $this->config['services'];
+        if ($serviceToRegister !== self::ALL_SERVICE) {
+            $serviceInfos = $this->serviceConfigurationManager->getServiceInformations($serviceToRegister);
 
-        $adminService = new ServiceAdminManager();
+            $adminService->registerService($serviceInfos);
+            $output->writeln('Registration success for <info>'.$serviceInfos->serviceId().'</info>');
 
-        $windowsLocalEncoding = $this->config['windows_local_encoding'];
+            return self::SUCCESS;
+        }
 
         $nbService = 0;
-        foreach ($services as $service) {
-            if ($serviceToRegister !== self::ALL_SERVICE && $serviceToRegister !== $service['service_id']) {
-                continue;
-            }
-            $threadNumber = $service['thread_count'];
-
-            for ($i = 0; $i < $threadNumber; $i++) {
-                $nbService++;
-                //Init the service informations
-                $serviceThreadId = sprintf($service['service_id'], $i);
-                $path = $service['script_path'];
-                $args = sprintf($service['script_params'], $i);
-                if ($path === null) {
-                    $path = realpath($_SERVER['PHP_SELF']);
-                    //$path = sprintf('%s\\bin\\console', $this->projectRoot);
-                    $args = sprintf('%s %s %d', ExecuteServiceCommand::getDefaultName(),$serviceThreadId, $i );
-                }
-
-                $serviceInfos = new ServiceInformations(
-                    ServiceIdentifier::identify($serviceThreadId, $service['machine']),
-                    mb_convert_encoding(sprintf($service['displayed_name'], $i), $windowsLocalEncoding, 'UTF-8'),
-                    mb_convert_encoding($service['description'], $windowsLocalEncoding, 'UTF-8'),
-                    mb_convert_encoding($path, $windowsLocalEncoding, 'UTF-8'),
-                    mb_convert_encoding($args, $windowsLocalEncoding, 'UTF-8')
-                );
-
-                $serviceInfos->defineIfStartIsDelayed($service['delayed_start']);
-
-                $recovery=$service['recovery'];
-                $serviceInfos->defineRecoverySettings(
-                    $recovery['delay'],
-                    $recovery['enable'],
-                    $recovery['action1'],
-                    $recovery['action2'],
-                    $recovery['action3'],
-                    $recovery['reboot_msg'],
-                    $recovery['command'],
-                    $recovery['reset_period']
-                );
-
-                if ($service['user']['account'] !== null) {
-                    $serviceInfos->defineUserService($service['user']['account'], $service['user']['password']);
-                }
-
-                if (count($service['dependencies']) >0) {
-                    $serviceInfos->defineDependencies($service['dependencies']);
-                }
-
-                try {
-                    $adminService->registerService($serviceInfos);
-                    $output->writeln('Registration success for <info>' . $serviceInfos->serviceId() . '</info>');
-                } catch (\Exception $e) {
-                    $output->writeln('<error> Error : ' . $serviceInfos->serviceId() . '(' . $e->getCode() . ') ' . $e->getMessage() . ' </error>');
-                }
+        foreach ($this->serviceConfigurationManager->getFullServiceList() as $serviceInfos) {
+            try {
+                $adminService->registerService($serviceInfos);
+                ++$nbService;
+                $output->writeln('Registration success for <info>'.$serviceInfos->serviceId().'</info>');
+            } catch (\Exception $e) {
+                $output->writeln('<error> Error : '.$serviceInfos->serviceId().'('.$e->getCode().') '.$e->getMessage().' </error>');
             }
         }
 
         if ($nbService === 0) {
             $output->writeln('<info>No service registred</info>');
-            return;
+
+            return self::FAILURE;
         }
 
         $output->writeln(sprintf('<info>%d</info> service(s) processed', $nbService));
+
+        return self::SUCCESS;
     }
 }

lib/Command/UnregisterServiceCommand.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copy Win32Service (c) 2019
  * Added by : macintoshplus at 19/02/19 15:59
@@ -6,78 +8,69 @@
 
 namespace Win32ServiceBundle\Command;
 
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use Win32Service\Model\ServiceIdentifier;
 use Win32Service\Service\ServiceAdminManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
 
+#[AsCommand(name: 'win32service:unregister')]
 class UnregisterServiceCommand extends Command
 {
-    // the name of the command (the part after "bin/console")
-    protected static $defaultName = 'win32service:unregister';
+    public const ALL_SERVICE = 'All';
 
-    const ALL_SERVICE = 'All';
-
-    /**
-     * @var array
-     */
-    private $config;
-
-    protected function configure()
+    public function __construct(private ServiceConfigurationManager $serviceConfigurationManager)
     {
-        $this->setDescription("Unregister all service into Windows Service Manager");
-        $this->addOption('service-name', 's', InputOption::VALUE_REQUIRED, 'Register the service with service_id. The value must be equal to the configuration.', self::ALL_SERVICE);
+        parent::__construct();
     }
 
-    /**
-     * @param array $config
-     *
-     */
-    public function defineBundleConfig(array $config) {
-        $this->config = $config;
-
+    protected function configure(): void
+    {
+        $this->setDescription('Unregister all service into Windows Service Manager');
+        $this->addOption(
+            'service-name',
+            's',
+            InputOption::VALUE_REQUIRED,
+            'Register the service with service_id. The value must be equal to the configuration.',
+            self::ALL_SERVICE
+        );
     }
 
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        if ($this->config === null) {
-            throw new \Exception('The configuration of win32Service is not defined into command');
-        }
-
-        $serviceToRegister = $input->getOption('service-name');
-
-        $services = $this->config['services'];
-
+        $serviceToUnregister = $input->getOption('service-name');
         $adminService = new ServiceAdminManager();
 
+        if ($serviceToUnregister !== self::ALL_SERVICE) {
+            $serviceInfos = $this->serviceConfigurationManager->getServiceInformations($serviceToUnregister);
+
+            $adminService->unregisterService($serviceInfos);
+            $output->writeln('Unregistration success for <info>'.$serviceInfos->serviceId().'</info>');
+
+            return self::SUCCESS;
+        }
+
         $nbService = 0;
-        foreach ($services as $service) {
-            if ($serviceToRegister !== self::ALL_SERVICE && $serviceToRegister !== $service['service_id']) {
-                continue;
-            }
-            $threadNumber = $service['thread_count'];
-
-            for ($i = 0; $i < $threadNumber; $i++) {
-                $nbService++;
-                //Init the service informations
-                $serviceInfos = ServiceIdentifier::identify(sprintf($service['service_id'], $i), $service['machine']);
-
-                try {
-                    $adminService->unregisterService($serviceInfos);
-                    $output->writeln('Unregistration success for <info>' . $serviceInfos->serviceId() . '</info>');
-                } catch (\Exception $e) {
-                    $output->writeln('<error> Error : ' . $serviceInfos->serviceId() . '(' . $e->getCode() . ') ' . $e->getMessage() . ' </error>');
-                }
+        foreach ($this->serviceConfigurationManager->getFullServiceList() as $serviceInfos) {
+            try {
+                $adminService->unregisterService($serviceInfos);
+                ++$nbService;
+                $output->writeln('Unregistration success for <info>'.$serviceInfos->serviceId().'</info>');
+            } catch (\Exception $e) {
+                $output->writeln('<error> Error : '.$serviceInfos->serviceId().'('.$e->getCode().') '.$e->getMessage().' </error>');
             }
         }
 
         if ($nbService === 0) {
             $output->writeln('<info>No service unregistred</info>');
-            return;
+
+            return self::FAILURE;
         }
 
         $output->writeln(sprintf('<info>%d</info> service(s) processed', $nbService));
+
+        return self::SUCCESS;
     }
 }

lib/DependencyInjection/Configuration.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copy Win32Service (c) 2019
  * Added by : macintoshplus at 19/02/19 13:34
@@ -18,6 +20,13 @@ class Configuration implements ConfigurationInterface
         $treeBuilder->getRootNode()
             ->children()
                 ->scalarNode('windows_local_encoding')->defaultValue('ISO-8859-15')->end()
+                ->scalarNode('project_code')->isRequired()->cannotBeEmpty()->info('Project specific code to distinguish service ID')
+                    ->validate()
+                        ->ifTrue(function ($value) {
+                            return \is_string($value) === false || \strlen($value) > 5 || \strlen($value) < 2;
+                        })->thenInvalid('Invalid project code (string length between 2 and 5 chars)')
+                    ->end()
+                ->end()
                 ->arrayNode('logging_extra')
                     ->addDefaultsIfNotSet()
                     ->children()
@@ -72,8 +81,8 @@ class Configuration implements ConfigurationInterface
                                         ->values([WIN32_SC_ACTION_NONE, WIN32_SC_ACTION_REBOOT, WIN32_SC_ACTION_RESTART, WIN32_SC_ACTION_RUN_COMMAND])
                                         ->defaultValue(WIN32_SC_ACTION_NONE)
                                     ->end()
-                                    ->scalarNode('reboot_msg')->defaultValue("")->end()
-                                    ->scalarNode('command')->defaultValue("")->end()
+                                    ->scalarNode('reboot_msg')->defaultValue('')->end()
+                                    ->scalarNode('command')->defaultValue('')->end()
                                     ->integerNode('reset_period')->defaultValue(86400)->min(1)->end()
                                 ->end()
                             ->end()
@@ -85,6 +94,31 @@ class Configuration implements ConfigurationInterface
                         ->end()
                     ->end()
                 ->end()
+                ->arrayNode('messenger')
+                    ->arrayPrototype()
+                        ->children()
+                            ->arrayNode('user')
+                                ->addDefaultsIfNotSet()
+                                ->children()
+                                    ->scalarNode('account')->defaultNull()->end()
+                                    ->scalarNode('password')->defaultNull()->end()
+                                ->end()
+                            ->end()
+                            ->arrayNode('receivers')
+                                ->scalarPrototype()->end()
+                            ->end()
+                            ->scalarNode('machine')->defaultValue('')->end()
+                            ->scalarNode('displayed_name')->isRequired()->cannotBeEmpty()->end()
+                            ->scalarNode('description')->defaultValue('')->end()
+                            ->integerNode('thread_count')->defaultValue(1)->min(1)->end()
+                            ->booleanNode('delayed_start')->defaultFalse()->end()
+                            ->integerNode('limit')->defaultValue(0)->min(0)->end()
+                            ->integerNode('failure_limit')->defaultValue(0)->min(0)->end()
+                            ->integerNode('time_limit')->defaultValue(0)->min(0)->end()
+                            ->scalarNode('memory_limit')->defaultValue('')->end()
+                        ->end()
+                    ->end()
+                ->end()
             ->end()
         ;
 

lib/DependencyInjection/MessengerPass.php

@@ -0,0 +1,95 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\DependencyInjection;
+
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Exception\OutOfBoundsException;
+use Symfony\Component\DependencyInjection\Reference;
+use Win32ServiceBundle\MessengerSubscriber\ResetServicesListener;
+use Win32ServiceBundle\MessengerSubscriber\SendFailedMessageForRetryListener;
+use Win32ServiceBundle\MessengerSubscriber\SendFailedMessageToFailureTransportListener;
+
+final class MessengerPass implements CompilerPassInterface
+{
+    private string $busTag = 'messenger.bus';
+    private string $receiverTag = 'messenger.receiver';
+    private string $win32ServiceRunnerTag = TagRunnerCompilerPass::WIN32SERVICE_RUNNER_TAG.'.messenger';
+
+    public function process(ContainerBuilder $container): void
+    {
+        $this->processService($container);
+        $this->processRetryConfig($container);
+        $this->processFailledConfig($container);
+    }
+
+    private function processService(ContainerBuilder $container): void
+    {
+        $busIds = [];
+        foreach ($container->findTaggedServiceIds($this->busTag) as $busId => $tags) {
+            $busIds[] = $busId;
+        }
+
+        $receiverMapping = [];
+        foreach ($container->findTaggedServiceIds($this->receiverTag) as $id => $tags) {
+            $receiverMapping[$id] = new Reference($id);
+
+            foreach ($tags as $tag) {
+                if (isset($tag['alias'])) {
+                    $receiverMapping[$tag['alias']] = $receiverMapping[$id];
+                }
+            }
+        }
+
+        $receiverNames = [];
+        foreach ($receiverMapping as $name => $reference) {
+            $receiverNames[(string) $reference] = $name;
+        }
+
+        foreach ($container->findTaggedServiceIds($this->win32ServiceRunnerTag) as $win32ServiceId => $tags) {
+            $serviceRunnerDefinition = $container->getDefinition($win32ServiceId);
+
+            $serviceRunnerDefinition->replaceArgument(1, new Reference('messenger.routable_message_bus'));
+            $serviceRunnerDefinition->replaceArgument(7, new Reference(ResetServicesListener::class));
+
+            $serviceRunnerDefinition->replaceArgument(6, array_values($receiverNames));
+            try {
+                $serviceRunnerDefinition->replaceArgument(8, $busIds);
+            } catch (OutOfBoundsException $e) {
+                // ignore to preserve compatibility with symfony/framework-bundle < 5.4
+            }
+        }
+    }
+
+    private function processFailledConfig(ContainerBuilder $container): void
+    {
+        if (
+            $container->hasDefinition('messenger.failure.send_failed_message_to_failure_transport_listener') === false
+            || $container->hasDefinition(SendFailedMessageToFailureTransportListener::class) === false
+        ) {
+            return;
+        }
+
+        $serviceSF = $container->findDefinition('messenger.failure.send_failed_message_to_failure_transport_listener');
+
+        $serviceWin32 = $container->findDefinition(SendFailedMessageToFailureTransportListener::class);
+        $serviceWin32->replaceArgument('$failureSenders', $serviceSF->getArgument(0));
+    }
+
+    private function processRetryConfig(ContainerBuilder $container): void
+    {
+        if (
+            $container->hasDefinition('messenger.retry.send_failed_message_for_retry_listener') === false
+            || $container->hasDefinition(SendFailedMessageForRetryListener::class) === false
+        ) {
+            return;
+        }
+
+        $serviceSF = $container->findDefinition('messenger.retry.send_failed_message_for_retry_listener');
+
+        $serviceWin32 = $container->findDefinition(SendFailedMessageForRetryListener::class);
+        $serviceWin32->replaceArgument('$sendersLocator', $serviceSF->getArgument(0));
+    }
+}

lib/DependencyInjection/TagRunnerCompilerPass.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copyright Macintoshplus (c) 2019
  * Added by : Macintoshplus at 19/02/19 23:09
@@ -6,14 +8,16 @@
 
 namespace Win32ServiceBundle\DependencyInjection;
 
-use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Reference;
 use Win32ServiceBundle\Service\RunnerManager;
 
 class TagRunnerCompilerPass implements CompilerPassInterface
 {
-    public function process(ContainerBuilder $container)
+    public const WIN32SERVICE_RUNNER_TAG = 'win32service.runner';
+
+    public function process(ContainerBuilder $container): void
     {
         if (!$container->has(RunnerManager::class)) {
             return;
@@ -22,17 +26,22 @@ class TagRunnerCompilerPass implements CompilerPassInterface
         $definition = $container->findDefinition(RunnerManager::class);
 
         // find all service IDs with the app.mail_transport tag
-        $taggedServices = $container->findTaggedServiceIds('win32service.runner');
+        $taggedServices = $container->findTaggedServiceIds(self::WIN32SERVICE_RUNNER_TAG);
 
         foreach ($taggedServices as $id => $tags) {
-
+            $serviceDefinition = $container->findDefinition($id);
+            $class = $serviceDefinition->getClass();
+            $alias = null;
+            if (method_exists($class, 'getAlias')) {
+                $alias = $class::getAlias();
+            }
             // a service could have the same tag twice
             foreach ($tags as $attributes) {
                 $definition->addMethodCall('addRunner', [
                     new Reference($id),
-                    $attributes["alias"]
+                    $alias ?? $attributes['alias'],
                 ]);
             }
         }
     }
-}
+}

lib/DependencyInjection/Win32ServiceExtension.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copy Win32Service (c) 2019
  * Added by : macintoshplus at 19/02/19 13:35
@@ -6,58 +8,135 @@
 
 namespace Win32ServiceBundle\DependencyInjection;
 
-
+use Symfony\Component\Config\FileLocator;
+use Symfony\Component\DependencyInjection\Argument\AbstractArgument;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\ContainerInterface;
 use Symfony\Component\DependencyInjection\Definition;
 use Symfony\Component\DependencyInjection\Extension\Extension;
-use Symfony\Component\Config\FileLocator;
 use Symfony\Component\DependencyInjection\Loader\YamlFileLoader;
+use Symfony\Component\DependencyInjection\Reference;
 use Win32ServiceBundle\Logger\ThreadNumberEvent;
 use Win32ServiceBundle\Logger\ThreadNumberProcessor;
+use Win32ServiceBundle\Model\MessengerServiceRunner;
 
 class Win32ServiceExtension extends Extension
 {
-    public function load(array $configs, ContainerBuilder $container)
+    public function load(array $configs, ContainerBuilder $container): void
     {
-
         $configuration = new Configuration();
 
         $config = $this->processConfiguration($configuration, $configs);
+        $config = $this->processMessengerConfig($config);
         $container->setParameter('win32service.config', $config);
 
-        $loader = new YamlFileLoader($container, new FileLocator(dirname(__DIR__).'/Resources/config'));
+        $this->processMessenger($config['messenger'], $config['project_code'], $container);
+
+        $loader = new YamlFileLoader($container, new FileLocator(\dirname(__DIR__).'/Resources/config'));
         $loader->load('services.yaml');
 
         if (!$config['logging_extra']['enable']) {
             return;
         }
-        //New service definition
+        // New service definition
         $definition = new Definition(ThreadNumberProcessor::class);
-        //If already register, get the current definition
+        // If already register, get the current definition
         if ($container->hasDefinition(ThreadNumberProcessor::class)) {
             $definition = $container->findDefinition(ThreadNumberProcessor::class);
         }
-        //If no definition, add the definition into container
+        // If no definition, add the definition into container
         if (!$container->hasDefinition(ThreadNumberProcessor::class)) {
             $container->setDefinition(ThreadNumberProcessor::class, $definition);
         }
 
-        //Add the tag for receive the thread number event
+        // Add the tag for receive the thread number event
         if (!$definition->hasTag('kernel.event_listener')) {
-            $definition->addTag('kernel.event_listener', ['event'=>ThreadNumberEvent::NAME, 'method'=>'setThreadNumber']);
+            $definition->addTag(
+                'kernel.event_listener',
+                ['event' => ThreadNumberEvent::NAME, 'method' => 'setThreadNumber']
+            );
         }
 
-        //Add tags for each channel defined
+        // Add tags for each channel defined
         $channels = $config['logging_extra']['channels'];
-        if (count($channels)>0) {
+        if (\count($channels) > 0) {
             foreach ($channels as $channel) {
                 $definition->addTag('monolog.processor', ['channel' => $channel]);
             }
+
             return;
         }
 
-        //If no channels defined, the processor is enable for all
+        // If no channels defined, the processor is enable for all
         $definition->addTag('monolog.processor');
+    }
 
+    public function processMessenger(array $messengerConfig, string $projectCode, ContainerBuilder $container): void
+    {
+        foreach ($messengerConfig as $service) {
+            $name = sprintf(
+                MessengerServiceRunner::SERVICE_TAG_PATTERN,
+                $projectCode,
+                implode('_', $service['receivers'])
+            );
+            $arguments = [
+                $service,
+                new AbstractArgument('Routable message bus'),
+                new Reference('messenger.receiver_locator'),
+                new Reference('event_dispatcher'),
+                new Reference('messenger.bus.default'),
+                new Reference('logger', ContainerInterface::NULL_ON_INVALID_REFERENCE),
+                [],
+                new Reference('messenger.listener.reset_services', ContainerInterface::NULL_ON_INVALID_REFERENCE),
+                [],
+            ];
+            $serviceDefinition = new Definition(MessengerServiceRunner::class, $arguments);
+            $serviceDefinition->addTag(TagRunnerCompilerPass::WIN32SERVICE_RUNNER_TAG, ['alias' => $name]);
+            $serviceDefinition->addTag(TagRunnerCompilerPass::WIN32SERVICE_RUNNER_TAG.'.messenger');
+            $container->setDefinition($name, $serviceDefinition);
+        }
+    }
+
+    private function processMessengerConfig(array $config): array
+    {
+        foreach ($config['messenger'] as $service) {
+            $strlen = \strlen((string) ($service['thread_count'] - 1)) - 2;
+            $templatedName = sprintf(
+                MessengerServiceRunner::SERVICE_TAG_PATTERN,
+                $config['project_code'],
+                implode('_', $service['receivers'])
+            );
+            if (($totalLength = $strlen + \strlen($templatedName)) > 80) {
+                throw new \Win32ServiceException(sprintf('The future service identity length "%s" is over 80 chars (%d). Reduce the project code or "receivers" number or name length to keep less than 80 chars.', sprintf($templatedName, $service['thread_count'] - 1), $totalLength));
+            }
+            $config['services'][] = [
+                'machine' => $service['machine'],
+                'displayed_name' => $service['displayed_name'],
+                'description' => $service['description'],
+                'delayed_start' => $service['delayed_start'],
+                'user' => $service['user'],
+                'thread_count' => $service['thread_count'],
+                'script_path' => null,
+                'script_params' => '',
+                'service_id' => $templatedName,
+                'recovery' => [
+                    'enable' => true,
+                    'delay' => 60_000,
+                    'action1' => WIN32_SC_ACTION_RESTART,
+                    'action2' => WIN32_SC_ACTION_RESTART,
+                    'action3' => WIN32_SC_ACTION_RESTART,
+                    'reboot_msg' => '',
+                    'command' => '',
+                    'reset_period' => 1,
+                ],
+                'exit' => [
+                    'graceful' => false,
+                    'code' => 1,
+                ],
+                'dependencies' => [],
+            ];
+        }
+
+        return $config;
     }
 }

lib/Event/MessengerWorkerMessageFailedEvent.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Event;
+
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Event\AbstractWorkerMessageEvent;
+
+final class MessengerWorkerMessageFailedEvent extends AbstractWorkerMessageEvent
+{
+    private \Throwable $throwable;
+    private bool $willRetry = false;
+
+    public function __construct(Envelope $envelope, string $receiverName, \Throwable $error)
+    {
+        $this->throwable = $error;
+
+        parent::__construct($envelope, $receiverName);
+    }
+
+    public function getThrowable(): \Throwable
+    {
+        return $this->throwable;
+    }
+
+    public function willRetry(): bool
+    {
+        return $this->willRetry;
+    }
+
+    public function setForRetry(): void
+    {
+        $this->willRetry = true;
+    }
+}

lib/Event/MessengerWorkerMessageHandledEvent.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Event;
+
+use Symfony\Component\Messenger\Event\AbstractWorkerMessageEvent;
+
+final class MessengerWorkerMessageHandledEvent extends AbstractWorkerMessageEvent
+{
+}

lib/Event/MessengerWorkerRunningEvent.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Event;
+
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+
+final class MessengerWorkerRunningEvent
+{
+    public function __construct(private MessengerServiceRunner $messengerServiceRunner, private bool $isWorkerIdle)
+    {
+    }
+
+    public function getMessengerServiceRunner(): MessengerServiceRunner
+    {
+        return $this->messengerServiceRunner;
+    }
+
+    public function isWorkerIdle(): bool
+    {
+        return $this->isWorkerIdle;
+    }
+}

lib/Event/MessengerWorkerStartedEvent.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Event;
+
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+
+final class MessengerWorkerStartedEvent
+{
+    public function __construct(
+        private MessengerServiceRunner $messengerServiceRunner
+    ) {
+    }
+
+    public function getMessengerServiceRunner(): MessengerServiceRunner
+    {
+        return $this->messengerServiceRunner;
+    }
+}

lib/Event/MessengerWorkerStoppedEvent.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Event;
+
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+
+final class MessengerWorkerStoppedEvent
+{
+    private MessengerServiceRunner $messengerServiceRunner;
+
+    public function __construct(MessengerServiceRunner $messengerServiceRunner)
+    {
+        $this->messengerServiceRunner = $messengerServiceRunner;
+    }
+
+    public function getMessengerServiceRunner(): MessengerServiceRunner
+    {
+        return $this->messengerServiceRunner;
+    }
+}

lib/Logger/ThreadNumberEvent.php

@@ -6,26 +6,16 @@
 
 namespace Win32ServiceBundle\Logger;
 
-use Symfony\Component\EventDispatcher\Event;
+use Symfony\Contracts\EventDispatcher\Event;
 
 class ThreadNumberEvent extends Event
 {
     public const NAME = 'win32service.thread_number';
 
-    private $threadNumber;
-
-    /**
-     * ThreadNumberEvent constructor.
-     * @param int $threadNumber
-     */
-    public function __construct(int $threadNumber)
+    public function __construct(private int $threadNumber)
     {
-        $this->threadNumber = $threadNumber;
     }
 
-    /**
-     * @return int
-     */
     public function getThreadNumber(): int
     {
         return $this->threadNumber;

lib/Logger/ThreadNumberProcessor.php

@@ -7,13 +7,14 @@
 namespace Win32ServiceBundle\Logger;
 
 
-use Symfony\Component\EventDispatcher\Event;
+use Symfony\Contracts\EventDispatcher\Event;
+use Monolog\LogRecord;
 
 class ThreadNumberProcessor
 {
-    private $threadNumber;
+    private ?int $threadNumber = null;
 
-    public function setThreadNumber(Event $evt) {
+    public function setThreadNumber(Event $evt): void {
         if (!$evt instanceof ThreadNumberEvent) {
             return;
         }
@@ -21,8 +22,12 @@ class ThreadNumberProcessor
     }
 
 
-    public function __invoke(array $record)
+    public function __invoke(array|LogRecord $record): array|LogRecord
     {
+        if ($record instanceof LogRecord) {
+            $record->extra['threadNumber'] = $this->threadNumber;
+            return $record;
+        }
         $record['extra']['threadNumber'] = $this->threadNumber;
         return $record;
     }

lib/MessengerSubscriber/AddErrorDetailsStampListener.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\MessengerSubscriber;
+
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Stamp\ErrorDetailsStamp;
+use Win32ServiceBundle\Event\MessengerWorkerMessageFailedEvent;
+
+final class AddErrorDetailsStampListener implements EventSubscriberInterface
+{
+    public function onMessageFailed(MessengerWorkerMessageFailedEvent $event): void
+    {
+        $stamp = ErrorDetailsStamp::create($event->getThrowable());
+        $previousStamp = $event->getEnvelope()->last(ErrorDetailsStamp::class);
+
+        // Do not append duplicate information
+        if ($previousStamp === null || !$previousStamp->equals($stamp)) {
+            $event->addStamps($stamp);
+        }
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            // must have higher priority than SendFailedMessageForRetryListener
+            MessengerWorkerMessageFailedEvent::class => ['onMessageFailed', 200],
+        ];
+    }
+}

lib/MessengerSubscriber/ResetServicesListener.php

@@ -0,0 +1,43 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\MessengerSubscriber;
+
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
+use Win32ServiceBundle\Event\MessengerWorkerRunningEvent;
+use Win32ServiceBundle\Event\MessengerWorkerStoppedEvent;
+
+/**
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ */
+class ResetServicesListener implements EventSubscriberInterface
+{
+    private ServicesResetter $servicesResetter;
+
+    public function __construct(ServicesResetter $servicesResetter)
+    {
+        $this->servicesResetter = $servicesResetter;
+    }
+
+    public function resetServices(MessengerWorkerRunningEvent $event): void
+    {
+        if (!$event->isWorkerIdle()) {
+            $this->servicesResetter->reset();
+        }
+    }
+
+    public function resetServicesAtStop(MessengerWorkerStoppedEvent $event): void
+    {
+        $this->servicesResetter->reset();
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            MessengerWorkerRunningEvent::class => ['resetServices', -1024],
+            MessengerWorkerStoppedEvent::class => ['resetServicesAtStop', -1024],
+        ];
+    }
+}

lib/MessengerSubscriber/SendFailedMessageForRetryListener.php

@@ -0,0 +1,163 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\MessengerSubscriber;
+
+use Psr\Container\ContainerInterface;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Event\WorkerMessageRetriedEvent;
+use Symfony\Component\Messenger\Exception\HandlerFailedException;
+use Symfony\Component\Messenger\Exception\RecoverableExceptionInterface;
+use Symfony\Component\Messenger\Exception\RuntimeException;
+use Symfony\Component\Messenger\Exception\UnrecoverableExceptionInterface;
+use Symfony\Component\Messenger\Retry\RetryStrategyInterface;
+use Symfony\Component\Messenger\Stamp\DelayStamp;
+use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
+use Symfony\Component\Messenger\Stamp\StampInterface;
+use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
+use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
+use Win32ServiceBundle\Event\MessengerWorkerMessageFailedEvent;
+
+class SendFailedMessageForRetryListener implements EventSubscriberInterface
+{
+    private ContainerInterface $sendersLocator;
+    private ContainerInterface $retryStrategyLocator;
+    private ?LoggerInterface $logger;
+    private ?EventDispatcherInterface $eventDispatcher;
+    private int $historySize;
+
+    public function __construct(ContainerInterface $sendersLocator, ContainerInterface $retryStrategyLocator, ?LoggerInterface $logger = null, ?EventDispatcherInterface $eventDispatcher = null, int $historySize = 10)
+    {
+        $this->sendersLocator = $sendersLocator;
+        $this->retryStrategyLocator = $retryStrategyLocator;
+        $this->logger = $logger;
+        $this->eventDispatcher = $eventDispatcher;
+        $this->historySize = $historySize;
+    }
+
+    public function onMessageFailed(MessengerWorkerMessageFailedEvent $event): void
+    {
+        $retryStrategy = $this->getRetryStrategyForTransport($event->getReceiverName());
+        $envelope = $event->getEnvelope();
+        $throwable = $event->getThrowable();
+
+        $message = $envelope->getMessage();
+        $context = [
+            'class' => $message::class,
+        ];
+
+        $shouldRetry = $retryStrategy && $this->shouldRetry($throwable, $envelope, $retryStrategy);
+
+        $retryCount = RedeliveryStamp::getRetryCountFromEnvelope($envelope);
+        if ($shouldRetry) {
+            $event->setForRetry();
+
+            ++$retryCount;
+
+            $delay = $retryStrategy->getWaitingTime($envelope, $throwable);
+
+            if ($this->logger !== null) {
+                $this->logger->warning('Error thrown while handling message {class}. Sending for retry #{retryCount} using {delay} ms delay. Error: "{error}"', $context + ['retryCount' => $retryCount, 'delay' => $delay, 'error' => $throwable->getMessage(), 'exception' => $throwable]);
+            }
+
+            // add the delay and retry stamp info
+            $retryEnvelope = $this->withLimitedHistory($envelope, new DelayStamp($delay), new RedeliveryStamp($retryCount));
+
+            // re-send the message for retry
+            $this->getSenderForTransport($event->getReceiverName())->send($retryEnvelope);
+
+            if ($this->eventDispatcher !== null) {
+                $this->eventDispatcher->dispatch(new WorkerMessageRetriedEvent($retryEnvelope, $event->getReceiverName()));
+            }
+        } else {
+            if ($this->logger !== null) {
+                $this->logger->critical('Error thrown while handling message {class}. Removing from transport after {retryCount} retries. Error: "{error}"', $context + ['retryCount' => $retryCount, 'error' => $throwable->getMessage(), 'exception' => $throwable]);
+            }
+        }
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            // must have higher priority than SendFailedMessageToFailureTransportListener
+            MessengerWorkerMessageFailedEvent::class => ['onMessageFailed', 100],
+        ];
+    }
+
+    /**
+     * Adds stamps to the envelope by keeping only the First + Last N stamps.
+     */
+    private function withLimitedHistory(Envelope $envelope, StampInterface ...$stamps): Envelope
+    {
+        foreach ($stamps as $stamp) {
+            $history = $envelope->all($stamp::class);
+            if (\count($history) < $this->historySize) {
+                $envelope = $envelope->with($stamp);
+                continue;
+            }
+
+            $history = array_merge(
+                [$history[0]],
+                \array_slice($history, -$this->historySize + 2),
+                [$stamp]
+            );
+
+            $envelope = $envelope->withoutAll($stamp::class)->with(...$history);
+        }
+
+        return $envelope;
+    }
+
+    private function shouldRetry(\Throwable $e, Envelope $envelope, RetryStrategyInterface $retryStrategy): bool
+    {
+        if ($e instanceof RecoverableExceptionInterface) {
+            return true;
+        }
+
+        // if one or more nested Exceptions is an instance of RecoverableExceptionInterface we should retry
+        // if ALL nested Exceptions are an instance of UnrecoverableExceptionInterface we should not retry
+        if ($e instanceof HandlerFailedException) {
+            $shouldNotRetry = true;
+            foreach ($e->getNestedExceptions() as $nestedException) {
+                if ($nestedException instanceof RecoverableExceptionInterface) {
+                    return true;
+                }
+
+                if (!$nestedException instanceof UnrecoverableExceptionInterface) {
+                    $shouldNotRetry = false;
+                    break;
+                }
+            }
+            if ($shouldNotRetry) {
+                return false;
+            }
+        }
+
+        if ($e instanceof UnrecoverableExceptionInterface) {
+            return false;
+        }
+
+        return $retryStrategy->isRetryable($envelope, $e);
+    }
+
+    private function getRetryStrategyForTransport(string $alias): ?RetryStrategyInterface
+    {
+        if ($this->retryStrategyLocator->has($alias)) {
+            return $this->retryStrategyLocator->get($alias);
+        }
+
+        return null;
+    }
+
+    private function getSenderForTransport(string $alias): SenderInterface
+    {
+        if ($this->sendersLocator->has($alias)) {
+            return $this->sendersLocator->get($alias);
+        }
+
+        throw new RuntimeException(\sprintf('Could not find sender "%s" based on the same receiver to send the failed message to for retry.', $alias));
+    }
+}

lib/MessengerSubscriber/SendFailedMessageToFailureTransportListener.php

@@ -0,0 +1,70 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\MessengerSubscriber;
+
+use Psr\Container\ContainerInterface;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Stamp\DelayStamp;
+use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
+use Symfony\Component\Messenger\Stamp\SentToFailureTransportStamp;
+use Win32ServiceBundle\Event\MessengerWorkerMessageFailedEvent;
+
+class SendFailedMessageToFailureTransportListener implements EventSubscriberInterface
+{
+    private ContainerInterface $failureSenders;
+    private ?LoggerInterface $logger;
+
+    public function __construct(ContainerInterface $failureSenders, ?LoggerInterface $logger = null)
+    {
+        $this->failureSenders = $failureSenders;
+        $this->logger = $logger;
+    }
+
+    public function onMessageFailed(MessengerWorkerMessageFailedEvent $event): void
+    {
+        if ($event->willRetry()) {
+            return;
+        }
+
+        if (!$this->failureSenders->has($event->getReceiverName())) {
+            return;
+        }
+
+        $failureSender = $this->failureSenders->get($event->getReceiverName());
+        if ($failureSender === null) {
+            return;
+        }
+
+        $envelope = $event->getEnvelope();
+
+        // avoid re-sending to the failed sender
+        if ($envelope->last(SentToFailureTransportStamp::class) !== null) {
+            return;
+        }
+
+        $envelope = $envelope->with(
+            new SentToFailureTransportStamp($event->getReceiverName()),
+            new DelayStamp(0),
+            new RedeliveryStamp(0)
+        );
+
+        if ($this->logger !== null) {
+            $this->logger->info('Rejected message {class} will be sent to the failure transport {transport}.', [
+                'class' => \get_class($envelope->getMessage()),
+                'transport' => $failureSender::class,
+            ]);
+        }
+
+        $failureSender->send($envelope);
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            MessengerWorkerMessageFailedEvent::class => ['onMessageFailed', -100],
+        ];
+    }
+}

lib/MessengerSubscriber/StopWorkerOnFailureLimitListener.php

@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\MessengerSubscriber;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Exception\InvalidArgumentException;
+use Win32ServiceBundle\Event\MessengerWorkerMessageFailedEvent;
+use Win32ServiceBundle\Event\MessengerWorkerRunningEvent;
+
+final class StopWorkerOnFailureLimitListener implements EventSubscriberInterface
+{
+    private int $maximumNumberOfFailures;
+    private ?LoggerInterface $logger;
+    private int $failedMessages = 0;
+
+    public function __construct(int $maximumNumberOfFailures, ?LoggerInterface $logger = null)
+    {
+        $this->maximumNumberOfFailures = $maximumNumberOfFailures;
+        $this->logger = $logger;
+
+        if ($maximumNumberOfFailures <= 0) {
+            throw new InvalidArgumentException('Failure limit must be greater than zero.');
+        }
+    }
+
+    public function onMessageFailed(MessengerWorkerMessageFailedEvent $event): void
+    {
+        ++$this->failedMessages;
+    }
+
+    public function onWorkerRunning(MessengerWorkerRunningEvent $event): void
+    {
+        if (!$event->isWorkerIdle() && $this->failedMessages >= $this->maximumNumberOfFailures) {
+            $this->failedMessages = 0;
+            $event->getMessengerServiceRunner()->stop();
+
+            if ($this->logger !== null) {
+                $this->logger->info('Worker stopped due to limit of {count} failed message(s) is reached', ['count' => $this->maximumNumberOfFailures]);
+            }
+        }
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            MessengerWorkerMessageFailedEvent::class => 'onMessageFailed',
+            MessengerWorkerRunningEvent::class => 'onWorkerRunning',
+        ];
+    }
+}

lib/MessengerSubscriber/StopWorkerOnMemoryLimitListener.php

@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\MessengerSubscriber;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Win32ServiceBundle\Event\MessengerWorkerRunningEvent;
+
+final class StopWorkerOnMemoryLimitListener implements EventSubscriberInterface
+{
+    private int $memoryLimit;
+    private ?LoggerInterface $logger;
+    private $memoryResolver;
+
+    public function __construct(int $memoryLimit, ?LoggerInterface $logger = null, ?callable $memoryResolver = null)
+    {
+        $this->memoryLimit = $memoryLimit;
+        $this->logger = $logger;
+        $this->memoryResolver = $memoryResolver ?: static function () {
+            return memory_get_usage(true);
+        };
+    }
+
+    public function onWorkerRunning(MessengerWorkerRunningEvent $event): void
+    {
+        $memoryResolver = $this->memoryResolver;
+        $usedMemory = $memoryResolver();
+        if ($usedMemory > $this->memoryLimit) {
+            $event->getMessengerServiceRunner()->stop();
+            if ($this->logger !== null) {
+                $this->logger->info('Worker stopped due to memory limit of {limit} bytes exceeded ({memory} bytes used)', ['limit' => $this->memoryLimit, 'memory' => $usedMemory]);
+            }
+        }
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            MessengerWorkerRunningEvent::class => 'onWorkerRunning',
+        ];
+    }
+}

lib/MessengerSubscriber/StopWorkerOnMessageLimitListener.php

@@ -0,0 +1,46 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\MessengerSubscriber;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Exception\InvalidArgumentException;
+use Win32ServiceBundle\Event\MessengerWorkerRunningEvent;
+
+final class StopWorkerOnMessageLimitListener implements EventSubscriberInterface
+{
+    private int $maximumNumberOfMessages;
+    private ?LoggerInterface $logger;
+    private int $receivedMessages = 0;
+
+    public function __construct(int $maximumNumberOfMessages, ?LoggerInterface $logger = null)
+    {
+        $this->maximumNumberOfMessages = $maximumNumberOfMessages;
+        $this->logger = $logger;
+
+        if ($maximumNumberOfMessages <= 0) {
+            throw new InvalidArgumentException('Message limit must be greater than zero.');
+        }
+    }
+
+    public function onWorkerRunning(MessengerWorkerRunningEvent $event): void
+    {
+        if (!$event->isWorkerIdle() && ++$this->receivedMessages >= $this->maximumNumberOfMessages) {
+            $this->receivedMessages = 0;
+            $event->getMessengerServiceRunner()->stop();
+
+            if ($this->logger !== null) {
+                $this->logger->info('Worker stopped due to maximum count of {count} messages processed', ['count' => $this->maximumNumberOfMessages]);
+            }
+        }
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            MessengerWorkerRunningEvent::class => 'onWorkerRunning',
+        ];
+    }
+}

lib/MessengerSubscriber/StopWorkerOnTimeLimitListener.php

@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\MessengerSubscriber;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Exception\InvalidArgumentException;
+use Win32ServiceBundle\Event\MessengerWorkerRunningEvent;
+use Win32ServiceBundle\Event\MessengerWorkerStartedEvent;
+
+final class StopWorkerOnTimeLimitListener implements EventSubscriberInterface
+{
+    private int $timeLimitInSeconds;
+    private ?LoggerInterface $logger;
+    private $endTime;
+
+    public function __construct(int $timeLimitInSeconds, ?LoggerInterface $logger = null)
+    {
+        $this->timeLimitInSeconds = $timeLimitInSeconds;
+        $this->logger = $logger;
+
+        if ($timeLimitInSeconds <= 0) {
+            throw new InvalidArgumentException('Time limit must be greater than zero.');
+        }
+    }
+
+    public function onWorkerStarted(): void
+    {
+        $startTime = microtime(true);
+        $this->endTime = $startTime + $this->timeLimitInSeconds;
+    }
+
+    public function onWorkerRunning(MessengerWorkerRunningEvent $event): void
+    {
+        if ($this->endTime < microtime(true)) {
+            $event->getMessengerServiceRunner()->stop();
+            if ($this->logger !== null) {
+                $this->logger->info('Worker stopped due to time limit of {timeLimit}s exceeded', ['timeLimit' => $this->timeLimitInSeconds]);
+            }
+        }
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            MessengerWorkerStartedEvent::class => 'onWorkerStarted',
+            MessengerWorkerRunningEvent::class => 'onWorkerRunning',
+        ];
+    }
+}

lib/Model/MessengerServiceRunner.php

@@ -0,0 +1,314 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Model;
+
+use Psr\Container\ContainerInterface;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Console\Exception\RuntimeException;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
+use Symfony\Component\Messenger\Exception\HandlerFailedException;
+use Symfony\Component\Messenger\Exception\RejectRedeliveredMessageException;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Messenger\RoutableMessageBus;
+use Symfony\Component\Messenger\Stamp\AckStamp;
+use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
+use Symfony\Component\Messenger\Stamp\FlushBatchHandlersStamp;
+use Symfony\Component\Messenger\Stamp\NoAutoAckStamp;
+use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
+use Win32Service\Exception\RecoveryActionException;
+use Win32Service\Model\AbstractServiceRunner;
+use Win32ServiceBundle\Event\MessengerWorkerMessageFailedEvent;
+use Win32ServiceBundle\Event\MessengerWorkerMessageHandledEvent;
+use Win32ServiceBundle\Event\MessengerWorkerRunningEvent;
+use Win32ServiceBundle\Event\MessengerWorkerStartedEvent;
+use Win32ServiceBundle\Event\MessengerWorkerStoppedEvent;
+use Win32ServiceBundle\MessengerSubscriber\ResetServicesListener;
+use Win32ServiceBundle\MessengerSubscriber\StopWorkerOnFailureLimitListener;
+use Win32ServiceBundle\MessengerSubscriber\StopWorkerOnMemoryLimitListener;
+use Win32ServiceBundle\MessengerSubscriber\StopWorkerOnMessageLimitListener;
+use Win32ServiceBundle\MessengerSubscriber\StopWorkerOnTimeLimitListener;
+
+final class MessengerServiceRunner extends AbstractServiceRunner
+{
+    public const SERVICE_TAG_PATTERN = 'win32service.%s.messenger.%s.%%s';
+    /**
+     * @var array<string, ReceiverInterface>
+     */
+    private array $receivers;
+    private bool $shouldStop = false;
+
+    private array $acks = [];
+    private \SplObjectStorage $unacks;
+
+    public function __construct(
+        private array $config,
+        private RoutableMessageBus $routableBus,
+        private ContainerInterface $receiverLocator,
+        private EventDispatcherInterface $eventDispatcher,
+        private MessageBusInterface $bus,
+        private ?LoggerInterface $logger = null,
+        private array $receiverNames = [],
+        private ?ResetServicesListener $resetServicesListener = null,
+        private array $busIds = [],
+    ) {
+        $this->unacks = new \SplObjectStorage();
+    }
+
+    protected function setup(): void
+    {
+        $this->config['sleep'] = $this->config['sleep'] ?? 1000000;
+
+        $this->eventDispatcher->addSubscriber($this->resetServicesListener);
+        $limit = (int) $this->config['limit'];
+        if ($limit > 0) {
+            $this->eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener($limit, $this->logger));
+        }
+        $failureLimit = (int) $this->config['failure_limit'];
+        if ($failureLimit > 0) {
+            $this->eventDispatcher->addSubscriber(new StopWorkerOnFailureLimitListener($failureLimit, $this->logger));
+        }
+        $timeLimit = (int) $this->config['time_limit'];
+        if ($timeLimit > 0) {
+            $this->eventDispatcher->addSubscriber(new StopWorkerOnTimeLimitListener($timeLimit, $this->logger));
+        }
+        $memoryLimit = (string) $this->config['memory_limit'];
+        if ($memoryLimit > 0) {
+            $this->eventDispatcher->addSubscriber(new StopWorkerOnMemoryLimitListener(
+                $this->convertToBytes($memoryLimit),
+                $this->logger
+            ));
+        }
+
+        $this->receivers = [];
+        foreach ($this->config['receivers'] as $receiverName) {
+            if (!$this->receiverLocator->has($receiverName)) {
+                $message = \sprintf('The receiver "%s" does not exist.', $receiverName);
+                if ($this->receiverNames) {
+                    $message .= \sprintf(' Valid receivers are: %s.', implode(', ', $this->receiverNames));
+                }
+
+                throw new RuntimeException($message);
+            }
+
+            $this->receivers[$receiverName] = $this->receiverLocator->get($receiverName);
+        }
+    }
+
+    public function stop(): void
+    {
+        if ($this->logger !== null) {
+            $this->logger->info('Stopping worker.', ['transport_names' => array_keys($this->receivers)]);
+        }
+
+        $this->shouldStop = true;
+        $this->requestStop();
+
+        throw new RecoveryActionException('Restart requested');
+    }
+
+    protected function beforeContinue(): void
+    {
+        // TODO: Implement beforeContinue() method.
+    }
+
+    protected function beforePause(): void
+    {
+        // TODO: Implement beforePause() method.
+    }
+
+    protected function run(int $control): void
+    {
+        $this->eventDispatcher->dispatch(new MessengerWorkerStartedEvent($this));
+
+        $envelopeHandled = false;
+        $envelopeHandledStart = microtime(true);
+        foreach ($this->receivers as $transportName => $receiver) {
+            $envelopes = $receiver->get();
+
+            foreach ($envelopes as $envelope) {
+                $envelopeHandled = true;
+
+                $this->handleMessage($envelope, $transportName);
+                $this->eventDispatcher->dispatch(new MessengerWorkerRunningEvent($this, false));
+
+                if ($this->shouldStop) {
+                    break 2;
+                }
+            }
+
+            if ($envelopeHandled) {
+                break;
+            }
+        }
+        if (!$envelopeHandled && $this->flush(false)) {
+            return;
+        }
+
+        if (!$envelopeHandled) {
+            $this->eventDispatcher->dispatch(new MessengerWorkerRunningEvent($this, true));
+
+            if (0 < $sleep = (int) ($this->config['sleep'] - 1e6 * (microtime(true) - $envelopeHandledStart))) {
+                usleep($sleep);
+            }
+        }
+
+        $this->eventDispatcher->dispatch(new MessengerWorkerStoppedEvent($this));
+    }
+
+    protected function lastRunIsTooSlow(float $duration): void
+    {
+        $this->logger->info('Last run is too low. Max 30s.', ['duration' => $duration]);
+    }
+
+    protected function beforeStop(): void
+    {
+        // TODO: Implement beforeStop() method.
+    }
+
+    private function handleMessage(Envelope $envelope, string $transportName): void
+    {
+        $event = new WorkerMessageReceivedEvent($envelope, $transportName);
+        $this->eventDispatcher->dispatch($event);
+        $envelope = $event->getEnvelope();
+
+        if (!$event->shouldHandle()) {
+            return;
+        }
+
+        $acked = false;
+        $ack = function (Envelope $envelope, ?\Throwable $e = null) use ($transportName, &$acked) {
+            $acked = true;
+            $this->acks[] = [$transportName, $envelope, $e];
+        };
+
+        try {
+            $e = null;
+            $envelope = $this->bus->dispatch($envelope->with(
+                new ReceivedStamp($transportName),
+                new ConsumedByWorkerStamp(),
+                new AckStamp($ack)
+            ));
+        } catch (\Throwable $e) {
+        }
+
+        $noAutoAckStamp = $envelope->last(NoAutoAckStamp::class);
+
+        if (!$acked && !$noAutoAckStamp) {
+            $this->acks[] = [$transportName, $envelope, $e];
+        } elseif ($noAutoAckStamp) {
+            $this->unacks[$noAutoAckStamp->getHandlerDescriptor()->getBatchHandler()] = [
+                $envelope->withoutAll(AckStamp::class),
+                $transportName,
+            ];
+        }
+
+        $this->ack();
+    }
+
+    private function ack(): bool
+    {
+        $acks = $this->acks;
+        $this->acks = [];
+
+        foreach ($acks as [$transportName, $envelope, $e]) {
+            $receiver = $this->receivers[$transportName];
+
+            if ($e !== null) {
+                if ($rejectFirst = $e instanceof RejectRedeliveredMessageException) {
+                    // redelivered messages are rejected first so that continuous failures in an event listener or while
+                    // publishing for retry does not cause infinite redelivery loops
+                    $receiver->reject($envelope);
+                }
+
+                if ($e instanceof HandlerFailedException) {
+                    $envelope = $e->getEnvelope();
+                }
+
+                $failedEvent = new MessengerWorkerMessageFailedEvent($envelope, $transportName, $e);
+
+                $this->eventDispatcher->dispatch($failedEvent);
+                $envelope = $failedEvent->getEnvelope();
+
+                if (!$rejectFirst) {
+                    $receiver->reject($envelope);
+                }
+
+                continue;
+            }
+
+            $handledEvent = new MessengerWorkerMessageHandledEvent($envelope, $transportName);
+            $this->eventDispatcher->dispatch($handledEvent);
+            $envelope = $handledEvent->getEnvelope();
+
+            if ($this->logger !== null) {
+                $message = $envelope->getMessage();
+                $context = [
+                    'class' => $message::class,
+                ];
+                $this->logger->info('{class} was handled successfully (acknowledging to transport).', $context);
+            }
+
+            $receiver->ack($envelope);
+        }
+
+        return (bool) $acks;
+    }
+
+    private function flush(bool $force): bool
+    {
+        $unacks = $this->unacks;
+
+        if (!$unacks->count()) {
+            return false;
+        }
+
+        $this->unacks = new \SplObjectStorage();
+
+        foreach ($unacks as $batchHandler) {
+            [$envelope, $transportName] = $unacks[$batchHandler];
+            try {
+                $this->bus->dispatch($envelope->with(new FlushBatchHandlersStamp($force)));
+                $envelope = $envelope->withoutAll(NoAutoAckStamp::class);
+                unset($unacks[$batchHandler], $batchHandler);
+            } catch (\Throwable $e) {
+                $this->acks[] = [$transportName, $envelope, $e];
+            }
+        }
+
+        return $this->ack();
+    }
+
+    private function convertToBytes(string $memoryLimit): int
+    {
+        $memoryLimit = strtolower($memoryLimit);
+        $max = ltrim($memoryLimit, '+');
+        if (str_starts_with($max, '0x')) {
+            $max = \intval($max, 16);
+        } elseif (str_starts_with($max, '0')) {
+            $max = \intval($max, 8);
+        } else {
+            $max = (int) $max;
+        }
+
+        switch (substr(rtrim($memoryLimit, 'b'), -1)) {
+            case 't':
+                $max *= 1024;
+                // no break
+            case 'g':
+                $max *= 1024;
+                // no break
+            case 'm':
+                $max *= 1024;
+                // no break
+            case 'k':
+                $max *= 1024;
+        }
+
+        return $max;
+    }
+}

lib/Resources/config/services.yaml

@@ -3,29 +3,33 @@ services:
   _defaults:
     autowire: true      # Automatically injects dependencies in your services.
     autoconfigure: true # Automatically registers your services as commands, event subscribers, etc.
-    bind:
-      $projectRoot: '%kernel.project_dir%'
 
   Win32ServiceBundle\:
     resource: '../../*'
-    exclude: '../../{DependencyInjection,Entity,Logger,Migrations,Model,Tests,Win32ServiceBundle.php}'
+    exclude: '../../{DependencyInjection,Entity,Logger,MessengerSubscriber,Migrations,Model,Tests,Win32ServiceBundle.php}'
+
+  Win32ServiceBundle\Service\ServiceConfigurationManager:
+    arguments:
+      - '%win32service.config%'
+      - '%kernel.environment%'
+
+  Win32ServiceBundle\MessengerSubscriber\AddErrorDetailsStampListener: ~
 
 
-  Win32ServiceBundle\Command\RegisterServiceCommand:
-    autowire: true
-    calls:
-      - ['defineBundleConfig', ['%win32service.config%']]
-  Win32ServiceBundle\Command\UnregisterServiceCommand:
-    autowire: true
-    calls:
-      - ['defineBundleConfig', ['%win32service.config%']]
-  Win32ServiceBundle\Command\ActionServiceCommand:
-    autowire: true
-    calls:
-      - ['defineBundleConfig', ['%win32service.config%']]
-  Win32ServiceBundle\Command\ExecuteServiceCommand:
-    autowire: true
-    calls:
-      - ['defineBundleConfig', ['%win32service.config%']]
-      - ['setService', ['@Win32ServiceBundle\Service\RunnerManager']]
-      - ['setEventDispatcher', ['@event_dispatcher']]
+  Win32ServiceBundle\MessengerSubscriber\SendFailedMessageForRetryListener:
+    arguments:
+      $retryStrategyLocator: '@messenger.retry_strategy_locator'
+      $logger: '@logger'
+      $eventDispatcher: '@event_dispatcher'
+    tags:
+      - { name: 'monolog.logger', channel: 'messenger' }
+
+  Win32ServiceBundle\MessengerSubscriber\SendFailedMessageToFailureTransportListener:
+    arguments:
+      $logger: '@logger'
+    tags:
+      - {name: 'monolog.logger', channel: 'messenger'}
+
+  Win32ServiceBundle\MessengerSubscriber\ResetServicesListener:
+    arguments:
+      - '@services_resetter'

lib/Service/RunnerManager.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copyright Macintoshplus (c) 2019
  * Added by : Macintoshplus at 19/02/19 23:06
@@ -13,26 +15,21 @@ class RunnerManager
     /**
      * @var RunnerServiceInterface[]
      */
-    private $runner;
+    private array $runner = [];
 
-    public function __construct()
+    public function addRunner(RunnerServiceInterface $runner, string $alias): void
     {
-        $this->runner = [];
-    }
-
-    public function addRunner(RunnerServiceInterface $runner, string $alias) {
         $this->runner[$alias] = $runner;
     }
 
-    /**
-     * @param string $alias
-     * @return RunnerServiceInterface|null
-     */
-    public function getRunner(string $alias) {
-        if (!isset($this->runner[$alias])) {
-            return null;
-        }
-
-        return $this->runner[$alias];
+    public function getRunner(string $alias): ?RunnerServiceInterface
+    {
+        return $this->runner[$alias] ?? null;
     }
-}
+
+    /** @return array<string, RunnerServiceInterface> */
+    public function getRunners(): array
+    {
+        return $this->runner;
+    }
+}

lib/Service/ServiceConfigurationManager.php

@@ -0,0 +1,118 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Service;
+
+use Win32Service\Model\ServiceIdentifier;
+use Win32Service\Model\ServiceInformations;
+use Win32ServiceBundle\Command\ExecuteServiceCommand;
+
+final class ServiceConfigurationManager
+{
+    private array $serviceIds = [];
+    private array $serviceIdsToRunnerAlias = [];
+
+    public function __construct(private array $configuration, string $environment)
+    {
+        if ($configuration === []) {
+            throw new \Win32ServiceException('The configuration of win32Service is not defined');
+        }
+
+        $services = $configuration['services'];
+
+        foreach ($services as $service) {
+            $threadNumber = $service['thread_count'];
+            $runnerAlias = $service['service_id'];
+            $runnerName = $service['displayed_name'];
+            $scriptParams = $service['script_params'];
+            $scriptPath = $service['script_path'];
+
+            for ($i = 0; $i < $threadNumber; ++$i) {
+                $serviceThreadId = sprintf($runnerAlias, $i);
+                $runnerNameId = sprintf($runnerName, $i);
+
+                $path = $service['script_path'];
+                $args = sprintf($scriptParams, $i);
+
+                if ($scriptPath === null) {
+                    $path = realpath($_SERVER['PHP_SELF']);
+                    $args = sprintf('-e %s %s %s %d', $environment, ExecuteServiceCommand::getDefaultName(), $serviceThreadId, $i);
+                }
+
+                $service['service_id'] = $serviceThreadId;
+                $service['displayed_name'] = $runnerNameId;
+                $service['script_path'] = $path;
+                $service['script_params'] = $args;
+
+                if (isset($this->serviceIds[$serviceThreadId]) === true) {
+                    throw new \Win32ServiceException(sprintf('The Win32Service "%s" is already defined. if the parameter "thread_count" is greater than 1, please add "%d" in "service_id" parameter. Otherwise, check if no other service have same name.', $serviceThreadId));
+                }
+                $this->serviceIds[$serviceThreadId] = $service;
+                $this->serviceIdsToRunnerAlias[$serviceThreadId] = $runnerAlias;
+            }
+        }
+    }
+
+    /** @return \Generator<int, ServiceInformations> */
+    public function getFullServiceList(): \Generator
+    {
+        foreach ($this->serviceIds as $serviceId => $service) {
+            yield $this->getServiceInformations($serviceId);
+        }
+    }
+
+    public function getServiceRawConfiguration(string $serviceId): array
+    {
+        if (isset($this->serviceIds[$serviceId]) === false) {
+            throw new \Win32ServiceException(sprintf('The Win32Service "%s" is not defined.', $serviceId));
+        }
+
+        return $this->serviceIds[$serviceId];
+    }
+
+    public function getServiceInformations(string $serviceId): ServiceInformations
+    {
+        if (isset($this->serviceIds[$serviceId]) === false) {
+            throw new \Win32ServiceException(sprintf('The Win32Service "%s" is not defined.', $serviceId));
+        }
+        $service = $this->serviceIds[$serviceId];
+        $windowsLocalEncoding = $this->configuration['windows_local_encoding'];
+
+        $serviceInfos = new ServiceInformations(
+            ServiceIdentifier::identify($serviceId, $service['machine']),
+            mb_convert_encoding($service['displayed_name'], $windowsLocalEncoding, 'UTF-8'),
+            mb_convert_encoding($service['description'], $windowsLocalEncoding, 'UTF-8'),
+            mb_convert_encoding($service['script_path'], $windowsLocalEncoding, 'UTF-8'),
+            mb_convert_encoding($service['script_params'], $windowsLocalEncoding, 'UTF-8')
+        );
+
+        $serviceInfos->defineIfStartIsDelayed($service['delayed_start']);
+        $recovery = $service['recovery'];
+        $serviceInfos->defineRecoverySettings(
+            $recovery['delay'],
+            $recovery['enable'],
+            $recovery['action1'],
+            $recovery['action2'],
+            $recovery['action3'],
+            $recovery['reboot_msg'],
+            $recovery['command'],
+            $recovery['reset_period']
+        );
+
+        if ($service['user']['account'] !== null) {
+            $serviceInfos->defineUserService($service['user']['account'], $service['user']['password']);
+        }
+
+        if (\count($service['dependencies']) > 0) {
+            $serviceInfos->defineDependencies($service['dependencies']);
+        }
+
+        return $serviceInfos;
+    }
+
+    public function getRunnerAliasForServiceId(string $serviceId): string
+    {
+        return $this->serviceIdsToRunnerAlias[$serviceId] ?? throw new \Win32ServiceException('The Win32Service "'.$serviceId.'" have no alias defined.');
+    }
+}

lib/Win32ServiceBundle.php

@@ -1,4 +1,6 @@
 <?php
+
+declare(strict_types=1);
 /**
  * @copy Win32Service (c) 2019
  * Added by : macintoshplus at 19/02/19 13:30
@@ -7,12 +9,19 @@
 namespace Win32ServiceBundle;
 
 use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\HttpKernel\Bundle\Bundle;
+use Win32Service\Model\RunnerServiceInterface;
+use Win32ServiceBundle\DependencyInjection\MessengerPass;
 use Win32ServiceBundle\DependencyInjection\TagRunnerCompilerPass;
 
-class Win32ServiceBundle extends \Symfony\Component\HttpKernel\Bundle\Bundle
+class Win32ServiceBundle extends Bundle
 {
-    public function build(ContainerBuilder $container)
+    public function build(ContainerBuilder $container): void
     {
+        $autoconfig = $container->registerForAutoconfiguration(RunnerServiceInterface::class);
+        $autoconfig->addTag(TagRunnerCompilerPass::WIN32SERVICE_RUNNER_TAG);
+
         $container->addCompilerPass(new TagRunnerCompilerPass());
+        $container->addCompilerPass(new MessengerPass());
     }
 }

phpunit.xml.dist

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/9.6/phpunit.xsd"
+         bootstrap="tests/bootstrap.php"
+         forceCoversAnnotation="false"
+         beStrictAboutCoversAnnotation="true"
+         beStrictAboutOutputDuringTests="true"
+         beStrictAboutTodoAnnotatedTests="true"
+         verbose="true">
+  <coverage processUncoveredFiles="true">
+    <include>
+      <directory suffix=".php">lib</directory>
+    </include>
+  </coverage>
+  <testsuite name="default">
+    <directory suffix="Test.php">tests/Unit</directory>
+  </testsuite>
+  <php>
+    <ini name="display_errors" value="1" />
+    <ini name="error_reporting" value="-1" />
+    <server name="APP_ENV" value="test" force="true" />
+    <env name="SYMFONY_DEPRECATIONS_HELPER" value="disabled"/>
+    <env name="KERNEL_CLASS" value="Win32ServiceBundle\Tests\Application\Kernel"/>
+  </php>
+</phpunit>

rector.php

@@ -0,0 +1,46 @@
+<?php
+
+declare(strict_types=1);
+
+use Rector\Config\RectorConfig;
+use Rector\Core\ValueObject\PhpVersion;
+use Rector\Set\ValueObject\SetList;
+use Rector\Symfony\Set\SymfonySetList;
+
+return static function (RectorConfig $rectorConfig): void {
+    $rectorConfig->phpVersion(phpVersion: PhpVersion::PHP_80);
+    $rectorConfig->importNames();
+    $rectorConfig->importShortClasses();
+    $rectorConfig->parallel();
+//    $rectorConfig->symfonyContainerPhp(filePath: __DIR__ . 'var/cache/dev/App_KernelDevDebugContainer.php');
+
+    $rectorConfig->autoloadPaths(autoloadPaths: [
+        __DIR__ . '/vendor/autoload.php',
+    ]);
+
+    $rectorConfig->paths(paths: [
+        __DIR__ . '/lib',
+    ]);
+
+    $rectorConfig->skip(criteria: [
+        __DIR__ . '/vendor',
+    ]);
+
+    $rectorConfig->sets(sets: [
+//        SetList::CODE_QUALITY,
+//        SetList::CODING_STYLE,
+//        SetList::DEAD_CODE,
+//        SetList::EARLY_RETURN,
+        SetList::PHP_80,
+//        SetList::PSR_4,
+//        SetList::PRIVATIZATION,
+//        SetList::TYPE_DECLARATION,
+//        SetList::TYPE_DECLARATION_STRICT,
+//        SymfonySetList::SYMFONY_STRICT,
+//        SymfonySetList::SYMFONY_44,
+//        SymfonySetList::SYMFONY_CODE_QUALITY,
+//        SymfonySetList::SYMFONY_CONSTRUCTOR_INJECTION,
+#SymfonySetList::ANNOTATIONS_TO_ATTRIBUTES,
+    ]);
+
+};

tests/Application/.env

@@ -0,0 +1,32 @@
+# In all environments, the following files are loaded if they exist,
+# the latter taking precedence over the former:
+#
+#  * .env                contains default values for the environment variables needed by the app
+#  * .env.local          uncommitted file with local overrides
+#  * .env.$APP_ENV       committed environment-specific defaults
+#  * .env.$APP_ENV.local uncommitted environment-specific overrides
+#
+# Real environment variables win over .env files.
+#
+# DO NOT DEFINE PRODUCTION SECRETS IN THIS FILE NOR IN ANY OTHER COMMITTED FILES.
+# https://symfony.com/doc/current/configuration/secrets.html
+#
+# Run "composer dump-env prod" to compile .env files for production use (requires symfony/flex >=1.2).
+# https://symfony.com/doc/current/best_practices.html#use-environment-variables-for-infrastructure-configuration
+
+###> symfony/framework-bundle ###
+APP_ENV=dev
+APP_SECRET=625318e78d9582a45c1483b809a1e780
+###< symfony/framework-bundle ###
+
+###> symfony/messenger ###
+# Choose one of the transports below
+# MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
+# MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
+MESSENGER_TRANSPORT_DSN=doctrine://default?auto_setup=0
+###< symfony/messenger ###
+
+#DATABASE_URL="sqlite:///%kernel.project_dir%/var/data.db"
+#DATABASE_URL="mysql://app:!ChangeMe!@127.0.0.1:3306/app?serverVersion=8.0.32&charset=utf8mb4"
+DATABASE_URL="mysql://user:nopassword@127.0.0.1:3306/app?serverVersion=mariadb-10.11.2&charset=utf8mb4"
+#DATABASE_URL="postgresql://app:!ChangeMe!@127.0.0.1:5432/app?serverVersion=16&charset=utf8"

tests/Application/.env.test

@@ -0,0 +1,2 @@
+
+DATABASE_URL="mysql://root:nopassword@127.0.0.1:3306/app?serverVersion=mariadb-10.11.2&charset=utf8mb4"

tests/Application/.gitignore

@@ -0,0 +1,10 @@
+
+###> symfony/framework-bundle ###
+/.env.local
+/.env.local.php
+/.env.*.local
+/config/secrets/prod/prod.decrypt.private.php
+/public/bundles/
+/var/
+/vendor/
+###< symfony/framework-bundle ###

tests/Application/.php-version

@@ -0,0 +1 @@
+8.0

tests/Application/bin/console

@@ -0,0 +1,25 @@
+#!/usr/bin/env php
+<?php
+
+use Win32ServiceBundle\Tests\Application\Kernel;
+use Symfony\Bundle\FrameworkBundle\Console\Application;
+
+if (!is_dir(dirname(__DIR__, 3) . '/vendor')) {
+    throw new LogicException('Dependencies are missing. Try running "composer install".');
+}
+
+if (!is_file(dirname(__DIR__, 3) . '/vendor/autoload_runtime.php')) {
+    throw new LogicException('Symfony Runtime is missing. Try running "composer require symfony/runtime".');
+}
+
+$_SERVER['APP_RUNTIME_OPTIONS'] = [
+    'project_dir' => dirname(__DIR__),
+];
+
+require_once dirname(__DIR__, 3) . '/vendor/autoload_runtime.php';
+
+return function (array $context) {
+    $kernel = new Kernel($context['APP_ENV'], (bool)$context['APP_DEBUG']);
+
+    return new Application($kernel);
+};

tests/Application/compose.override.yaml.dist

@@ -0,0 +1,4 @@
+services:
+  db:
+    ports:
+      - "3306:3306"

tests/Application/compose.yaml

@@ -0,0 +1,13 @@
+services:
+  db:
+    image: mariadb:10.11
+    environment:
+      MYSQL_DATABASE: app
+      MYSQL_ROOT_PASSWORD: nopassword
+      MYSQL_USER: user
+      MYSQL_PASSWORD: nopassword
+    volumes:
+      - win32service_bundle_mariadb_data:/var/lib/mysql
+
+volumes:
+  win32service_bundle_mariadb_data: ~

tests/Application/composer.json

@@ -0,0 +1,4 @@
+{
+    "type": "project",
+    "license": "proprietary"
+}

tests/Application/config/bundles.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+return [
+    Symfony\Bundle\FrameworkBundle\FrameworkBundle::class => ['all' => true],
+    Symfony\Bundle\MonologBundle\MonologBundle::class => ['all' => true],
+    Doctrine\Bundle\DoctrineBundle\DoctrineBundle::class => ['all' => true],
+    Doctrine\Bundle\MigrationsBundle\DoctrineMigrationsBundle::class => ['all' => true],
+    Win32ServiceBundle\Win32ServiceBundle::class => ['all' => true],
+];

tests/Application/config/packages/cache.yaml

@@ -0,0 +1,19 @@
+framework:
+    cache:
+        # Unique name of your app: used to compute stable namespaces for cache keys.
+        #prefix_seed: your_vendor_name/app_name
+
+        # The "app" cache stores to the filesystem by default.
+        # The data in this cache should persist between deploys.
+        # Other options include:
+
+        # Redis
+        #app: cache.adapter.redis
+        #default_redis_provider: redis://localhost
+
+        # APCu (not recommended with heavy random-write workloads as memory fragmentation can cause perf issues)
+        #app: cache.adapter.apcu
+
+        # Namespaced pools use the above "app" backend by default
+        #pools:
+            #my.dedicated.cache: null

tests/Application/config/packages/doctrine.yaml

@@ -0,0 +1,52 @@
+doctrine:
+  dbal:
+    url: '%env(resolve:DATABASE_URL)%'
+
+    # IMPORTANT: You MUST configure your server version,
+    # either here or in the DATABASE_URL env var (see .env file)
+    #server_version: '16'
+
+    profiling_collect_backtrace: '%kernel.debug%'
+    use_savepoints: true
+  orm:
+    auto_generate_proxy_classes: true
+    enable_lazy_ghost_objects: false # set true when minimal version of Symfony is 6.2 + composer req --dev symfony/var-exporter
+    report_fields_where_declared: true
+    validate_xml_mapping: true
+    naming_strategy: doctrine.orm.naming_strategy.underscore_number_aware
+    auto_mapping: true
+    mappings:
+      App:
+        type: attribute
+        is_bundle: false
+        dir: '%kernel.project_dir%/src/Entity'
+        prefix: 'App\Entity'
+        alias: App
+    controller_resolver:
+      auto_mapping: false
+
+when@test:
+  doctrine:
+    dbal:
+      # "TEST_TOKEN" is typically set by ParaTest
+      dbname_suffix: '_test%env(default::TEST_TOKEN)%'
+
+when@prod:
+  doctrine:
+    orm:
+      auto_generate_proxy_classes: false
+      proxy_dir: '%kernel.build_dir%/doctrine/orm/Proxies'
+      query_cache_driver:
+        type: pool
+        pool: doctrine.system_cache_pool
+      result_cache_driver:
+        type: pool
+        pool: doctrine.result_cache_pool
+
+  framework:
+    cache:
+      pools:
+        doctrine.result_cache_pool:
+          adapter: cache.app
+        doctrine.system_cache_pool:
+          adapter: cache.system

tests/Application/config/packages/doctrine_migrations.yaml

@@ -0,0 +1,6 @@
+doctrine_migrations:
+  migrations_paths:
+    # namespace is arbitrary but should be different from App\Migrations
+    # as migrations classes should NOT be autoloaded
+    'DoctrineMigrations': '%kernel.project_dir%/migrations'
+  enable_profiler: false

tests/Application/config/packages/framework.yaml

@@ -0,0 +1,24 @@
+# see https://symfony.com/doc/current/reference/configuration/framework.html
+framework:
+    secret: '%env(APP_SECRET)%'
+    #csrf_protection: true
+    http_method_override: false
+
+    # Enables session support. Note that the session will ONLY be started if you read or write from it.
+    # Remove or comment this section to explicitly disable session support.
+    session:
+        handler_id: null
+        cookie_secure: auto
+        cookie_samesite: lax
+        storage_factory_id: session.storage.factory.native
+
+    #esi: true
+    #fragments: true
+    php_errors:
+        log: true
+
+when@test:
+    framework:
+        test: true
+        session:
+            storage_factory_id: session.storage.factory.mock_file

tests/Application/config/packages/messenger.yaml

@@ -0,0 +1,27 @@
+framework:
+  messenger:
+    # reset services after consuming messages
+    reset_on_message: true
+
+    # Uncomment this (and the failed transport below) to send failed messages to this transport for later handling.
+    failure_transport: failed
+
+    transports:
+      # https://symfony.com/doc/current/messenger.html#transport-configuration
+      async:
+        dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
+        retry_strategy:
+          max_retries: 1
+          delay: 1000
+
+      failed: 'doctrine://default?queue_name=failed'
+      # sync: 'sync://'
+
+    routing:
+    # Route your messages to the transports
+    # 'App\Message\YourMessage': async
+      'Win32ServiceBundle\Tests\Application\Event\TestFailedMessage': async
+      'Win32ServiceBundle\Tests\Application\Event\TestMemoryLimitMessage': async
+      'Win32ServiceBundle\Tests\Application\Event\TestMessage': async
+      'Win32ServiceBundle\Tests\Application\Event\TestRetryMessage': async
+      'Win32ServiceBundle\Tests\Application\Event\TestTimeLimitMessage': async

tests/Application/config/packages/monolog.yaml

@@ -0,0 +1,62 @@
+monolog:
+  channels:
+    - deprecation # Deprecations are logged in the dedicated "deprecation" channel when it exists
+
+when@dev:
+  monolog:
+    handlers:
+      main:
+        type: stream
+        path: "%kernel.logs_dir%/%kernel.environment%.log"
+        level: debug
+        channels: ["!event", "!doctrine"]
+      # uncomment to get logging in your browser
+      # you may have to allow bigger header sizes in your Web server configuration
+      #firephp:
+      #    type: firephp
+      #    level: info
+      #chromephp:
+      #    type: chromephp
+      #    level: info
+      console:
+        type: console
+        process_psr_3_messages: false
+        channels: ["!event", "!doctrine", "!console"]
+
+when@test:
+  monolog:
+    handlers:
+      main:
+        type: fingers_crossed
+        action_level: error
+        handler: nested
+        excluded_http_codes: [404, 405]
+        channels: ["!event"]
+      nested:
+        type: stream
+        path: "%kernel.logs_dir%/%kernel.environment%.log"
+        level: debug
+
+when@prod:
+  monolog:
+    handlers:
+      main:
+        type: fingers_crossed
+        action_level: error
+        handler: nested
+        excluded_http_codes: [404, 405]
+        buffer_size: 50 # How many messages should be saved? Prevent memory leaks
+      nested:
+        type: stream
+        path: php://stderr
+        level: debug
+        formatter: monolog.formatter.json
+      console:
+        type: console
+        process_psr_3_messages: false
+        channels: ["!event", "!doctrine"]
+      deprecation:
+        type: stream
+        channels: [deprecation]
+        path: php://stderr
+        formatter: monolog.formatter.json

tests/Application/config/packages/routing.yaml

@@ -0,0 +1,12 @@
+framework:
+    router:
+        utf8: true
+
+        # Configure how to generate URLs in non-HTTP contexts, such as CLI commands.
+        # See https://symfony.com/doc/current/routing.html#generating-urls-in-commands
+        #default_uri: http://localhost
+
+when@prod:
+    framework:
+        router:
+            strict_requirements: null

tests/Application/config/packages/win32service.yaml

@@ -0,0 +1,11 @@
+win32_service:
+  project_code: demo
+  messenger:
+    -
+      receivers: [async]
+      limit: 10
+      displayed_name: Demo Messenger Consumer Async %d
+      thread_count: 2
+      memory_limit: 128M
+      time_limit: 1
+

tests/Application/config/preload.php

@@ -0,0 +1,7 @@
+<?php
+
+declare(strict_types=1);
+
+if (file_exists(\dirname(__DIR__).'/var/cache/prod/App_KernelProdContainer.preload.php')) {
+    require \dirname(__DIR__).'/var/cache/prod/App_KernelProdContainer.preload.php';
+}

tests/Application/config/routes.yaml

@@ -0,0 +1,3 @@
+#index:
+#    path: /
+#    controller: App\Controller\DefaultController::index

tests/Application/config/routes/framework.yaml

@@ -0,0 +1,4 @@
+when@dev:
+    _errors:
+        resource: '@FrameworkBundle/Resources/config/routing/errors.xml'
+        prefix: /_error

tests/Application/config/services.yaml

@@ -0,0 +1,25 @@
+# This file is the entry point to configure your own services.
+# Files in the packages/ subdirectory configure your dependencies.
+
+# Put parameters here that don't need to change on each machine where the app is deployed
+# https://symfony.com/doc/current/best_practices.html#use-parameters-for-application-configuration
+parameters:
+
+services:
+    # default configuration for services in *this* file
+    _defaults:
+        autowire: true      # Automatically injects dependencies in your services.
+        autoconfigure: true # Automatically registers your services as commands, event subscribers, etc.
+
+    # makes classes in src/ available to be used as services
+    # this creates a service per class whose id is the fully-qualified class name
+    Win32ServiceBundle\Tests\Application\:
+        resource: '../src/'
+        exclude:
+            - '../src/DependencyInjection/'
+            - '../src/Entity/'
+            - '../src/Kernel.php'
+            - '../Event/'
+
+    # add more service definitions when explicit configuration is needed
+    # please note that last definitions always *replace* previous ones

tests/Application/migrations/Version20240725152658.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DoctrineMigrations;
+
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\Migrations\AbstractMigration;
+
+/**
+ * Auto-generated Migration: Please modify to your needs!
+ */
+final class Version20240725152658 extends AbstractMigration
+{
+    public function getDescription(): string
+    {
+        return '';
+    }
+
+    public function up(Schema $schema): void
+    {
+        // this up() migration is auto-generated, please modify it to your needs
+        $this->addSql('CREATE TABLE messenger_messages (id BIGINT AUTO_INCREMENT NOT NULL, body LONGTEXT NOT NULL, headers LONGTEXT NOT NULL, queue_name VARCHAR(190) NOT NULL, created_at DATETIME NOT NULL, available_at DATETIME NOT NULL, delivered_at DATETIME DEFAULT NULL, INDEX IDX_75EA56E0FB7336F0 (queue_name), INDEX IDX_75EA56E0E3BD61CE (available_at), INDEX IDX_75EA56E016BA31DB (delivered_at), PRIMARY KEY(id)) DEFAULT CHARACTER SET utf8mb4 COLLATE `utf8mb4_unicode_ci` ENGINE = InnoDB');
+    }
+
+    public function down(Schema $schema): void
+    {
+        // this down() migration is auto-generated, please modify it to your needs
+        $this->addSql('DROP TABLE messenger_messages');
+    }
+}

tests/Application/public/index.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+use App\Kernel;
+
+require_once \dirname(__DIR__).'/vendor/autoload_runtime.php';
+
+return function (array $context) {
+    return new Kernel($context['APP_ENV'], (bool) $context['APP_DEBUG']);
+};

tests/Application/src/Command/SendMessageCommand.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Command;
+
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Win32ServiceBundle\Tests\Application\Event\TestMessage;
+
+#[AsCommand('test:send-message', 'Send a Test Message')]
+final class SendMessageCommand extends Command
+{
+    public function __construct(private MessageBusInterface $messageBus)
+    {
+        parent::__construct();
+    }
+
+    protected function configure(): void
+    {
+        $this->addArgument('message', InputArgument::REQUIRED, 'Your test message');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $message = $input->getArgument('message');
+
+        $this->messageBus->dispatch(new TestMessage($message));
+
+        return self::SUCCESS;
+    }
+}

tests/Application/src/Event/TestFailedMessage.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Event;
+
+final class TestFailedMessage
+{
+}

tests/Application/src/Event/TestMemoryLimitMessage.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Event;
+
+final class TestMemoryLimitMessage
+{
+    public function __construct(public int $size)
+    {
+    }
+}

tests/Application/src/Event/TestMessage.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Event;
+
+final class TestMessage
+{
+    public function __construct(public string $message)
+    {
+    }
+}

tests/Application/src/Event/TestRetryMessage.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Event;
+
+final class TestRetryMessage
+{
+}

tests/Application/src/Event/TestTimeLimitMessage.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Event;
+
+final class TestTimeLimitMessage
+{
+    public function __construct(public int $durationInSeconds)
+    {
+    }
+}

tests/Application/src/Handler/FailMessageHandler.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Handler;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+use Win32ServiceBundle\Tests\Application\Event\TestFailedMessage;
+
+#[AsMessageHandler(fromTransport: 'async')]
+final class FailMessageHandler
+{
+    public function __construct(private LoggerInterface $logger)
+    {
+    }
+
+    public function __invoke(TestFailedMessage $message): void
+    {
+        $this->logger->info('Failed Message');
+        throw new \LogicException('Fail to process');
+    }
+}

tests/Application/src/Handler/MemoryLimitMessageHandler.php

@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Handler;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+use Win32ServiceBundle\Tests\Application\Event\TestMemoryLimitMessage;
+
+#[AsMessageHandler(fromTransport: 'async')]
+final class MemoryLimitMessageHandler
+{
+    /** @var string Buffer to consume memory to stop service */
+    private string $buffer = '';
+
+    public function __construct(private LoggerInterface $logger)
+    {
+    }
+
+    public function __invoke(TestMemoryLimitMessage $message): void
+    {
+        $this->logger->info('Memory Limit Message : '.$message->size);
+        $this->buffer = str_repeat('-*+45defse', (int) ($message->size / 10));
+    }
+}

tests/Application/src/Handler/RetryMessageHandler.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Handler;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+use Symfony\Component\Messenger\Exception\RecoverableMessageHandlingException;
+use Win32ServiceBundle\Tests\Application\Event\TestRetryMessage;
+
+#[AsMessageHandler(fromTransport: 'async')]
+final class RetryMessageHandler
+{
+    public function __construct(private LoggerInterface $logger)
+    {
+    }
+
+    public function __invoke(TestRetryMessage $message): void
+    {
+        $this->logger->info('Retry Message');
+        throw new RecoverableMessageHandlingException('Retry Message');
+    }
+}

tests/Application/src/Handler/TestMessageHandler.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Handler;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+use Win32ServiceBundle\Tests\Application\Event\TestMessage;
+
+#[AsMessageHandler(fromTransport: 'async')]
+final class TestMessageHandler
+{
+    public function __construct(private LoggerInterface $logger)
+    {
+    }
+
+    public function __invoke(TestMessage $message): void
+    {
+        $this->logger->info(__METHOD__.' - message : '.$message->message);
+    }
+}

tests/Application/src/Handler/TimeLimitMessageHandler.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application\Handler;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+use Win32ServiceBundle\Tests\Application\Event\TestTimeLimitMessage;
+
+#[AsMessageHandler(fromTransport: 'async')]
+final class TimeLimitMessageHandler
+{
+    public function __construct(private LoggerInterface $logger)
+    {
+    }
+
+    public function __invoke(TestTimeLimitMessage $message): void
+    {
+        $this->logger->info('Time Limit Message : '.$message->durationInSeconds);
+        sleep($message->durationInSeconds);
+    }
+}

tests/Application/src/Kernel.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Application;
+
+use Symfony\Bundle\FrameworkBundle\Kernel\MicroKernelTrait;
+use Symfony\Component\HttpKernel\Kernel as BaseKernel;
+
+class Kernel extends BaseKernel
+{
+    use MicroKernelTrait;
+}

tests/Unit/MessengerIntegration/FaillureRetryMessageTest.php

@@ -0,0 +1,86 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Unit\MessengerIntegration;
+
+require_once \dirname(__DIR__, 2).'/Win32serviceState.php';
+
+use Doctrine\DBAL\Driver\Connection;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Win32Service\Model\AbstractServiceRunner;
+use Win32Service\Model\ServiceIdentifier;
+use Win32Service\Model\Win32serviceState;
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+use Win32ServiceBundle\Service\RunnerManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
+use Win32ServiceBundle\Tests\Application\Event\TestFailedMessage;
+
+final class FaillureRetryMessageTest extends KernelTestCase
+{
+    protected function setUp(): void
+    {
+        Win32serviceState::reset();
+    }
+
+    protected function tearDown(): void
+    {
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->rollBack();
+    }
+
+    public function testFailureMessage(): void
+    {
+        $serviceName = 'win32service.demo.messenger.async.0';
+        self::bootKernel();
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->beginTransaction();
+        $connexion->query('DELETE FROM messenger_messages');
+        /** @var MessageBusInterface $messengerBus */
+        $messengerBus = $container->get('messenger.bus.default');
+        $messengerBus->dispatch(new TestFailedMessage());
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\'');
+
+        $this->assertSame(1, (int) $c->fetchOne());
+
+        $runnerManager = $container->get(RunnerManager::class);
+        $serviceConfigurationManager = $container->get(ServiceConfigurationManager::class);
+        /** @var MessengerServiceRunner $runner */
+        $runner = $runnerManager->getRunner($serviceConfigurationManager->getRunnerAliasForServiceId($serviceName));
+        $runner->setServiceId(new ServiceIdentifier($serviceName));
+        $runner->doRun(1, 0);
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NULL');
+        $this->assertSame(1, (int) $c->fetchOne());
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NOT NULL');
+        $this->assertSame(1, (int) $c->fetchOne());
+
+        $msrRefrection = new \ReflectionClass(AbstractServiceRunner::class);
+        $stopRequestedProperty = $msrRefrection->getProperty('stopRequested');
+        $stopRequestedProperty->setAccessible(true);
+
+        Win32serviceState::reset();
+        $stopRequestedProperty->setValue($runner, false);
+
+        usleep(1_500_000);
+
+        $runner->doRun(1, 0);
+        $connexion->commit();
+        $connexion->beginTransaction();
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NOT NULL');
+        $this->assertSame(1, (int) $c->fetchOne());
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'failed\' AND delivered_at IS NULL');
+        $this->assertSame(1, (int) $c->fetchOne());
+    }
+}

tests/Unit/MessengerIntegration/LimitNbMessageTest.php

@@ -0,0 +1,78 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Unit\MessengerIntegration;
+
+require_once \dirname(__DIR__, 2).'/Win32serviceState.php';
+
+use Doctrine\DBAL\Driver\Connection;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Win32Service\Model\AbstractServiceRunner;
+use Win32Service\Model\ServiceIdentifier;
+use Win32Service\Model\Win32serviceState;
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+use Win32ServiceBundle\Service\RunnerManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
+use Win32ServiceBundle\Tests\Application\Event\TestMessage;
+
+final class LimitNbMessageTest extends KernelTestCase
+{
+    protected function setUp(): void
+    {
+        Win32serviceState::reset();
+    }
+
+    protected function tearDown(): void
+    {
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->rollBack();
+    }
+
+    public function testLimitMessage(): void
+    {
+        $serviceName = 'win32service.demo.messenger.async.0';
+        self::bootKernel();
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->beginTransaction();
+        $connexion->query('DELETE FROM messenger_messages');
+        /** @var MessageBusInterface $messengerBus */
+        $messengerBus = $container->get('messenger.bus.default');
+        $messagesTotal = 20;
+        for ($i = 1; $i <= $messagesTotal; ++$i) {
+            $messengerBus->dispatch(new TestMessage('message '.$i));
+        }
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\'');
+
+        $this->assertSame($messagesTotal, (int) $c->fetchOne());
+
+        $runnerManager = $container->get(RunnerManager::class);
+        $serviceConfigurationManager = $container->get(ServiceConfigurationManager::class);
+        /** @var MessengerServiceRunner $runner */
+        $runner = $runnerManager->getRunner($serviceConfigurationManager->getRunnerAliasForServiceId($serviceName));
+        $runner->setServiceId(new ServiceIdentifier($serviceName));
+        $runner->doRun($messagesTotal, 0);
+
+        $rClass = new \ReflectionClass(AbstractServiceRunner::class);
+        $value = $rClass->getProperty('stopRequested');
+        $value->setAccessible(true);
+
+        $this->assertTrue($value->getValue($runner));
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NULL');
+
+        $this->assertSame(10, (int) $c->fetchOne());
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NOT NULL');
+        // Other message has been deleted, only last processed message is keep
+        $this->assertSame(1, (int) $c->fetchOne());
+    }
+}

tests/Unit/MessengerIntegration/MemoryLimitMessageTest.php

@@ -0,0 +1,77 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Unit\MessengerIntegration;
+
+require_once \dirname(__DIR__, 2).'/Win32serviceState.php';
+
+use Doctrine\DBAL\Driver\Connection;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Win32Service\Model\AbstractServiceRunner;
+use Win32Service\Model\ServiceIdentifier;
+use Win32Service\Model\Win32serviceState;
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+use Win32ServiceBundle\Service\RunnerManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
+use Win32ServiceBundle\Tests\Application\Event\TestMemoryLimitMessage;
+use Win32ServiceBundle\Tests\Application\Event\TestMessage;
+
+final class MemoryLimitMessageTest extends KernelTestCase
+{
+    protected function setUp(): void
+    {
+        Win32serviceState::reset();
+    }
+
+    protected function tearDown(): void
+    {
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->rollBack();
+    }
+
+    public function testMemoryLimitMessage(): void
+    {
+        $serviceName = 'win32service.demo.messenger.async.0';
+        self::bootKernel();
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->beginTransaction();
+        $connexion->query('DELETE FROM messenger_messages');
+        /** @var MessageBusInterface $messengerBus */
+        $messengerBus = $container->get('messenger.bus.default');
+        $messengerBus->dispatch(new TestMemoryLimitMessage( /* 129 Mio */1024 * 1024 * 129));
+        $messengerBus->dispatch(new TestMessage('message 1'));
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\'');
+
+        $this->assertSame(2, (int) $c->fetchOne());
+
+        $runnerManager = $container->get(RunnerManager::class);
+        $serviceConfigurationManager = $container->get(ServiceConfigurationManager::class);
+        /** @var MessengerServiceRunner $runner */
+        $runner = $runnerManager->getRunner($serviceConfigurationManager->getRunnerAliasForServiceId($serviceName));
+        $runner->setServiceId(new ServiceIdentifier($serviceName));
+        $runner->doRun(5, 0);
+
+        $rClass = new \ReflectionClass(AbstractServiceRunner::class);
+        $value = $rClass->getProperty('stopRequested');
+        $value->setAccessible(true);
+
+        $this->assertTrue($value->getValue($runner));
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NULL');
+
+        $this->assertSame(1, (int) $c->fetchOne());
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NOT NULL');
+
+        $this->assertSame(1, (int) $c->fetchOne());
+    }
+}

tests/Unit/MessengerIntegration/MessageTest.php

@@ -0,0 +1,68 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Unit\MessengerIntegration;
+
+require_once \dirname(__DIR__, 2).'/Win32serviceState.php';
+
+use Doctrine\DBAL\Driver\Connection;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Win32Service\Model\ServiceIdentifier;
+use Win32Service\Model\Win32serviceState;
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+use Win32ServiceBundle\Service\RunnerManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
+use Win32ServiceBundle\Tests\Application\Event\TestMessage;
+
+final class MessageTest extends KernelTestCase
+{
+    protected function setUp(): void
+    {
+        Win32serviceState::reset();
+    }
+
+    protected function tearDown(): void
+    {
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->rollBack();
+    }
+
+    public function testNormalMessage(): void
+    {
+        $serviceName = 'win32service.demo.messenger.async.0';
+        self::bootKernel();
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->beginTransaction();
+        $connexion->query('DELETE FROM messenger_messages');
+        /** @var MessageBusInterface $messengerBus */
+        $messengerBus = $container->get('messenger.bus.default');
+        $messengerBus->dispatch(new TestMessage('message 1'));
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\'');
+
+        $this->assertSame(1, (int) $c->fetchOne());
+
+        $runnerManager = $container->get(RunnerManager::class);
+        $serviceConfigurationManager = $container->get(ServiceConfigurationManager::class);
+        /** @var MessengerServiceRunner $runner */
+        $runner = $runnerManager->getRunner($serviceConfigurationManager->getRunnerAliasForServiceId($serviceName));
+        $runner->setServiceId(new ServiceIdentifier($serviceName));
+        $runner->doRun(1, 0);
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NULL');
+
+        $this->assertSame(0, (int) $c->fetchOne());
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NOT NULL');
+
+        $this->assertSame(1, (int) $c->fetchOne());
+    }
+}

tests/Unit/MessengerIntegration/RetryMessageTest.php

@@ -0,0 +1,64 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Unit\MessengerIntegration;
+
+require_once \dirname(__DIR__, 2).'/Win32serviceState.php';
+
+use Doctrine\DBAL\Driver\Connection;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Win32Service\Model\ServiceIdentifier;
+use Win32Service\Model\Win32serviceState;
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+use Win32ServiceBundle\Service\RunnerManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
+use Win32ServiceBundle\Tests\Application\Event\TestRetryMessage;
+
+final class RetryMessageTest extends KernelTestCase
+{
+    protected function setUp(): void
+    {
+        Win32serviceState::reset();
+    }
+
+    protected function tearDown(): void
+    {
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->rollBack();
+    }
+
+    public function testRetryMessage(): void
+    {
+        $serviceName = 'win32service.demo.messenger.async.0';
+        self::bootKernel();
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->beginTransaction();
+        $connexion->query('DELETE FROM messenger_messages');
+        /** @var MessageBusInterface $messengerBus */
+        $messengerBus = $container->get('messenger.bus.default');
+        $messengerBus->dispatch(new TestRetryMessage());
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\'');
+
+        $this->assertSame(1, (int) $c->fetchOne());
+
+        $runnerManager = $container->get(RunnerManager::class);
+        $serviceConfigurationManager = $container->get(ServiceConfigurationManager::class);
+        /** @var MessengerServiceRunner $runner */
+        $runner = $runnerManager->getRunner($serviceConfigurationManager->getRunnerAliasForServiceId($serviceName));
+        $runner->setServiceId(new ServiceIdentifier($serviceName));
+        $runner->doRun(1, 0);
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\'');
+
+        $this->assertSame(2, (int) $c->fetchOne());
+    }
+}

tests/Unit/MessengerIntegration/TimeLimitMessageTest.php

@@ -0,0 +1,77 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Win32ServiceBundle\Tests\Unit\MessengerIntegration;
+
+require_once \dirname(__DIR__, 2).'/Win32serviceState.php';
+
+use Doctrine\DBAL\Driver\Connection;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Win32Service\Model\AbstractServiceRunner;
+use Win32Service\Model\ServiceIdentifier;
+use Win32Service\Model\Win32serviceState;
+use Win32ServiceBundle\Model\MessengerServiceRunner;
+use Win32ServiceBundle\Service\RunnerManager;
+use Win32ServiceBundle\Service\ServiceConfigurationManager;
+use Win32ServiceBundle\Tests\Application\Event\TestMessage;
+use Win32ServiceBundle\Tests\Application\Event\TestTimeLimitMessage;
+
+final class TimeLimitMessageTest extends KernelTestCase
+{
+    protected function setUp(): void
+    {
+        Win32serviceState::reset();
+    }
+
+    protected function tearDown(): void
+    {
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->rollBack();
+    }
+
+    public function testTimeLimitMessage(): void
+    {
+        $serviceName = 'win32service.demo.messenger.async.0';
+        self::bootKernel();
+        $container = static::getContainer();
+
+        /** @var Connection $connexion */
+        $connexion = $container->get('doctrine.dbal.default_connection');
+        $connexion->beginTransaction();
+        $connexion->query('DELETE FROM messenger_messages');
+        /** @var MessageBusInterface $messengerBus */
+        $messengerBus = $container->get('messenger.bus.default');
+        $messengerBus->dispatch(new TestTimeLimitMessage(2));
+        $messengerBus->dispatch(new TestMessage('message 1'));
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\'');
+
+        $this->assertSame(2, (int) $c->fetchOne());
+
+        $runnerManager = $container->get(RunnerManager::class);
+        $serviceConfigurationManager = $container->get(ServiceConfigurationManager::class);
+        /** @var MessengerServiceRunner $runner */
+        $runner = $runnerManager->getRunner($serviceConfigurationManager->getRunnerAliasForServiceId($serviceName));
+        $runner->setServiceId(new ServiceIdentifier($serviceName));
+        $runner->doRun(5, 0);
+
+        $rClass = new \ReflectionClass(AbstractServiceRunner::class);
+        $value = $rClass->getProperty('stopRequested');
+        $value->setAccessible(true);
+
+        $this->assertTrue($value->getValue($runner));
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NULL');
+
+        $this->assertSame(1, (int) $c->fetchOne());
+
+        $c = $connexion->query('SELECT count(*) FROM messenger_messages WHERE queue_name = \'default\' AND delivered_at IS NOT NULL');
+
+        $this->assertSame(1, (int) $c->fetchOne());
+    }
+}

tests/Win32serviceState.php

@@ -0,0 +1,85 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * Mock for service library abstact.
+ */
+
+namespace Win32Service\Model;
+
+function win32_start_service_ctrl_dispatcher(string $serviceName): bool
+{
+    return Win32serviceState::getInstance()->setServiceName($serviceName);
+}
+
+function win32_set_service_status(int $newState): void
+{
+    Win32serviceState::getInstance()->changeState($newState);
+}
+
+function win32_get_last_control_message(): int
+{
+    return Win32serviceState::getInstance()->getLastControlMessage();
+}
+
+class Win32serviceState
+{
+    private static ?self $instance = null;
+
+    private int $state = WIN32_SERVICE_STOPPED;
+
+    private ?string $serviceName = null;
+
+    private int $lastControlMessage = WIN32_SERVICE_CONTROL_INTERROGATE;
+
+    public static function getInstance(): self
+    {
+        if (self::$instance === null) {
+            self::$instance = new self();
+        }
+
+        return self::$instance;
+    }
+
+    public static function reset(): void
+    {
+        self::$instance = null;
+    }
+
+    public function setServiceName(string $serviceName): bool
+    {
+        if ($this->serviceName === null) {
+            $this->serviceName = $serviceName;
+
+            return true;
+        }
+
+        return false;
+    }
+
+    public function getServiceName(): string
+    {
+        return $this->serviceName;
+    }
+
+    public function changeState(int $newState): void
+    {
+        $this->state = $newState;
+    }
+
+    public function getState(): int
+    {
+        return $this->state;
+    }
+
+    public function getLastControlMessage(): int
+    {
+        return $this->lastControlMessage;
+    }
+
+    public function setLastControlMessage(int $newControlMessage): void
+    {
+        $this->lastControlMessage = $newControlMessage;
+    }
+}

tests/bootstrap.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+use Symfony\Component\Dotenv\Dotenv;
+
+require \dirname(__DIR__).'/vendor/autoload.php';
+
+if (file_exists(__DIR__.'/Application/config/bootstrap.php')) {
+    require __DIR__.'/Application/config/bootstrap.php';
+} elseif (method_exists(Dotenv::class, 'bootEnv')) {
+    (new Dotenv())->bootEnv(__DIR__.'/Application/.env', 'test');
+}