Oro Integration

OroCommerce v6.1 Integration Development

Message Queue: Async Processing

OroCommerce uses message queues to defer heavy work outside the HTTP request cycle. Producers send messages to topics, consumers process them asynchronously.

Processor Pattern

A processor implements MessageProcessorInterface and TopicSubscriberInterface. It must return self::ACK (success), self::REJECT (discard), or self::REQUEUE (retry). Never return void or null.

use Oro\Component\MessageQueue\Client\TopicSubscriberInterface;
use Oro\Component\MessageQueue\Consumption\MessageProcessorInterface;

class ProcessDocumentProcessor implements MessageProcessorInterface, TopicSubscriberInterface
{
    #[\Override]
    public function process(MessageInterface $message, SessionInterface $session): string
    {
        $data = json_decode($message->getBody(), true);
        // Process data...
        return self::ACK;
    }

    #[\Override]
    public static function getSubscribedTopics(): array
    {
        return ['acme_demo.document.process'];
    }
}

Register with the oro.message_queue.client.message_processor tag:

services:
    acme_demo.async.process_document:
        class: Acme\Bundle\DemoBundle\Async\Processor\ProcessDocumentProcessor
        arguments: ['@logger']
        tags:
            - { name: 'oro.message_queue.client.message_processor' }

For the full processor example, producer pattern, and message sending, see integration-patterns.md.

Import/Export

Oro's import/export uses a pipeline: Reader -> Processor -> Writer. Each processor handles one item at a time.

• Export: Entity -> Serializer normalizes -> DataConverter -> flat array (CSV row)
• Import: Flat array (CSV row) -> DataConverter -> Serializer denormalizes -> Strategy -> Entity

Most custom import/export needs only a DataConverter and service config — no custom processor class. Use oro_importexport.processor.export_abstract / oro_importexport.processor.import_abstract as parent services.

Import processors need both import and import_validation tags. Tag attributes: type, entity (FQCN), alias (unique identifier).

For complete DataConverter example, service registration YAML, custom processor class, and import strategy setup, see integration-patterns.md.

Cron Jobs

Create a command implementing CronCommandInterface:

use Oro\Bundle\CronBundle\Command\CronCommandInterface;
use Symfony\Component\Console\Attribute\AsCommand;

#[AsCommand(name: 'acme:cron:sync-documents')]
class SyncDocumentsCronCommand extends Command implements CronCommandInterface
{
    #[\Override]
    public function getDefaultDefinition(): string { return '0 */4 * * *'; }

    #[\Override]
    public function isActive(): bool { return true; }

    #[\Override]
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        // Perform sync work
        return Command::SUCCESS;
    }
}

Schedule format is standard Linux cron: minute hour dayOfMonth month dayOfWeek.

Key Pitfalls

1. Processor return values: Processors MUST return self::ACK, self::REJECT, or self::REQUEUE. Returning void or null causes messages to loop indefinitely.

2. DBAL polling: DBAL transport polls every second. Use AMQP (RabbitMQ) for production workloads.

3. Message serialization: Message bodies are JSON strings. Always json_decode() on receipt and json_encode() before sending.

For integration channels, transports, webhooks, common commands, and additional pitfalls, see integration-patterns.md.

Version Notes

For version-specific details, see v6.1 notes | v7.0 notes.

See also message-queue-config.md for MQ configuration options.

Integration Patterns Reference

Complete MQ Processor Example

A processor handles messages from a specific topic. It must implement MessageProcessorInterface and TopicSubscriberInterface:

namespace Acme\Bundle\DemoBundle\Async\Processor;

use Oro\Component\MessageQueue\Client\TopicSubscriberInterface;
use Oro\Component\MessageQueue\Consumption\MessageProcessorInterface;
use Oro\Component\MessageQueue\Transport\MessageInterface;
use Oro\Component\MessageQueue\Transport\SessionInterface;
use Psr\Log\LoggerInterface;

class ProcessDocumentProcessor implements
    MessageProcessorInterface,
    TopicSubscriberInterface
{
    private LoggerInterface $logger;

    public function __construct(LoggerInterface $logger)
    {
        $this->logger = $logger;
    }

    #[\Override]
    public function process(
        MessageInterface $message,
        SessionInterface $session
    ): string
    {
        $data = json_decode($message->getBody(), true);

        try {
            $documentId = $data['document_id'] ?? null;
            if (!$documentId) {
                $this->logger->error('Missing document_id');
                return self::REJECT;
            }

            // Process the document...
            $this->logger->info("Processing document {$documentId}");

            return self::ACK;
        } catch (\Exception $e) {
            $this->logger->error("Error processing document: {$e->getMessage()}");
            return self::REQUEUE;
        }
    }

    #[\Override]
    public static function getSubscribedTopics(): array
    {
        return ['acme_demo.document.process'];
    }
}

Producing Messages

Inject MessageProducerInterface and send messages:

namespace Acme\Bundle\DemoBundle\Service;

use Oro\Component\MessageQueue\Client\MessageProducerInterface;

class DocumentService
{
    public function __construct(
        private MessageProducerInterface $producer
    ) {}

    public function submitDocumentForProcessing(int $documentId): void
    {
        $this->producer->send(
            'acme_demo.document.process',
            json_encode(['document_id' => $documentId])
        );
    }
}

The producer accepts a topic name and a JSON string.

Data Converter Example

Map between external column names and entity field names. Implement DataConverterInterface:

namespace Acme\Bundle\DemoBundle\ImportExport\Converter;

use Oro\Bundle\ImportExportBundle\Converter\DataConverterInterface;

class DocumentDataConverter implements DataConverterInterface
{
    #[\Override]
    public function convertToExportFormat(array $exportedRecord, $skipNullValues = true): array
    {
        return [
            'Document Title' => $exportedRecord['title'] ?? '',
            'Created Date' => $exportedRecord['createdAt'] ?? '',
        ];
    }

    #[\Override]
    public function convertToImportFormat(array $importedRecord, $skipNullValues = true): array
    {
        return [
            'title' => $importedRecord['Document Title'] ?? '',
            'createdAt' => $importedRecord['Created Date'] ?? '',
        ];
    }
}

Import/Export Service Registration

Use abstract parent services — no custom processor class needed for standard cases:

services:
    # Data Converter
    acme_demo.importexport.data_converter.document:
        class: Acme\Bundle\DemoBundle\ImportExport\Converter\DocumentDataConverter

    # Export processor (uses parent abstract + your converter)
    acme_demo.importexport.processor.export.document:
        parent: oro_importexport.processor.export_abstract
        calls:
            - [setDataConverter, ['@acme_demo.importexport.data_converter.document']]
        tags:
            - { name: oro_importexport.processor, type: export, entity: 'Acme\Bundle\DemoBundle\Entity\Document', alias: acme_document }

    # Import processor (uses parent abstract + converter + strategy)
    acme_demo.importexport.processor.import.document:
        parent: oro_importexport.processor.import_abstract
        calls:
            - [setDataConverter, ['@acme_demo.importexport.data_converter.document']]
            - [setStrategy, ['@acme_demo.importexport.strategy.document.add_or_replace']]
        tags:
            - { name: oro_importexport.processor, type: import, entity: 'Acme\Bundle\DemoBundle\Entity\Document', alias: acme_document.add_or_replace }
            - { name: oro_importexport.processor, type: import_validation, entity: 'Acme\Bundle\DemoBundle\Entity\Document', alias: acme_document.add_or_replace }

    # Import strategy (usually extends the configurable add-or-replace)
    acme_demo.importexport.strategy.document.add_or_replace:
        parent: oro_importexport.strategy.configurable_add_or_replace

Tag attributes: type (export, import, import_validation, export_template), entity (FQCN), alias (unique identifier). Import processors need both import and import_validation tags.

Custom Processor (only when needed)

Only create a custom processor class when you need logic beyond what the DataConverter + Strategy provide. Override process($item) — the argument is a single entity (export) or single row array (import):

namespace Acme\Bundle\DemoBundle\ImportExport\Processor;

use Oro\Bundle\ImportExportBundle\Processor\ExportProcessor;

class DocumentExportProcessor extends ExportProcessor
{
    #[\Override]
    public function process($item)
    {
        // $item is a single Document entity
        $result = parent::process($item);

        // Add custom computed fields
        $result['full_path'] = $item->getCategory()?->getPath() . '/' . $item->getTitle();

        return $result;
    }
}

Integration Channels & Transports

An integration channel defines how OroCommerce connects to an external system (ERP, PIM, payment gateway, etc.). A transport implements the actual communication protocol.

Transport Implementation

namespace Acme\Bundle\DemoBundle\Transport;

use Oro\Bundle\IntegrationBundle\Transport\TransportInterface;
use Oro\Bundle\IntegrationBundle\Entity\Transport as TransportEntity;
use Oro\Bundle\IntegrationBundle\Provider\TransportSettingsInterface;

class MyApiTransport implements TransportInterface
{
    private $baseUrl;
    private $apiKey;

    #[\Override]
    public function init(TransportSettingsInterface $settings): void
    {
        $this->baseUrl = $settings->getSetting('base_url');
        $this->apiKey = $settings->getSetting('api_key');
    }

    #[\Override]
    public function getLabel(): string
    {
        return 'My External API';
    }

    #[\Override]
    public function getSettingsFormType(): string
    {
        return MyApiTransportSettingsFormType::class;
    }

    #[\Override]
    public function getSettingsEntityFQCN(): string
    {
        return MyApiTransportSettings::class;
    }

    public function sendRequest(string $endpoint, array $payload): array
    {
        // Call external API
        return $response;
    }
}

Register the transport:

services:
    acme_demo.transport.my_api:
        class: Acme\Bundle\DemoBundle\Transport\MyApiTransport
        tags:
            - { name: 'oro_integration.transport' }

Webhooks

OroCommerce does not provide a built-in webhook framework. To implement outbound webhooks (notify external systems on entity changes), use Doctrine lifecycle events or Oro's message queue:

1. Listen to Doctrine postPersist/postUpdate/postRemove events 2. Send a message to the MQ with the entity change data 3. A dedicated MQ processor dispatches the HTTP webhook call asynchronously

This avoids blocking the request and handles retries via the MQ's built-in retry mechanism.

Common Commands

# Consume messages from the queue
php bin/console oro:message-queue:consume

# Import data from a file
php bin/console oro:import-export:import /path/to/file.csv

# Export entities to a file
php bin/console oro:import-export:export Document

# List scheduled cron jobs
php bin/console oro:cron:definitions:load

# Run cron jobs manually
php bin/console oro:cron:run

Additional Pitfalls

1. Import processor entity class: Always implement getProcessedEntityClass(). It tells the import framework which entity the processor handles.

2. Integration transports: All TransportInterface methods must be implemented. Missing init(), getLabel(), or settings methods will cause runtime errors.

3. Cron timing: Cron expressions are evaluated by the server; ensure the cron daemon is running (php bin/console oro:cron:run).

Message Queue Configuration Reference — OroCommerce v6.1

Complete configuration guide for the Oro message queue system in v6.1. Covers transport setup, consumer options, and advanced features.

Transport Configuration

Configure the default transport and available transports in config/config.yml.

DBAL Transport (Default)

DBAL transport stores messages in a database table and polls for new messages.

oro_message_queue:
    transport:
        default: 'dbal'
        dbal:
            connection: default              # Symfony doctrine connection name
            table: oro_message_queue         # Database table
            polling_interval: 1000           # Poll interval in milliseconds
            time_to_live: 86400              # Message TTL in seconds (0 = no limit)

Parameters:

• connection — Doctrine connection to use. Defaults to default.
• table — Table name for message storage. Defaults to oro_message_queue.
• polling_interval — How often (ms) the consumer checks for messages. Lower = more responsive, higher = lower CPU. Default is 1000ms.
• time_to_live — Seconds before a message expires. Default 0 = no expiration. Expired messages are automatically purged.

Pros:

• No external broker required
• Works with any Doctrine-supported database
• Simple setup

Cons:

• Polling-based (latency ~1 second)
• Not suitable for high-volume production (> 100 msg/sec)
• Database load increases with message volume

AMQP Transport (Enterprise)

AMQP (RabbitMQ, etc.) uses a message broker for efficient async processing.

oro_message_queue:
    transport:
        default: 'amqp'
        amqp:
            host: localhost
            port: 5672
            user: guest
            password: guest
            vhost: /
            lazy: false
            ssl: false
            ssl_options:
                cert_file: /path/to/cert.pem
                key_file: /path/to/key.pem
                verify_peer: true

Parameters:

• host — RabbitMQ server hostname
• port — RabbitMQ port (5672 for non-SSL, 5671 for SSL)
• user / password — RabbitMQ authentication credentials
• vhost — Virtual host (default /)
• lazy — Create connection on first use (true) or immediately (false)
• ssl — Enable SSL/TLS
• ssl_options — SSL certificate and verification settings

Pros:

• Push-based (no polling)
• Efficient for high-volume messaging
• Distributed processing

Cons:

• Requires external RabbitMQ broker
• More complex infrastructure

Multiple Transports

Define multiple transports and switch between them:

oro_message_queue:
    transport:
        default: 'dbal'
        dbal:
            connection: default
            table: oro_message_queue
        amqp:
            host: rabbitmq.example.com
            user: guest
            password: guest

Switch the active transport by changing the default key or by configuration override in environment-specific configs.

Topic Configuration

Topics are message channels. Producers send to topics; consumers subscribe to topics.

Topic Definition

Topics are typically defined by processors that subscribe to them:

class MyProcessor implements TopicSubscriberInterface
{
    public static function getSubscribedTopics(): array
    {
        return ['acme_demo.my_topic'];
    }
}

OroCommerce registers processors via service tags, and topics are auto-discovered.

Topic Naming Convention

Use lowercase, dot-separated names:

• acme_demo.document.process
• oro_notification.send_notification
• my_integration.sync_products

Consumer Configuration

The message consumer consumes messages from the queue. Configure consumer behavior:

oro_message_queue:
    consume:
        time_limit: 900                      # Max run time in seconds
        memory_limit: 1024                   # Max memory in MB
        batch_size: 10                       # Messages per batch
        log_level: info                      # Logging level

Parameters:

• time_limit — Consumer exits after N seconds. Default 900 (15 min). Use with cron to keep consumer fresh.
• memory_limit — Consumer exits if memory usage exceeds N MB. Prevents memory leaks. Default 1024 MB.
• batch_size — How many messages to process before committing (if broker supports). Default 10.
• log_level — Logging level (debug, info, warning, error). Default info.

Consumer Command Options

Pass options to the consumer command:

# Run for 60 seconds, then exit
php bin/console oro:message-queue:consume --time-limit=60

# Run with specific memory limit
php bin/console oro:message-queue:consume --memory-limit=512

# Verbose logging
php bin/console oro:message-queue:consume -vv

# Consume from specific transport
php bin/console oro:message-queue:consume --transport=amqp

Delayed Messages

Some transports support delayed message delivery. This is useful for retry logic or scheduled processing.

use Oro\Component\MessageQueue\Client\Message;

$message = new Message(json_encode(['data' => 'value']));
$message->setDelay(300); // Delay 300 seconds (5 minutes)

$producer->send('acme_demo.my_topic', $message);

Note: DBAL transport doesn't natively support delayed messages. Use a custom delay mechanism or upgrade to AMQP.

Job Processors

Some processors handle long-running jobs. Configure job processor settings:

oro_message_queue:
    job:
        max_jobs: 50                         # Max concurrent job processors
        job_timeout: 3600                    # Job timeout in seconds (1 hour)

A job is a special message type that tracks status and progress. Useful for import/export and batch operations.

Processors & Error Handling

Processor Return Values

A processor must return one of:

// Success — message removed from queue
return self::ACK;

// Failed — message discarded (not retried)
return self::REJECT;

// Transient failure — message re-queued for retry
return self::REQUEUE;

Retry Logic

When a processor returns REQUEUE, the message goes back to the queue. The consumer will retry it.

For exponential backoff or max retries, implement in the processor:

public function process(MessageInterface $message, SessionInterface $session): string
{
    $data = json_decode($message->getBody(), true);
    $retries = $data['retries'] ?? 0;

    if ($retries > 5) {
        return self::REJECT; // Give up after 5 retries
    }

    try {
        // Process...
        return self::ACK;
    } catch (\Exception $e) {
        $data['retries'] = $retries + 1;
        $this->producer->send($topic, json_encode($data));
        return self::ACK; // Acknowledge the current message, re-send with retries
    }
}

Dead Letter Queue (DLQ)

Some brokers (RabbitMQ) support dead letter exchanges. When a message is rejected multiple times, it moves to a DLQ for manual inspection.

DBAL transport doesn't have a DLQ. Rejected messages are permanently deleted. Use custom logging if you need to audit rejections.

Monitoring & Debugging

Queue Size

Check how many messages are pending:

php bin/console oro:message-queue:status

Consumer Logs

The consumer logs all activity. Monitor the log file:

tail -f var/logs/prod.log | grep message.queue

Processor Discovery

List registered processors:

php bin/console debug:container --tag=oro.message_queue.client.message_processor

Cron Integration

Run the consumer as a background service. One approach is a cron job that restarts the consumer periodically:

services:
    oro_cron.message_queue.start:
        class: Oro\Bundle\CronBundle\Command\CronCommand
        tags:
            - { name: 'oro_cron_command', cron: '*/5 * * * *' }

This restarts the consumer every 5 minutes, clearing memory leaks and refreshing connections.

Alternatively, use systemd or supervisor to keep the consumer running continuously.

Testing Message Queue

In tests, use a synchronous transport to process messages immediately:

# config/config_test.yml
oro_message_queue:
    transport:
        default: 'sync'

The sync transport processes messages inline without queueing. Useful for unit/integration tests.

Performance Tuning

DBAL Tuning

For high-volume DBAL usage:

• Increase polling_interval to reduce database load (trade-off: higher latency)
• Add database indexes on created_at and processed columns
• Periodically purge old messages (set time_to_live)
• Use a separate database connection if messages are high-volume

AMQP Tuning

For RabbitMQ:

• Configure multiple consumer instances (parallel processing)
• Use prefetch limits to balance load across consumers
• Monitor broker memory and disk usage
• Enable durable queues for persistence

Configuration Examples

Development Setup (DBAL)

oro_message_queue:
    transport:
        default: 'dbal'
        dbal:
            connection: default
            table: oro_message_queue
            polling_interval: 1000
            time_to_live: 3600
    consume:
        time_limit: 60
        memory_limit: 512

Production Setup (AMQP)

oro_message_queue:
    transport:
        default: 'amqp'
        amqp:
            host: '%env(MQ_HOST)%'
            port: '%env(MQ_PORT)%'
            user: '%env(MQ_USER)%'
            password: '%env(MQ_PASSWORD)%'
            vhost: '/'
    consume:
        time_limit: 3600
        memory_limit: 2048
        batch_size: 50

Use environment variables for sensitive data (host, credentials).

Troubleshooting

Consumer Hangs

If the consumer process doesn't respond:

pkill -f 'oro:message-queue:consume'

Check for deadlocked processors. Review logs for errors.

Message Not Processing

1. Verify the processor is registered: debug:container --tag=oro.message_queue.client.message_processor 2. Check that the topic name matches: processor's getSubscribedTopics() should match the producer's topic 3. Ensure the consumer is running: ps aux | grep oro:message-queue:consume 4. Check logs for exceptions in the processor

Queue Backlog

If messages accumulate faster than they're processed:

• Increase batch_size in consumer config
• Add more consumer instances (parallel processing)
• Optimize processor logic (reduce processing time)
• Switch to AMQP for better throughput

Memory Leaks

If consumer memory grows unbounded:

• Set memory_limit to force restarts
• Review processor code for resource leaks (unclosed connections, etc.)
• Use cron-based restart strategy (restart consumer every 15 min)

Integration — v6.1 Notes

Transport Defaults

v6.1 defaults to DBAL transport (database polling). AMQP/RabbitMQ is available but requires separate broker infrastructure. For production with high message volume (> 100 msg/sec), use AMQP.

Known Issues

DBAL Message Queue Latency

DBAL transport polls every ~1 second (polling_interval). Messages have 0-1s latency before processing begins. For time-critical processing, use AMQP or increase polling frequency (trade-off: higher DB load).

Large Batch Imports Timeout

Imports of 1000+ records may timeout with slow processors. Mitigations: 1. Implement pagination in the import processor 2. Break imports into smaller batches 3. Use message queue to defer processing 4. Increase PHP max_execution_time and memory_limit

Transport Connection Pooling

DBAL transport uses a single connection. Many concurrent consumers can exhaust the connection pool. Limit concurrent consumers or upgrade to AMQP.

Import/Export Memory Usage

Bulk exports load all entities into memory. Use pagination/batching or implement streaming export for large datasets.

API Stability

Stable (unlikely to break in v7.0):

• MessageProcessorInterface, TopicSubscriberInterface, MessageProducerInterface
• TransportInterface, ImportProcessor / ExportProcessor

Semi-stable (may have migration paths):

• DataConverterInterface, ContextInterface, integration transport settings

Unstable / internal (avoid direct use):

• MessageInterface, SessionInterface (transport-specific), internal queue implementation details

Performance Characteristics

Import speed: 100-1000 records/second depending on entity complexity.

Monitoring

• Queue status: php bin/console oro:message-queue:status
• Cron definitions: php bin/console oro:cron:definitions:load
• MQ debug logging: configure a monolog handler on channel oro.message_queue

Integration — v7.0 Notes

v7.0 is not yet released. This file will be updated when v7.0 stabilizes.

Expected Changes

• TBD