Wzorzec Command – kolejka komend, undo/redo, Magento 2 MessageQueue

Wzorzec Command enkapsuluje żądanie jako obiekt, co pozwala parametryzować operacje, kolejkować je, logować i implementować cofanie. W Magento 2 ten wzorzec jest podstawą systemu kolejkowania wiadomości (MessageQueue) – każda wiadomość to Command który jest wykonywany asynchronicznie przez consumer. Pokażę implementację od klasycznej wersji po integrację z kolejką Magento.

Klasyczna implementacja

<?php

declare(strict_types=1);

// Interfejs Command
interface CommandInterface
{
    public function execute(): void;
    public function undo(): void;
}

// Receiver - obiekt który faktycznie wykonuje pracę
class ProductPriceManager
{
    private array $priceHistory = [];

    public function setPrice(string $sku, float $price): void
    {
        $this->priceHistory[$sku][] = $this->getPrice($sku);
        echo "Ustawiam cenę {$sku} na {$price} PLN\n";
        // W prawdziwym kodzie: zapis do DB
    }

    public function getPrice(string $sku): float
    {
        // Symulacja pobierania z DB
        return 99.99;
    }

    public function getPreviousPrice(string $sku): ?float
    {
        $history = $this->priceHistory[$sku] ?? [];
        return empty($history) ? null : end($history);
    }
}

// Konkretny Command
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
    {
        // Cofaj w odwrotnej kolejności
        foreach (array_reverse($this->commands) as $cmd) {
            $cmd->undo();
        }
    }
}

// Invoker z historią - umożliwia undo/redo
class CommandHistory
{
    /** @var CommandInterface[] */
    private array $history = [];
    private int $position  = -1;

    public function execute(CommandInterface $command): void
    {
        // Usuń "przyszłość" jeśli cofnęliśmy się i wykonujemy nową akcję
        $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;
    }
}

// Użycie
$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(); // cofnij bulk discount
$history->undo(); // cofnij zmianę ceny SKU-001
$history->redo(); // ponów zmianę ceny SKU-001

Command jako DTO – kolejka Magento 2

<?php

declare(strict_types=1);

namespace Vendor\Module\Model\Message;

// W Magento 2 MessageQueue Command to DTO (Data Transfer Object)
// Serializowany do JSON i wysyłany przez broker (RabbitMQ lub MySQL)

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     = 'pl_PL',
    ) {}
}

Publisher – wysyłanie wiadomości do kolejki

<?php

declare(strict_types=1);

namespace Vendor\Module\Model;

use Magento\Framework\MessageQueue\PublisherInterface;
use Vendor\Module\Model\Message\ReindexProductMessage;

class ProductUpdateService
{
    // Nazwa kolejki - zdefiniowana w communication.xml
    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,
        );

        // Wiadomość trafia do kolejki - consumer wykona ją asynchronicznie
        $this->publisher->publish(self::REINDEX_TOPIC, $message);
    }

    public function scheduleReindexBatch(array $skus): void
    {
        foreach ($skus as $sku) {
            $this->scheduleReindex($sku, 'batch_update');
        }
    }
}

Consumer – odbieranie i wykonywanie komend

<?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,
    ) {}

    // Magento wywołuje tę metodę dla każdej wiadomości z kolejki
    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; // wiadomość wróci do kolejki (retry)
            }
        }
    }

    private function getProductId(string $sku): int
    {
        // pobierz ID produktu po SKU
        return 1; // uproszczone
    }
}

Konfiguracja XML

<!-- 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>

# Uruchom consumer
bin/magento queue:consumers:start vendorProductReindex

# Lub w DDEV
ddev exec bin/magento queue:consumers:start vendorProductReindex --max-messages=100

Podsumowanie

Command enkapsuluje operację jako obiekt – co pozwala na kolejkowanie, logowanie i undo/redo. W Magento 2 wzorzec ten realizowany jest przez MessageQueue: Publisher wysyła DTO do brokera, Consumer odbiera i wykonuje. Kluczowe elementy konfiguracji: communication.xml (definicja topic), queue_consumer.xml (definicja consumer), queue_topology.xml (routing w RabbitMQ). Rzucenie wyjątku w consumer powoduje ponowne przetworzenie wiadomości – pamiętaj o idempotentności operacji.