The Command pattern encapsulates a request as an object, allowing operations to be parameterised, queued, logged and undone. In Magento 2 this pattern is the foundation of the MessageQueue system – each message is a Command executed asynchronously by a consumer. I show the implementation from the classic version through to integration with the Magento queue.
Classic implementation
<?php
declare(strict_types=1);
interface CommandInterface
{
public function execute(): void;
public function undo(): void;
}
class ProductPriceManager
{
private array $priceHistory = [];
public function setPrice(string $sku, float $price): void
{
$this->priceHistory[$sku][] = $this->getPrice($sku);
echo "Setting price {$sku} to {$price}\n";
}
public function getPrice(string $sku): float { return 99.99; }
}
class SetProductPriceCommand implements CommandInterface
{
private ?float $previousPrice = null;
public function __construct(
private ProductPriceManager $manager,
private string $sku,
private float $newPrice,
) {}
public function execute(): void
{
$this->previousPrice = $this->manager->getPrice($this->sku);
$this->manager->setPrice($this->sku, $this->newPrice);
}
public function undo(): void
{
if ($this->previousPrice !== null) {
$this->manager->setPrice($this->sku, $this->previousPrice);
}
}
}
class BulkDiscountCommand implements CommandInterface
{
/** @var CommandInterface[] */
private array $commands = [];
public function __construct(
private ProductPriceManager $manager,
private array $skus,
private float $discountPercent,
) {}
public function execute(): void
{
foreach ($this->skus as $sku) {
$currentPrice = $this->manager->getPrice($sku);
$newPrice = $currentPrice * (1 - $this->discountPercent / 100);
$cmd = new SetProductPriceCommand($this->manager, $sku, $newPrice);
$cmd->execute();
$this->commands[] = $cmd;
}
}
public function undo(): void
{
foreach (array_reverse($this->commands) as $cmd) {
$cmd->undo();
}
}
}
class CommandHistory
{
/** @var CommandInterface[] */
private array $history = [];
private int $position = -1;
public function execute(CommandInterface $command): void
{
$this->history = array_slice($this->history, 0, $this->position + 1);
$command->execute();
$this->history[] = $command;
$this->position++;
}
public function undo(): bool
{
if ($this->position < 0) return false;
$this->history[$this->position]->undo();
$this->position--;
return true;
}
public function redo(): bool
{
if ($this->position >= count($this->history) - 1) return false;
$this->position++;
$this->history[$this->position]->execute();
return true;
}
}
$manager = new ProductPriceManager();
$history = new CommandHistory();
$history->execute(new SetProductPriceCommand($manager, 'SKU-001', 149.99));
$history->execute(new BulkDiscountCommand($manager, ['SKU-001', 'SKU-002', 'SKU-003'], 10.0));
$history->undo();
$history->undo();
$history->redo();
Command as DTO – Magento 2 queue
<?php
declare(strict_types=1);
namespace Vendor\Module\Model\Message;
class ReindexProductMessage
{
public function __construct(
public readonly string $sku,
public readonly array $indexers = ['catalog_product_price', 'catalogsearch_fulltext'],
public readonly string $reason = '',
) {}
}
class SendOrderEmailMessage
{
public function __construct(
public readonly int $orderId,
public readonly string $templateId = 'sales_email_order_template',
public readonly string $locale = 'en_US',
) {}
}
Publisher – sending messages to the queue
<?php
declare(strict_types=1);
namespace Vendor\Module\Model;
use Magento\Framework\MessageQueue\PublisherInterface;
use Vendor\Module\Model\Message\ReindexProductMessage;
class ProductUpdateService
{
private const REINDEX_TOPIC = 'vendor.product.reindex';
public function __construct(
private PublisherInterface $publisher,
) {}
public function scheduleReindex(string $sku, string $reason = ''): void
{
$message = new ReindexProductMessage(sku: $sku, reason: $reason);
$this->publisher->publish(self::REINDEX_TOPIC, $message);
}
public function scheduleReindexBatch(array $skus): void
{
foreach ($skus as $sku) {
$this->scheduleReindex($sku, 'batch_update');
}
}
}
Consumer – receiving and executing commands
<?php
declare(strict_types=1);
namespace Vendor\Module\Model\Consumer;
use Vendor\Module\Model\Message\ReindexProductMessage;
use Magento\Indexer\Model\IndexerFactory;
use Psr\Log\LoggerInterface;
class ReindexProductConsumer
{
public function __construct(
private IndexerFactory $indexerFactory,
private LoggerInterface $logger,
) {}
public function process(ReindexProductMessage $message): void
{
$this->logger->info('Reindexing product', [
'sku' => $message->sku,
'reason' => $message->reason,
'indexers' => $message->indexers,
]);
foreach ($message->indexers as $indexerId) {
try {
$indexer = $this->indexerFactory->create();
$indexer->load($indexerId);
$indexer->reindexRow($this->getProductId($message->sku));
} catch (\Exception $e) {
$this->logger->error('Reindex failed', [
'sku' => $message->sku,
'indexer' => $indexerId,
'error' => $e->getMessage(),
]);
throw $e; // message returns to queue (retry)
}
}
}
private function getProductId(string $sku): int { return 1; }
}
XML configuration
<!-- etc/communication.xml -->
<config>
<topic name="vendor.product.reindex"
request="Vendor\Module\Model\Message\ReindexProductMessage"/>
</config>
<!-- etc/queue_consumer.xml -->
<config>
<consumer name="vendorProductReindex"
queue="vendor.product.reindex"
connection="amqp"
handler="Vendor\Module\Model\Consumer\ReindexProductConsumer::process"
maxMessages="100"/>
</config>
<!-- etc/queue_topology.xml -->
<config>
<exchange name="magento" type="topic" connection="amqp">
<binding id="vendorProductReindex"
topic="vendor.product.reindex"
destinationType="queue"
destination="vendor.product.reindex"/>
</exchange>
</config>
bin/magento queue:consumers:start vendorProductReindex # or in DDEV: ddev exec bin/magento queue:consumers:start vendorProductReindex --max-messages=100
Summary
Command encapsulates an operation as an object – enabling queuing, logging and undo/redo. In Magento 2 this pattern is implemented through MessageQueue: Publisher sends a DTO to the broker, Consumer receives and executes it. Key configuration elements: communication.xml (topic definition), queue_consumer.xml (consumer definition), queue_topology.xml (RabbitMQ routing). Throwing an exception in the consumer causes the message to be reprocessed – remember to make operations idempotent.
