Guida avanzata per implementare workflow asincroni in Pimcore con Symfony Messenger. Gestisci importazioni massive e operazioni distribuite senza bloccare il sistema.

Architettura Event-Driven in Pimcore: Implementare un Sistema di Workflow Asincrono con Symfony Messenger

Chiunque abbia gestito un’istanza Pimcore in produzione con centinaia di migliaia di DataObjects e Assets sa cosa significa vedere l’interfaccia admin bloccarsi durante un’importazione massiva. O peggio, ricevere una chiamata alle 3 di notte perché il cron di sincronizzazione con l’ERP ha saturato la memoria del server.

La soluzione non è “comprare più RAM”. La soluzione è ripensare l’architettura: spostare le operazioni pesanti fuori dal ciclo request-response e orchestrarle in modo asincrono. Pimcore, essendo costruito su Symfony, ci offre Messenger come strumento nativo per farlo. Configurarlo correttamente per un contesto PIM/DAM enterprise richiede però una comprensione profonda sia di Messenger che delle peculiarità di Pimcore.

In questo articolo costruiremo un sistema di workflow asincrono completo, partendo dalla configurazione dei transport fino all’implementazione del Saga Pattern per operazioni distribuite. Vedremo come gestire importazioni di 500.000 prodotti senza impattare gli utenti, come orchestrare la generazione di thumbnail su più worker e come implementare retry intelligenti che non trasformino un problema temporaneo in un disastro.

Prerequisiti

Pimcore 11.x con PHP 8.2+
Conoscenza solida di Symfony (service container, eventi, console commands)
Redis 7+ o RabbitMQ 3.12+ installato e configurato
Familiarità con i concetti di message queue (producer, consumer, acknowledgment)
Docker per l’ambiente di sviluppo (opzionale ma consigliato)

Assumo che tu abbia già un’installazione Pimcore funzionante. Se stai partendo da zero, la documentazione ufficiale copre l’installazione base.

Architettura e Concetti Chiave

Prima di scrivere codice, capiamo cosa stiamo costruendo:

flowchart TD
    subgraph APP["PIMCORE APPLICATION"]
        HR[HTTP Request] --> ED[Event Dispatcher]
        UI[Admin UI] --> PE[Pimcore Events\npostUpdate / postAdd]
        CLI[CLI / Cron] --> MB
        ED --> MB[Message Bus]
        PE --> MB
    end

    MB --> T

    subgraph T["MESSAGE TRANSPORTS"]
        RDS["Redis Streams\n── async_high\n── asset_processing"]
        RMQ["RabbitMQ\n── async_bulk\n── sync_external"]
        DOC["Doctrine\n── async_low\n(fallback)"]
    end

    T --> W

    subgraph W["WORKER POOL"]
        W1[Worker 1\nhigh priority]
        W2[Worker 2\nhigh priority]
        W3[Worker 3\nbulk]
        WN[Worker N\nasset proc]
        DLQ[Dead Letter Queue\n+ Monitoring]
    end

I concetti fondamentali:

Message: un oggetto serializzabile che rappresenta un’operazione da eseguire (es. ProcessProductImportMessage)
Handler: la classe che esegue l’operazione quando il messaggio viene consumato
Transport: il sistema di storage/delivery dei messaggi (Redis, RabbitMQ, Doctrine)
Worker: il processo che consuma i messaggi da un transport e li passa agli handler
Envelope: il wrapper che contiene il messaggio più i metadata (stamps)

Il routing dei messaggi verso i transport avviene in base alla classe del messaggio:

flowchart LR
    MSG[Message] --> R{Router}
    R -->|HighPriority/*| RH[async_high\nRedis Stream]
    R -->|Bulk/*| RB[async_bulk\nRabbitMQ]
    R -->|Asset/*| RA[asset_processing\nRedis Stream]
    R -->|Sync/*| RS[sync_external\nRabbitMQ Topic]
    R -->|"*"| RH

    RH --> W1[Worker Pool\nHigh]
    RB --> W2[Worker Pool\nBulk]
    RA --> W3[Worker Pool\nAsset]
    RS --> W4[Worker Pool\nSync]

    W1 & W2 & W3 & W4 -->|max retries| DLQ[Dead Letter\nQueue]

Implementazione Passo-Passo

Configurazione Multi-Transport per Scenari Pimcore

La configurazione di default di Messenger non è adatta a un PIM enterprise. Dobbiamo definire transport separati per tipologie di operazioni diverse, con caratteristiche di performance e resilienza specifiche.

 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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
# config/packages/messenger.yaml
framework:
    messenger:
        # Serializzatore personalizzato per gestire entità Pimcore
        default_serializer: App\Messenger\Serializer\PimcoreMessageSerializer
        
        # Configurazione failure transport globale
        failure_transport: failed
        
        transports:
            # Transport ad alta priorità per operazioni critiche
            # Usa Redis Streams per bassa latenza
            async_high:
                dsn: '%env(REDIS_MESSENGER_DSN)%'
                options:
                    stream: pimcore_high_priority
                    group: pimcore_workers
                    consumer: '%env(HOSTNAME)%'
                    auto_setup: true
                    # Claim messaggi orfani dopo 5 minuti
                    claim_interval: 300000
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 2
                    max_delay: 30000
            
            # Transport per operazioni bulk (importazioni massive)
            async_bulk:
                dsn: '%env(RABBITMQ_MESSENGER_DSN)%'
                options:
                    exchange:
                        name: pimcore_bulk
                        type: direct
                    queues:
                        pimcore_bulk_queue:
                            binding_keys: [bulk]
                    prefetch_count: 5
                retry_strategy:
                    max_retries: 5
                    delay: 5000
                    multiplier: 3
                    max_delay: 3600000
            
            # Transport per elaborazione asset
            asset_processing:
                dsn: '%env(REDIS_MESSENGER_DSN)%'
                options:
                    stream: pimcore_assets
                    group: asset_workers
                    consumer: '%env(HOSTNAME)%'
                retry_strategy:
                    max_retries: 2
                    delay: 10000
                    multiplier: 2
            
            # Transport per sincronizzazione esterna
            sync_external:
                dsn: '%env(RABBITMQ_MESSENGER_DSN)%'
                options:
                    exchange:
                        name: pimcore_sync
                        type: topic
                    queues:
                        erp_sync:
                            binding_keys: ['sync.erp.*']
                        ecommerce_sync:
                            binding_keys: ['sync.ecommerce.*']
                retry_strategy:
                    max_retries: 10
                    delay: 60000
                    multiplier: 2
                    max_delay: 7200000
            
            failed:
                dsn: 'doctrine://default?queue_name=failed_messages'
            
            sync:
                dsn: 'sync://'
        
        routing:
            'App\Message\HighPriority\*': async_high
            'App\Message\Bulk\*': async_bulk
            'App\Message\Asset\*': asset_processing
            'App\Message\Sync\*': sync_external
            '*': async_high

⚠️ Attenzione: non usare mai sync:// in produzione per operazioni pesanti. È utile solo per test e debugging.

💡 Suggerimento: il claim_interval di Redis Streams è fondamentale. Se un worker crasha, i messaggi “pending” vengono reclamati da altri worker dopo questo intervallo. Impostalo in base al tempo massimo previsto per l’elaborazione di un messaggio.

1
2
3
# .env.local
REDIS_MESSENGER_DSN=redis://localhost:6379/messages?serializer=1
RABBITMQ_MESSENGER_DSN=amqp://guest:guest@localhost:5672/%2f/messages

Event Subscribers per Intercettare Operazioni CRUD

Il cuore di un sistema event-driven in Pimcore sono gli Event Subscribers che reagiscono alle operazioni su DataObjects e Assets.

  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
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
<?php
// src/EventSubscriber/DataObjectEventSubscriber.php

declare(strict_types=1);

namespace App\EventSubscriber;

use App\Message\Bulk\ReindexProductMessage;
use App\Message\Bulk\ResyncCategoryProductsMessage;
use App\Message\HighPriority\InvalidateCacheMessage;
use App\Message\Sync\SyncToErpMessage;
use App\Message\Sync\SyncToEcommerceMessage;
use Pimcore\Event\DataObjectEvents;
use Pimcore\Event\Model\DataObjectEvent;
use Pimcore\Model\DataObject\AbstractObject;
use Pimcore\Model\DataObject\Product;
use Pimcore\Model\DataObject\Category;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\DelayStamp;
use Symfony\Component\Messenger\Stamp\TransportNamesStamp;

final class DataObjectEventSubscriber implements EventSubscriberInterface
{
    private static bool $processingBatch = false;
    
    public function __construct(
        private readonly MessageBusInterface $messageBus,
        private readonly LoggerInterface $logger,
    ) {}
    
    public static function getSubscribedEvents(): array
    {
        return [
            DataObjectEvents::POST_UPDATE => ['onPostUpdate', 0],
            DataObjectEvents::POST_ADD => ['onPostAdd', 0],
            DataObjectEvents::POST_DELETE => ['onPostDelete', 0],
            'pimcore.dataobject.batch.complete' => ['onBatchComplete', -100],
        ];
    }
    
    public function onPostUpdate(DataObjectEvent $event): void
    {
        $object = $event->getObject();
        
        if (self::$processingBatch || !$object->isPublished()) {
            return;
        }
        
        if ($this->isInternalUpdate($event)) {
            return;
        }
        
        $this->dispatchForObject($object, 'update');
    }
    
    private function handleProductChange(
        Product $product, 
        string $operation,
        bool $lowPriority
    ): void {
        $stamps = $lowPriority ? [new DelayStamp(5000)] : [];
        
        // 1. Invalida cache immediatamente
        $this->messageBus->dispatch(
            new InvalidateCacheMessage(
                objectId: $product->getId(),
                objectType: 'Product',
                cacheKeys: $this->buildCacheKeys($product),
            ),
            [new TransportNamesStamp(['async_high'])]
        );
        
        // 2. Reindicizza per search
        $this->messageBus->dispatch(
            new ReindexProductMessage(
                productId: $product->getId(),
                operation: $operation,
            ),
            array_merge($stamps, [new TransportNamesStamp(['async_bulk'])])
        );
        
        // 3. Sincronizza con ERP
        if ($operation !== 'delete' && $product->getSyncToErp()) {
            $this->messageBus->dispatch(
                new SyncToErpMessage(
                    productId: $product->getId(),
                    sku: $product->getSku(),
                    operation: $operation,
                ),
                [new TransportNamesStamp(['sync_external'])]
            );
        }
    }
    
    public static function enableBatchMode(): void
    {
        self::$processingBatch = true;
    }
    
    public static function disableBatchMode(): void
    {
        self::$processingBatch = false;
    }
}

📝 Nota: il flag $processingBatch è statico intenzionalmente. Durante importazioni massive, vogliamo prevenire la generazione di migliaia di messaggi individuali.

Saga Pattern per Workflow Complessi

Le operazioni reali in un PIM/DAM non sono mai atomiche. Un’importazione prodotto tipica coinvolge più step interdipendenti:

sequenceDiagram
    participant C as CLI Command
    participant S as ProductPublishSaga
    participant MB as Message Bus
    participant H1 as CreateObjectHandler
    participant H2 as UploadAssetHandler
    participant H3 as GenerateThumbnailHandler
    participant H4 as SyncEcommerceHandler

    C->>S: startSaga(importData)
    S->>MB: dispatch(CreateProductMessage)
    MB->>H1: handle()
    H1-->>S: ProductCreatedEvent
    
    S->>MB: dispatch(UploadAssetMessage)
    MB->>H2: handle()
    H2-->>S: AssetUploadedEvent

    S->>MB: dispatch(GenerateThumbnailMessage)
    MB->>H3: handle()
    H3-->>S: ThumbnailReadyEvent

    S->>MB: dispatch(SyncToEcommerceMessage)
    MB->>H4: handle()
    H4-->>S: SyncCompletedEvent

    S-->>C: sagaCompleted
    
    note over S,H4: Se uno step fallisce → compensating transactions

 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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
<?php
// src/Saga/ProductPublishSaga.php

declare(strict_types=1);

namespace App\Saga;

use App\Message\Asset\GenerateThumbnailsMessage;
use App\Message\Bulk\ReindexProductMessage;
use App\Message\Sync\SyncToEcommerceMessage;
use App\Saga\State\ProductPublishState;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\MessageBusInterface;

final class ProductPublishSaga
{
    public function __construct(
        private readonly MessageBusInterface $messageBus,
        private readonly LoggerInterface $logger,
    ) {}

    public function start(int $productId, string $correlationId): ProductPublishState
    {
        $state = new ProductPublishState(
            productId: $productId,
            correlationId: $correlationId,
        );

        $this->messageBus->dispatch(
            new GenerateThumbnailsMessage($productId, $correlationId)
        );

        return $state;
    }

    public function onThumbnailsReady(ProductPublishState $state): void
    {
        $this->messageBus->dispatch(
            new ReindexProductMessage($state->productId, 'publish')
        );
    }

    public function onIndexed(ProductPublishState $state): void
    {
        $this->messageBus->dispatch(
            new SyncToEcommerceMessage(
                productId: $state->productId,
                channels: ['web', 'mobile'],
                operation: 'publish',
            )
        );
    }

    /**
     * Compensating transaction: rollback in caso di errore.
     */
    public function compensate(ProductPublishState $state, \Throwable $error): void
    {
        $this->logger->error('Saga compensation triggered', [
            'productId' => $state->productId,
            'correlationId' => $state->correlationId,
            'error' => $error->getMessage(),
        ]);

        // Unpublish, rimuovi asset parziali, notifica admin...
    }
}

Configurazione per Produzione

1
2
3
4
5
6
7
8
# config/packages/prod/messenger.yaml
framework:
    messenger:
        transports:
            async_high:
                options:
                    # In produzione: più consumer per parallelismo
                    consumer: '%env(POD_NAME)%'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# Avvia worker con supervisord
[program:messenger_high]
command=php bin/console messenger:consume async_high --limit=100 --memory-limit=256M
numprocs=3
autostart=true
autorestart=true

[program:messenger_bulk]
command=php bin/console messenger:consume async_bulk --limit=50 --memory-limit=512M
numprocs=2
autostart=true
autorestart=true

Errori Comuni e Troubleshooting

⚠️ Loop infiniti: se il tuo handler modifica un DataObject, scatena un nuovo evento, che dispatcha un nuovo messaggio, che viene consumato dall’handler… hai un loop. Usa sempre $processingBatch o un flag per identificare operazioni “interne”.

⚠️ Serializzazione di entità Pimcore: non mettere mai un DataObject direttamente nel messaggio. Pimcore usa lazy loading e proxy Doctrine — la serializzazione fallirà o produrrà oggetti inconsistenti. Passa sempre solo l’ID.

💡 Monitoring: usa messenger:stats e messenger:failed:show per monitorare code e messaggi falliti in produzione.

Performance e Scalabilità

Con questa architettura, un’importazione di 500.000 prodotti diventa:

Prima: 8 ore, server bloccato, admin inutilizzabile
Dopo: 45 minuti con 4 worker paralleli, admin completamente funzionante durante l’import

Il segreto è il prefetch_count: 5 su RabbitMQ — ogni worker prende solo 5 messaggi alla volta, distribuendo naturalmente il carico senza thundering herd.

Conclusioni e Next Steps

Abbiamo costruito un sistema di workflow asincrono production-ready per Pimcore che scala orizzontalmente e gestisce i fallimenti in modo resiliente. I concetti chiave da ricordare:

Usa transport separati per priorità diverse
Non serializzare entità Pimcore — solo ID
Il Saga Pattern è la soluzione per workflow multi-step
enableBatchMode() prima di qualsiasi importazione massiva

Come next steps, esplora:

Symfony Messenger docs
Pimcore Event System
Worker autoscaling con Kubernetes HPA basato sulla lunghezza della coda

Architettura Event-Driven Pimcore con Symfony Messenger