Message bus 

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 

Implement the handler class

<?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
  }
}

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

A Services.yaml entry is needed to use before / after for registration if you need to define an order:

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'

Bind a handler to a receiving transport with fromTransport 

The fromTransport property of AsMessageHandler binds a handler to a receiving transport. The handler then runs only for messages that a worker received from this transport, for example with vendor/bin/typo3 messenger:consume doctrine:

<?php

declare(strict_types=1);

namespace MyVendor\MyExtension\Queue\Handler;

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

// Runs only for messages that a worker received from "doctrine"
#[AsMessageHandler(fromTransport: 'async')]
final class AsyncDemoHandler
{
  public function __invoke(DemoMessage $message): void
  {
    // do something with $message
  }
}

A message that is dispatched and handled synchronously still reaches the handler. A handler without fromTransport runs for every message, whatever the transport.

To bind a handler to several transports, repeat the attribute once for each transport. In the Services.yaml tag, fromTransport also takes a list of transport names:

MyVendor\MyExtension\Queue\Handler\AsyncDemoHandler:
  tags:
    - name: 'messenger.message_handler'
      fromTransport: ['async', 'priority']

Async message handling - the consume command 

To consume messages, run the command:

vendor/bin/typo3 messenger:consume <receiver-name>

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

By default, you should run:

vendor/bin/typo3 messenger:consume doctrine

typo3/sysext/core/bin/typo3 messenger:consume doctrine

The command takes almost all options of the Symfony command of the same name.

A worker does not stop on its own. In continuous operation, configure a limit so that the PHP process is periodically recycled to release memory:

--time-limit, -t

Stop after this number of seconds.

--limit, -l

Stop after this number of messages.

--failure-limit, -f

Stop after this number of failed messages.

--memory-limit, -m

Stop when the worker uses more than this much memory, for example 128M.

A limit only takes effect after the current message is handled, so the worker never stops in the middle of one. These options select what the worker reads and how it waits:

--all

Consume from every configured receiver.

--bus, -b

Dispatch the received messages to this bus. Without it, TYPO3 picks the bus itself.

--queues

Consume only from these queues of the receiver. Repeat the option per queue.

--sleep

Wait this number of seconds before asking for messages again when none were found. The default is 1.

--keepalive

Keep the connection of the transport alive, if the transport implements it. The default interval is 5 seconds.

Whichever limit is reached, the worker process exits with code 0. Run it under a service manager like systemd that keeps the service active and automatically restarts a fresh worker process.

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

[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.5 /var/www/myproject/vendor/bin/typo3 messenger:consume doctrine --time-limit=3600
# The command exits with 0 on its time limit, so restart it either way
Restart=always
RestartSec=1

[Install]
WantedBy=multi-user.target

Where no service manager is available, the command can run as a scheduler task instead.

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.

<?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',
  ],
];

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'

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)

Add rate limiter 

Rate limiting can be applied to asynchronous messages processed through the consume command. This allows controlling message processing rates to:

• Stay within external service limits (API quotas, mail sending thresholds)
• Manage server resource utilization

messenger.rate_limiter.limit_db_messages:
  class: Symfony\Component\RateLimiter\RateLimiterFactory
  arguments:
    $config:
      id: 'limit_db_messages'
      policy: 'sliding_window'
      limit: 100
      interval: '60 seconds'
    $storage: '@TYPO3\CMS\Core\RateLimiter\Storage\CachingFrameworkStorage'
  tags:
    - name: 'messenger.rate_limiter'
      identifier: 'doctrine'

InMemoryTransport for testing 

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

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'

Configure a custom middleware 

The middleware is set up in the services configuration. By default, these middlewares are registered, in this order:

1. SendMessageMiddleware
2. DoctrineCloseConnectionMiddleware
3. DoctrinePingConnectionMiddleware
4. HandleMessageMiddleware

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:

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'