PublisherConsumerTest.php

Go to the documentation of this file.

<?php
namespace Magento\MysqlMq\Model;

use Magento\Framework\MessageQueue\PublisherInterface;

class PublisherConsumerTest extends \PHPUnit\Framework\TestCase
{
    const MAX_NUMBER_OF_TRIALS = 3;

    protected $publisher;

    protected $objectManager;

    protected function setUp()
    {
        $this->markTestIncomplete('Should be converted to queue config v2.');
        $this->objectManager = \Magento\TestFramework\Helper\Bootstrap::getObjectManager();

        $configPath = __DIR__ . '/../etc/queue.xml';
        $fileResolverMock = $this->createMock(\Magento\Framework\Config\FileResolverInterface::class);
        $fileResolverMock->expects($this->any())
            ->method('get')
            ->willReturn([$configPath => file_get_contents(($configPath))]);

        $xmlReader = $this->objectManager->create(
            \Magento\Framework\MessageQueue\Config\Reader\Xml::class,
            ['fileResolver' => $fileResolverMock]
        );

        $newData = $xmlReader->read();

        $configData = $this->objectManager->get(\Magento\Framework\MessageQueue\Config\Data::class);
        $configData->reset();
        $configData->merge($newData);

        $this->publisher = $this->objectManager->create(\Magento\Framework\MessageQueue\PublisherInterface::class);
    }

    protected function tearDown()
    {
        $this->markTestIncomplete('Should be converted to queue config v2.');
        $this->consumeMessages('demoConsumerQueueOne', PHP_INT_MAX);
        $this->consumeMessages('demoConsumerQueueTwo', PHP_INT_MAX);
        $this->consumeMessages('demoConsumerQueueThree', PHP_INT_MAX);
        $this->consumeMessages('demoConsumerQueueFour', PHP_INT_MAX);
        $this->consumeMessages('demoConsumerQueueFive', PHP_INT_MAX);
        $this->consumeMessages('demoConsumerQueueOneWithException', PHP_INT_MAX);

        $objectManagerConfiguration = [\Magento\Framework\MessageQueue\Config\Reader\Xml::class => [
                'arguments' => [
                    'fileResolver' => ['instance' => \Magento\Framework\Config\FileResolverInterface::class],
                ],
            ],
        ];
        $this->objectManager->configure($objectManagerConfiguration);
        $queueConfig = $this->objectManager->get(\Magento\Framework\MessageQueue\Config\Data::class);
        $queueConfig->reset();
    }

    public function testPublishConsumeFlow()
    {
        $objectFactory = $this->objectManager->create(\Magento\MysqlMq\Model\DataObjectFactory::class);
        $object = $objectFactory->create();
        for ($i = 0; $i < 10; $i++) {
            $object->setName('Object name ' . $i)->setEntityId($i);
            $this->publisher->publish('demo.object.created', $object);
        }
        for ($i = 0; $i < 5; $i++) {
            $object->setName('Object name ' . $i)->setEntityId($i);
            $this->publisher->publish('demo.object.updated', $object);
        }
        for ($i = 0; $i < 3; $i++) {
            $object->setName('Object name ' . $i)->setEntityId($i);
            $this->publisher->publish('demo.object.custom.created', $object);
        }

        $outputPattern = '/(Processed \d+\s)/';
        $this->consumeMessages('demoConsumerQueueOne', 7, 7, $outputPattern);
        $this->consumeMessages('demoConsumerQueueOne', PHP_INT_MAX, 3, $outputPattern);
        $this->consumeMessages('demoConsumerQueueOne', 7, 0, $outputPattern);

        $this->consumeMessages('demoConsumerQueueTwo', 20, 15, $outputPattern);

        $this->consumeMessages('demoConsumerQueueFour', 11, 0, $outputPattern);

        $this->consumeMessages('demoConsumerQueueThree', 20, 15, $outputPattern);

        $this->consumeMessages('demoConsumerQueueFive', 20, 18, $outputPattern);
    }

    public function testPublishAndConsumeWithFailedJobs()
    {
        $objectFactory = $this->objectManager->create(\Magento\MysqlMq\Model\DataObjectFactory::class);
        $object = $objectFactory->create();
        for ($i = 0; $i < 5; $i++) {
            $object->setName('Object name ' . $i)->setEntityId($i);
            $this->publisher->publish('demo.object.created', $object);
        }
        $outputPattern = '/(Processed \d+\s)/';
        for ($i = 0; $i < self::MAX_NUMBER_OF_TRIALS; $i++) {
            $this->consumeMessages('demoConsumerQueueOneWithException', PHP_INT_MAX, 0, $outputPattern);
        }
        $this->consumeMessages('demoConsumerQueueOne', PHP_INT_MAX, 0, $outputPattern);

        for ($i = 0; $i < 5; $i++) {
            $object->setName('Object name ' . $i)->setEntityId($i);
            $this->publisher->publish('demo.object.created', $object);
        }
        for ($i = 0; $i < self::MAX_NUMBER_OF_TRIALS + 1; $i++) {
            $this->consumeMessages('demoConsumerQueueOneWithException', PHP_INT_MAX, 0, $outputPattern);
        }
        $this->consumeMessages('demoConsumerQueueOne', PHP_INT_MAX, 0, $outputPattern);
    }

    public function testPublishAndConsumeSchemaDefinedByMethod()
    {
        $objectFactory = $this->objectManager->create(\Magento\MysqlMq\Model\DataObjectFactory::class);
        $object = $objectFactory->create();
        $id = 33;
        $object->setName('Object name ' . $id)->setEntityId($id);
        $requiredStringParam = 'Required value';
        $optionalIntParam = 44;
        $this->publisher->publish('test.schema.defined.by.method', [$object, $requiredStringParam, $optionalIntParam]);
        $outputPattern = "/Processed '{$object->getEntityId()}'; "
            . "Required param '{$requiredStringParam}'; Optional param '{$optionalIntParam}'/";
        $this->consumeMessages('delayedOperationConsumer', PHP_INT_MAX, 1, $outputPattern);
    }

    protected function consumeMessages(
        $consumerName,
        $messagesToProcess,
        $expectedNumberOfProcessedMessages = null,
        $outputPattern = null
    ) {
        $consumerFactory = $this->objectManager->create(\Magento\Framework\MessageQueue\ConsumerFactory::class);
        $consumer = $consumerFactory->get($consumerName);
        ob_start();
        $consumer->process($messagesToProcess);
        $consumersOutput = ob_get_contents();
        ob_end_clean();
        if ($outputPattern) {
            $this->assertEquals(
                $expectedNumberOfProcessedMessages,
                preg_match_all($outputPattern, $consumersOutput)
            );
        }
    }
}