Message bus

New in version 12.2

TYPO3 provides a message bus solution based on symfony/messenger. It has the ability to send messages and then handle them immediately (synchronous) or send them through transports (asynchronous, for example, queues) to be handled later.

For backwards compatibility, the default implementation uses the synchronous transport. This means that the message bus will behave exactly as before, but it will be possible to switch to a different (asynchronous) transport on a per-project base.

To offer asynchronicity, TYPO3 also provides a transport implementation based on the Doctrine DBAL messenger transport from Symfony and a basic implementation of a consumer command.

"Everyday" usage - as a developer

Dispatch a message

  1. Add a PHP class for your message object (which is an arbitrary PHP class)

    EXT:my_extension/Classes/Queue/Message/DemoMessage.php
    <?php
    
    declare(strict_types=1);
    
    namespace MyVendor\MyExtension\Queue\Message;
    
    final class DemoMessage
    {
        public function __construct(
            public readonly string $content,
        ) {}
    }
    
    Copied!
  2. Inject the MessageBusInterface into your class and call the dispatch() method

    EXT:my_extension/Classes/MyClass.php
    <?php
    
    declare(strict_types=1);
    
    namespace MyVendor\MyExtension;
    
    use MyVendor\MyExtension\Queue\Message\DemoMessage;
    use Symfony\Component\Messenger\MessageBusInterface;
    
    final class MyClass
    {
        public function __construct(
            private readonly MessageBusInterface $bus,
        ) {}
    
        public function doSomething(): void
        {
            // ...
            $this->bus->dispatch(new DemoMessage('test'));
            // ...
        }
    }
    
    Copied!

Register a handler

Changed in version 13.0

A message handler can be registered using the symfony PHP attribute \Symfony\Component\Messenger\Attribute\AsMessageHandler.

Implement the handler class

EXT:my_extension/Classes/Queue/Handler/DemoHandler.php
<?php

declare(strict_types=1);

namespace MyVendor\MyExtension\Queue\Handler;

use MyVendor\MyExtension\Queue\Message\DemoMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final class DemoHandler
{
    public function __invoke(DemoMessage $message): void
    {
        // do something with $message
    }
}
Copied!

If your extension needs to be compatible with TYPO3 v13 and v12, use a tag to register the handler. A Services.yaml entry is also needed to use before/after to define an order.

EXT:my_extension/Configuration/Services.yaml
MyVendor\MyExtension\Queue\Handler\DemoHandler:
  tags:
    - name: 'messenger.message_handler'

# Define another handler which should be called before DemoHandler:
MyVendor\MyExtension\Queue\Handler\DemoHandler2:
  tags:
    - name: 'messenger.message_handler'
      before: 'MyVendor\MyExtension\Queue\Handler\DemoHandler'
Copied!

"Everyday" usage - as a system administrator/integrator

By default, TYPO3 will behave like in versions before TYPO3 v12. This means that the message bus will use the synchronous transport and all messages will be handled immediately. To benefit from the message bus, it is recommended to switch to an asynchronous transport. Using asynchronous transports increases the resilience of the system by decoupling external dependencies even further.

Currently, the TYPO3 Core provides an asynchronous transport based on the Doctrine DBAL messenger transport. This transport is configured to use the default TYPO3 database connection. It is pre-configured and can be used by changing the settings:

config/settings.php | config.additional.php
$GLOBALS['TYPO3_CONF_VARS']['SYS']['messenger']['routing']['*'] = 'doctrine';
Copied!

This will route all messages to the asynchronous transport (mind the *).

Async message handling - The consume command

To consume messages, run the command:

vendor/bin/typo3 messenger:consume <receiver-name>
Copied!
typo3/sysext/core/bin/typo3 messenger:consume <receiver-name>
Copied!

By default, you should run:

vendor/bin/typo3 messenger:consume doctrine
Copied!
typo3/sysext/core/bin/typo3 messenger:consume doctrine
Copied!

The command is a slimmed-down wrapper for the Symfony command messenger:consume, it only provides the basic consumption functionality. As this command is running as a worker, it is stopped after 1 hour to avoid memory leaks. Therefore, the command should be run from a service manager like systemd to restart automatically after the command exits due to the time limit.

The following code provides an example for a service. Create the following file on your server:

/etc/systemd/system/typo3-message-consumer.service
[Unit]
Description=Run the TYPO3 message consumer
Requires=mariadb.service
After=mariadb.service

[Service]
Type=simple
User=www-data
Group=www-data
ExecStart=/usr/bin/php8.1 /var/www/myproject/vendor/bin/typo3 messenger:consume doctrine --exit-code-on-limit 133
# Generally restart on error
Restart=on-failure
# Restart on exit code 133 (which is returned by the command when limits are reached)
RestartForceExitStatus=133
# ..but do not interpret exit code 133 as an error (as it's just a restart request)
SuccessExitStatus=133

[Install]
WantedBy=multi-user.target
Copied!

Advanced usage

Configure a custom transport (Senders/Receivers)

Transports are configured in the services configuration. To allow the configuration of a transport per message, the TYPO3 configuration (settings.php, additional.php on system level, or ext_localconf.php in an extension) is utilized. The transport/sender name used in the settings is resolved to a service that has been tagged with message.sender and the respective identifier.

config/settings.php | config/additional.php | EXT:my_extension/ext_localconf.php
$GLOBALS['TYPO3_CONF_VARS']['SYS']['messenger'] = [
    'routing' => [
        // Use "messenger.transport.demo" as transport for DemoMessage
        \MyVendor\MyExtension\Queue\Message\DemoMessage::class => 'demo',

        // Use "messenger.transport.default" as transport for all other messages
        '*' => 'default',
    ]
];
Copied!
EXT:my_extension/Configuration/Services.yaml | config/system/services.yaml
messenger.transport.demo:
  factory: [ '@TYPO3\CMS\Core\Messenger\DoctrineTransportFactory', 'createTransport' ]
  class: 'Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport'
  arguments:
    $options:
      queue_name: 'demo'
  tags:
    - name: 'messenger.sender'
      identifier: 'demo'
    - name: 'messenger.receiver'
      identifier: 'demo'

messenger.transport.default:
  factory: [ '@Symfony\Component\Messenger\Transport\InMemory\InMemoryTransportFactory', 'createTransport' ]
  class: 'Symfony\Component\Messenger\Transport\InMemory\InMemoryTransport'
  arguments:
    $dsn: 'in-memory://default'
    $options: [ ]
  tags:
    - name: 'messenger.sender'
      identifier: 'default'
    - name: 'messenger.receiver'
      identifier: 'default'
Copied!

The TYPO3 Core has been tested with three transports:

  • \Symfony\Component\Messenger\Transport\Sync\SyncTransport (default)
  • \Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport (using the Doctrine DBAL messenger transport)
  • \Symfony\Component\Messenger\Transport\InMemory\InMemoryTransport (for testing)

InMemoryTransport for testing

The InMemoryTransport is a transport that should only be used while testing.

EXT:my_extension/Configuration/Services.yaml | config/system/services.yaml
messenger.transport.default:
  factory: [ '@Symfony\Component\Messenger\Transport\InMemory\InMemoryTransportFactory', 'createTransport' ]
  class: 'Symfony\Component\Messenger\Transport\InMemory\InMemoryTransport'
  public: true
  arguments:
    $dsn: 'in-memory://default'
    $options: [ ]
  tags:
    - name: 'messenger.sender'
      identifier: 'default'
    - name: 'messenger.receiver'
      identifier: 'default'
Copied!

Configure a custom middleware

The middleware is set up in the services configuration. By default, the \Symfony\Component\Messenger\Middleware\SendMessageMiddleware and the \Symfony\Component\Messenger\Middleware\HandleMessageMiddleware are registered. See also the Custom middleware section in the Symfony documentation.

To add your own middleware, tag it as messenger.middleware and set the order using TYPO3's before and after ordering mechanism:

EXT:my_extension/Configuration/Services.yaml | config/system/services.yaml
Symfony\Component\Messenger\Middleware\SendMessageMiddleware:
  arguments:
    $sendersLocator: '@Symfony\Component\Messenger\Transport\Sender\SendersLocatorInterface'
    $eventDispatcher: '@Psr\EventDispatcher\EventDispatcherInterface'
  tags:
    - { name: 'messenger.middleware' }

Symfony\Component\Messenger\Middleware\HandleMessageMiddleware:
  arguments:
    $handlersLocator: '@Symfony\Component\Messenger\Handler\HandlersLocatorInterface'
  tags:
    - name: 'messenger.middleware'
      after: 'Symfony\Component\Messenger\Middleware\SendMessageMiddleware'
Copied!