Skip to content
  • Documentation >
  • Infrastructure and maintenance >
  • Background tasks

Background tasks

Some operations in Ibexa DXP don’t have to run immediately when a user clicks a button, for example, re-indexing product prices or processing bulk data. Running such operations in real time could slow down the system and disrupt the user experience.

To solve this, Ibexa DXP provides a package called Ibexa Messenger, which is an overlay to Symfony Messenger, and it's job is to queue tasks and run them in the background. Ibexa DXP sends messages (or commands) that represent the work to be done later. These messages are stored in a queue and picked up by a background worker, which ensures that resource-heavy tasks are executed at a convenient time, without putting excessive load on the system.

Ibexa Messenger supports multiple storage backends, such as Doctrine, Redis/Valkey, and PostgreSQL, and gives developers the flexibility to create their own message handlers for custom use cases.

Installation

To use Ibexa Messenger, you must first install the package and set up the database tables.

Install package

Install the ibexa/messenger package:

1
composer require ibexa/messenger:4.6.31

Set up database

Run the following SQL script to create the required database tables.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
-- ibexa/messenger
CREATE TABLE IF NOT EXISTS ibexa_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 COMMENT '(DC2Type:datetime_immutable)',
    available_at DATETIME NOT NULL COMMENT '(DC2Type:datetime_immutable)',
    delivered_at DATETIME DEFAULT NULL COMMENT '(DC2Type:datetime_immutable)',
    INDEX ibexa_messenger_created_at_idx (created_at),
    INDEX ibexa_messenger_available_at_idx (available_at),
    INDEX ibexa_messenger_delivered_at_idx (delivered_at),
    PRIMARY KEY(id)
) DEFAULT CHARACTER SET utf8mb4 COLLATE `utf8mb4_unicode_520_ci` ENGINE = InnoDB;

CREATE TABLE IF NOT EXISTS ibexa_messenger_lock_keys (
    key_id VARCHAR(64) NOT NULL,
    key_token VARCHAR(44) NOT NULL,
    key_expiration INT UNSIGNED NOT NULL,
    PRIMARY KEY(key_id)
) DEFAULT CHARACTER SET utf8mb4 COLLATE `utf8mb4_unicode_520_ci` ENGINE = InnoDB;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
-- ibexa/messenger
CREATE TABLE IF NOT EXISTS ibexa_messenger_messages (
    id BIGSERIAL NOT NULL,
    body TEXT NOT NULL,
    headers TEXT NOT NULL,
    queue_name VARCHAR(190) NOT NULL,
    created_at TIMESTAMP(0) WITHOUT TIME ZONE NOT NULL,
    available_at TIMESTAMP(0) WITHOUT TIME ZONE NOT NULL,
    delivered_at TIMESTAMP(0) WITHOUT TIME ZONE DEFAULT NULL,
    PRIMARY KEY(id)
);

CREATE INDEX IF NOT EXISTS ibexa_messenger_created_at_idx ON ibexa_messenger_messages (created_at);
CREATE INDEX IF NOT EXISTS ibexa_messenger_available_at_idx ON ibexa_messenger_messages (available_at);
CREATE INDEX IF NOT EXISTS ibexa_messenger_delivered_at_idx ON ibexa_messenger_messages (delivered_at);
COMMENT ON COLUMN ibexa_messenger_messages.created_at IS '(DC2Type:datetime_immutable)';
COMMENT ON COLUMN ibexa_messenger_messages.available_at IS '(DC2Type:datetime_immutable)';
COMMENT ON COLUMN ibexa_messenger_messages.delivered_at IS '(DC2Type:datetime_immutable)';

CREATE TABLE IF NOT EXISTS ibexa_messenger_lock_keys (
    key_id VARCHAR(64) NOT NULL,
    key_token VARCHAR(44) NOT NULL,
    key_expiration INT NOT NULL,
    PRIMARY KEY(key_id)
);

How it works

Ibexa Messenger uses a command bus as a queue that stores messages, or commands, which tell the system what you want to happen, and separates them from the handler, which is the code that actually performs the task.

The process works as follows:

  1. A message PHP object is dispatched, for example, ProductPriceReindex.
  2. The message is wrapped in an envelope, which may contain additional metadata, called stamps.
  3. The message is placed in the transport queue. It can be a Doctrine table, a Redis/Valkey queue, and so on.
  4. A worker process continuously reads messages from the queue, pulls them into the default bus ibexa.messenger.bus and assigns them to the right handler.
  5. A handler service processes the message (executes the command). You can register multiple handlers for different jobs.

Here is an example of how you can extend your code and use Ibexa Messenger to process your tasks:

Configure package

Create a config file, for example, config/packages/ibexa_messenger.yaml and define your transport:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
ibexa_messenger:

    # The DSN of the transport, as expected by Symfony Messenger transport factory.
    transport_dsn:        'doctrine://default?table_name=ibexa_messenger_messages&auto_setup=false'
    deduplication_lock_storage:
        enabled:              true

        # Doctrine DBAL primary connection or custom service
        type:                 doctrine # One of "doctrine"; "custom"; "service"

        # The service ID of a custom Lock Store, if "service" type is selected
        service:              null

        # The DSN of the lock store, if "custom" type is selected
        dsn:                  null

Supported transports

You can define different transports: Ibexa Messenger has been tested to work with Redis, MySQL, PostgreSQL. For more information, see Symfony Messenger documentation or Symfony Messenger tutorial.

Start worker

Use a process manager of your choice to run the following command, or make it start together with the server:

1
php bin/console messenger:consume ibexa.messenger.transport --bus=ibexa.messenger.bus --siteaccess=<OPTIONAL>

Use the --siteaccess option to set the default SiteAccess and repository for the worker process. The worker uses this SiteAccess for every message that doesn't have a SiteAccessStamp.

If a message has a SiteAccessStamp, the worker uses the SiteAccess from the stamp instead to processes this message. Thanks to this, one worker process can handle messages coming from different SiteAccesses.

In multi-repository setups, run one worker process for each repository. With this setup, each worker process can connect to the right database.

Multi-repository setups

Doctrine transport works across multiple repositories without issues, but other transports may need to be adjusted, so that queues across different repositories are not accidentally shared.

Deploying Ibexa Messenger

Additional considerations regarding the deployment of Symfony Messenger to production, which you can find in Symfony documentation apply to Ibexa Messenger as well.

Dispatch message

To have a task processed in the background by Ibexa Messenger:

  1. Inject the ibexa.messenger.bus service as an object implementing the Symfony\Component\Messenger\MessageBusInterface interface.

  2. Dispatch an appropriate message by using the MessageBusInterface::dispatch() method, exactly as described in Symfony Messenger documentation.

    1
2
3
4
    services:
    SomeClassThatSchedulesExecutionInTheBackground:
        arguments:
            $bus: '@ibexa.messenger.bus'

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
    <?php declare(strict_types=1);

namespace App\Dispatcher;

use Ibexa\Bundle\Messenger\Stamp\DeduplicateStamp;
use Symfony\Component\Messenger\MessageBusInterface;

final class SomeClassThatSchedulesExecutionInTheBackground
{
    private MessageBusInterface $bus;

    public function __construct(MessageBusInterface $bus)
    {
        $this->bus = $bus;
    }

    public function schedule(object $message): void
    {
        $this->bus->dispatch($message);
    }
}

  3. Route the message to the background queue.

  4. Additionally, attach message metadata by using stamps.

Stamps

You can attach Stamps to a message envelope to add additional metadata and control processing of the message.

The ibexa.messenger.bus message bus uses the default Symfony Messenger middleware and doesn't support all stamps that are available in Symfony.

You can use the following Symfony stamps:

On top of the supported Symfony stamps, Ibexa DXP provides the following ones:

DeduplicateStamp

Ibexa\Bundle\Messenger\Stamp\DeduplicateStamp prevents duplicate messages from being processed. When you attach it to a message, the system uses a lock to ensure that only one message with the same key is handled at a time.

This stamp is backported from Symfony 7. For more information, see Symfony 7.4 documentation about message deduplication.

SiteAccessStamp

Ibexa\Contracts\Messenger\Stamp\SiteAccessStamp contains the name of the SiteAccess that dispatched the message.

You don't need to add this stamp manually, Ibexa Messenger attaches this stamp to each dispatched message automatically. The stamp contains the SiteAccess that is current at the moment of dispatch.

Before the worker calls the handler, it changes the configuration scope to the SiteAccess from the stamp. The handler then reads SiteAccess-aware configuration for the SiteAccess that dispatched the message, and not for the SiteAccess that the worker process started with.

The stamp doesn't change the current SiteAccess

The stamp changes the configuration scope only. It doesn't change the SiteAccess in the Ibexa\Core\MVC\Symfony\SiteAccess\SiteAccessServiceInterface service. SiteAccessServiceInterface::getCurrent() always returns the SiteAccess that the worker process started with, for all messages.

To get a SiteAccess-aware value in a handler, use the ConfigResolverInterface service.

Extend Ibexa Messenger

Register custom message and handler

To handle additional use cases with background tasks, you can create custom message and handler class:

1
2
3
4
5
6
7
8
<?php declare(strict_types=1);

namespace App\Message;

class SomeMessage
{
    // Add properties and methods as needed for your message.
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
<?php declare(strict_types=1);

namespace App\MessageHandler;

use App\Message\SomeMessage;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;

final class SomeHandler implements MessageHandlerInterface
{
    public function __invoke(SomeMessage $message): void
    {
        // Handle message.
    }
}

Add a service definition to config/services.yaml and set the bus to ibexa.messenger.bus:

1
2
3
4
5
services:
    App\MessageHandler\SomeHandler:
        tags:
            - name: messenger.message_handler
              bus: ibexa.messenger.bus

Route message to background queue

To have a message processed in the background, it must be sent to a transport queue. Ibexa Messenger uses message providers instead of Symfony framework.messenger.routing configuration.

A message provider is a service that implements the MessageProviderInterface interface, and the getHandledClasses() method must return the list of message classes that Ibexa Messenger must send to the queue to process in the background.

The getHandledClasses() method can also return a parent class or an interface. In this case, all messages that extend this class, or implement this interface, go to the background queue.

If no message provider returns the class of your message, the bus calls the handler immediately, in the same process that dispatches the message.

To send SomeMessage to the background queue, create the following provider:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
<?php declare(strict_types=1);

namespace App\Messenger;

use App\Message\SomeMessage;
use Ibexa\Contracts\Messenger\Transport\MessageProviderInterface;

final class SomeMessageProvider implements MessageProviderInterface
{
    public function getHandledClasses(): iterable
    {
        return [SomeMessage::class];
    }
}

If you're not using service autoconfiguration, add the ibexa.messenger.sender_message_provider tag to the service:

1
2
3
4
services:
    App\Messenger\SomeMessageProvider:
        tags:
            - name: ibexa.messenger.sender_message_provider