Init repository

Author: Bartosz Kubicki

.gitignore

@@ -0,0 +1,6 @@
+.idea
+build/
+vendor/
+coverage/
+.env
+docker-compose.yml

Api/Envelope/RetryLimitOverflowResolverInterface.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * File: RetryLimitOverflowResolverInterface.php
+ *
+ * @author Bartosz Kubicki bartosz.kubicki@lizardmedia.pl>
+ * @copyright Copyright (C) 2020 Lizard Media (http://lizardmedia.pl)
+ */
+
+namespace LizardMedia\MessageQueue\Api\Envelope;
+
+use Magento\Framework\MessageQueue\EnvelopeInterface;
+use Magento\Framework\MessageQueue\QueueInterface;
+
+/**
+ * Interface RetryLimitOverflowResolverInterface
+ * @package LizardMedia\MessageQueue\Api\Envelope
+ */
+interface RetryLimitOverflowResolverInterface
+{
+    /**
+     * @param EnvelopeInterface $envelope
+     * @param string $queueName
+     * @return bool
+     */
+    public function isLimitReached(EnvelopeInterface $envelope, string $queueName): bool;
+}

Api/Queue/Consumer/EnvelopeCallbackFactoryInterface.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * File: EnvelopeCallbackFactoryInterface.php
+ *
+ * @author Bartosz Kubicki bartosz.kubicki@lizardmedia.pl>
+ * @copyright Copyright (C) 2020 Lizard Media (http://lizardmedia.pl)
+ */
+
+namespace LizardMedia\MessageQueue\Api\Queue\Consumer;
+
+use InvalidArgumentException;
+use LizardMedia\MessageQueue\Queue\Consumer\EnvelopeCallback\EnvelopeCallbackInterface;
+use Magento\Framework\MessageQueue\ConsumerConfigurationInterface as UsedConsumerConfig;
+
+/**
+ * Interface EnvelopeCallbackFactoryInterface
+ * @package LizardMedia\MessageQueue\Api\Queue\Consumer
+ */
+interface EnvelopeCallbackFactoryInterface
+{
+    /**
+     * @param string $type
+     * @param UsedConsumerConfig $usedConsumerConfiguration
+     * @return EnvelopeCallbackInterface
+     * @throws InvalidArgumentException
+     */
+    public function create(string $type, UsedConsumerConfig $usedConsumerConfiguration): EnvelopeCallbackInterface;
+}

CHANGELOG.md

@@ -0,0 +1,5 @@
+### 1.0.0 ###
+* custom implementation of `Magento\Framework\MessageQueue\ConsumerInterface` making possible injection of envelope callback,
+which allows to introduce custom message consumption easily without copy-paste of whole class
+* a few implementations of `LizardMedia\MessageQueue\Queue\Consumer\EnvelopeCallback\EnvelopeCallbackInterface`, each handling
+message in its specific way, including `x-death` parameters support

Console/Command/PutPoisonPillCommand.php

@@ -0,0 +1,66 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * File: PutPoisonPillCommand.php
+ *
+ * @author Bartosz Kubicki bartosz.kubicki@lizardmedia.pl>
+ * @copyright Copyright (C) 2020 Lizard Media (http://lizardmedia.pl)
+ */
+
+namespace LizardMedia\MessageQueue\Console\Command;
+
+use Exception;
+use Magento\Framework\Console\Cli;
+use Magento\Framework\MessageQueue\PoisonPill\PoisonPillPutInterface;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+/**
+ * Class PutPoisonPillCommand
+ * @package LizardMedia\MessageQueue\Console\Command
+ * @codeCoverageIgnore
+ */
+class PutPoisonPillCommand extends Command
+{
+    /**
+     * @var PoisonPillPutInterface
+     */
+    private $poisonPillPut;
+
+    /**
+     * PutPoisonPill constructor.
+     * @param PoisonPillPutInterface $poisonPillPut
+     * @param string|null $name
+     */
+    public function __construct(PoisonPillPutInterface $poisonPillPut, string $name = null)
+    {
+        parent::__construct($name);
+        $this->poisonPillPut = $poisonPillPut;
+    }
+
+    /**
+     * @return void
+     */
+    protected function configure(): void
+    {
+        $this->setName('lm:queue:consumers:poison')
+            ->setDescription('Poison queue consumers');
+    }
+
+    /**
+     * @param InputInterface $input
+     * @param OutputInterface $output
+     * @return int
+     * @throws Exception
+     */
+    public function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $poisonPillVersion = $this->poisonPillPut->put();
+        $output->writeln('Queue consumers have been poisoned...');
+        $output->writeln(sprintf('New Poison Pill Version: %s', $poisonPillVersion));
+        return Cli::RETURN_SUCCESS;
+    }
+}

Envelope/RetryLimitOverflowResolver.php

@@ -0,0 +1,106 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * File: RetryLimitOverflowResolver.php
+ *
+ * @author Bartosz Kubicki bartosz.kubicki@lizardmedia.pl>
+ * @copyright Copyright (C) 2020 Lizard Media (http://lizardmedia.pl)
+ */
+
+namespace LizardMedia\MessageQueue\Envelope;
+
+use Magento\Framework\MessageQueue\QueueInterface;
+use function array_key_exists;
+use LizardMedia\MessageQueue\Api\Envelope\RetryLimitOverflowResolverInterface;
+use Magento\Framework\MessageQueue\EnvelopeInterface;
+use PhpAmqpLib\Wire\AMQPTable;
+
+/**
+ * Class RetryLimitOverflowResolver
+ * @package LizardMedia\MessageQueue\Envelope
+ */
+class RetryLimitOverflowResolver implements RetryLimitOverflowResolverInterface
+{
+    /**
+     * @var string
+     */
+    private const APPLICATION_HEADERS_KEY = 'application_headers';
+
+    /**
+     * @var string
+     */
+    private const X_DEATH_KEY = 'x-death';
+
+    /**
+     * @var string
+     */
+    private const QUEUE_KEY = 'queue';
+
+    /**
+     * @var string
+     */
+    private const X_DEATH_COUNT_KEY = 'count';
+
+    /**
+     * @var int
+     */
+    private $limit;
+
+    /**
+     * RetryLimitOverflowResolver constructor.
+     * @param int $limit
+     */
+    public function __construct(int $limit = 3)
+    {
+        $this->limit = $limit;
+    }
+
+    /**
+     * @param EnvelopeInterface $envelope
+     * @param string $queueName
+     * @return bool
+     */
+    public function isLimitReached(EnvelopeInterface $envelope, string $queueName): bool
+    {
+        $xdeathParams = $this->getXDeathParameters($envelope, $queueName);
+        return !empty($xdeathParams[self::X_DEATH_COUNT_KEY])
+            && (int) $xdeathParams[self::X_DEATH_COUNT_KEY] > $this->limit;
+    }
+
+    /**
+     * @param EnvelopeInterface $envelope
+     * @param string $queueName
+     * @return array
+     */
+    private function getXDeathParameters(EnvelopeInterface $envelope, string $queueName): array
+    {
+        $properties = $envelope->getProperties();
+        if (!array_key_exists(self::APPLICATION_HEADERS_KEY, $properties)) {
+            return [];
+        }
+
+        /** @var $applicationHeaders AMQPTable */
+        $applicationHeaders = $properties[self::APPLICATION_HEADERS_KEY];
+        if ($applicationHeaders === null) {
+            return [];
+        }
+
+        $data = $applicationHeaders->getNativeData();
+
+        if (empty($data[self::X_DEATH_KEY])) {
+            return [];
+        }
+
+        $xdeathParamsArray = $data[self::X_DEATH_KEY];
+
+        foreach ($xdeathParamsArray as $key => $queueXdeathParams) {
+            if (isset($queueXdeathParams[self::QUEUE_KEY]) && $queueXdeathParams[self::QUEUE_KEY] === $queueName) {
+                return $xdeathParamsArray[$key];
+            }
+        }
+
+        return [];
+    }
+}

LICENSE.md

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2020 Lizard Media | UX & software house
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Queue/Consumer/ConsumerWithInjectableEnvelopeCallback.php

@@ -0,0 +1,94 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * File: ConsumerWithInjectableEnvelopeCallback.php
+ *
+ * @author Bartosz Kubicki bartosz.kubicki@lizardmedia.pl>
+ * @copyright Copyright (C) 2020 Lizard Media (http://lizardmedia.pl)
+ */
+
+namespace LizardMedia\MessageQueue\Queue\Consumer;
+
+use Closure;
+use LizardMedia\MessageQueue\Api\Queue\Consumer\EnvelopeCallbackFactoryInterface;
+use Magento\Framework\MessageQueue\CallbackInvokerInterface;
+use Magento\Framework\MessageQueue\ConsumerConfigurationInterface as UsedConsumerConfig;
+use Magento\Framework\MessageQueue\ConsumerInterface;
+use Magento\Framework\MessageQueue\EnvelopeInterface;
+
+/**
+ * Class ConsumerWithInjectableEnvelopeCallback
+ * @package LizardMedia\MessageQueue\Queue\Consumer
+ * @SuppressWarnings(PHPMD.LongVariable)
+ * @codeCoverageIgnore
+ */
+class ConsumerWithInjectableEnvelopeCallback implements ConsumerInterface
+{
+    /**
+     * @var string
+     */
+    private $envelopeCallbackType;
+
+    /**
+     * @var EnvelopeCallbackFactoryInterface
+     */
+    private $envelopeCallbackFactory;
+
+    /**
+     * @var CallbackInvokerInterface
+     */
+    private $invoker;
+
+    /**
+     * @var UsedConsumerConfig
+     */
+    private $usedConsumerConfig;
+
+    /**
+     * @param EnvelopeCallbackFactoryInterface $envelopeCallbackFactory
+     * @param CallbackInvokerInterface $invoker
+     * @param UsedConsumerConfig $configuration
+     * @param string $envelopeCallbackType
+     */
+    public function __construct(
+        EnvelopeCallbackFactoryInterface $envelopeCallbackFactory,
+        CallbackInvokerInterface $invoker,
+        UsedConsumerConfig $configuration,
+        string $envelopeCallbackType
+    ) {
+        $this->envelopeCallbackType = $envelopeCallbackType;
+        $this->envelopeCallbackFactory = $envelopeCallbackFactory;
+        $this->invoker = $invoker;
+        $this->usedConsumerConfig = $configuration;
+    }
+
+    /**
+     * @param null $maxNumberOfMessages
+     * @return void
+     */
+    public function process($maxNumberOfMessages = null): void
+    {
+        $queue = $this->usedConsumerConfig->getQueue();
+
+        if (!isset($maxNumberOfMessages)) {
+            $queue->subscribe($this->getTransactionCallback($this->usedConsumerConfig));
+        } else {
+            $this->invoker->invoke($queue, $maxNumberOfMessages, $this->getTransactionCallback($this->usedConsumerConfig));
+        }
+    }
+
+    /**
+     * @param UsedConsumerConfig $usedConsumerConfig
+     * @return Closure
+     */
+    protected function getTransactionCallback(UsedConsumerConfig $usedConsumerConfig): Closure
+    {
+        $callbackInstance =  $this->envelopeCallbackFactory->create($this->envelopeCallbackType, $usedConsumerConfig);
+
+        return static function (EnvelopeInterface $message) use ($callbackInstance) {
+            $callbackInstance->execute($message);
+        };
+    }
+}