Create DefaultConsumer

Console/Command/DefaultMessage.php

+<?php
+namespace NicolasBejean\MySqlMq\Console\Command;
+
+use Magento\AsynchronousOperations\Api\Data\OperationInterface;
+use Magento\AsynchronousOperations\Api\Data\OperationInterfaceFactory;
+use Magento\Framework\App\Area;
+use Magento\Framework\App\State;
+use Magento\Framework\Bulk\BulkManagementInterface;
+use Magento\Framework\Bulk\OperationInterface as BulkOperationInterface;
+use Magento\Framework\DataObject\IdentityGeneratorInterface;
+use Magento\Framework\Exception\LocalizedException;
+use Magento\Framework\Serialize\Serializer\Json as JsonSerializer;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+
+/**
+ * Class DefaultMessage
+ *
+ * Crée un message dans le topic par défaut
+ *
+ * @author   Nicolas Béjean <nicolas@bejean.eu>
+ * @license  https://github.com/nicolasbejean/mysqlmq/blob/master/licence.txt BSD Licence
+ * @link     https://www.bejean.eu
+ */
+class DefaultMessage extends Command
+{
+    /**
+     * Description de la commande
+     */
+    const DESCRIPTION = 'Crée un message dans le topic par défaut';
+
+    /**
+     * @var JsonSerializer
+     */
+    protected $jsonSerializer;
+
+    /**
+     * @var State
+     */
+    protected $state;
+
+    /**
+     * @var int
+     */
+    private $bulkSize;
+
+    /**
+     * @var IdentityGeneratorInterface
+     */
+    private $identityService;
+
+    /**
+     * @var BulkManagementInterface
+     */
+    private $bulkManagement;
+
+    /**
+     * @var OperationInterfaceFactory
+     */
+    private $operationFactory;
+
+    /**
+     * DefaultConsumer constructor.
+     *
+     * @param string|null $name
+     * @param JsonSerializer $jsonSerializer
+     * @param State $state
+     * @param IdentityGeneratorInterface $identityService
+     * @param BulkManagementInterface $bulkManagement
+     * @param OperationInterfaceFactory $operationFactory
+     * @param int $bulkSize
+     */
+    public function __construct(
+        JsonSerializer $jsonSerializer,
+        State $state,
+        IdentityGeneratorInterface $identityService,
+        BulkManagementInterface $bulkManagement,
+        OperationInterfaceFactory $operationFactory,
+        int $bulkSize = 100,
+        ?string $name = null
+    ) {
+        $this->jsonSerializer = $jsonSerializer;
+        $this->state = $state;
+
+        $this->bulkSize = $bulkSize;
+        $this->identityService = $identityService;
+        $this->bulkManagement = $bulkManagement;
+        $this->operationFactory = $operationFactory;
+
+        parent::__construct($name);
+    }
+
+    /**
+     * Configuration de la commande
+     */
+    protected function configure()
+    {
+        /* Options de la commande */
+        $options = [];
+
+        $this->setDefinition($options);
+        $this->setDescription(self::DESCRIPTION);
+
+        parent::configure();
+    }
+
+    /**
+     * Méthode d'exécution de la commande
+     *
+     * @param InputInterface $input
+     * @param OutputInterface $output
+     * @return null
+     * @throws LocalizedException
+     */
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $output->write('Start execution', true);
+        $time_start = microtime(true);
+
+        $this->state->setAreaCode(Area::AREA_ADMINHTML);
+
+        $this->publish('Données de test');
+
+        $time_end = microtime(true);
+        $time = $time_end - $time_start;
+        $output->write('Total execution time in seconds: ' . $time, true);
+    }
+
+    /**
+     * Publie le mapping dans MySqlMq
+     *
+     * @param string $data
+     * @throws LocalizedException
+     */
+    private function publish(string $data):void
+    {
+        $bulkUuid = $this->identityService->generateId();
+        $bulkDescription = __('Insert Message: ' . $data);
+        $operations = [];
+
+        $operations[] = $this->makeOperation(
+            $data,
+            'nicolasbejean.mysqlmq.consumer.default',
+            $bulkUuid
+        );
+
+        if (!empty($operations)) {
+            $result = $this->bulkManagement->scheduleBulk(
+                $bulkUuid,
+                $operations,
+                $bulkDescription
+            );
+            if (!$result) {
+                throw new LocalizedException(
+                    __('Something went wrong while processing the request.')
+                );
+            }
+        }
+    }
+
+    /**
+     * Crée le bulk dans Magento Bulk
+     *
+     * @param $dataToEncode
+     * @param $queue
+     * @param $bulkUuid
+     * @return OperationInterface
+     */
+    private function makeOperation(
+        $dataToEncode,
+        $queue,
+        $bulkUuid
+    ):OperationInterface {
+        $data = [
+            'data' => [
+                'bulk_uuid' => $bulkUuid,
+                'topic_name' => $queue,
+                'serialized_data' => $this->jsonSerializer->serialize($dataToEncode),
+                'status' => BulkOperationInterface::STATUS_TYPE_OPEN,
+            ]
+        ];
+
+        return $this->operationFactory->create($data);
+    }
+}

Model/Consumer/DefaultConsumer.php

+<?php
+declare(strict_types=1);
+
+namespace NicolasBejean\MySqlMq\Model\Consumer;
+
+use Magento\Framework\Serialize\Serializer\Json as JsonSerializer;
+use Psr\Log\LoggerInterface;
+use Exception;
+
+use Magento\AsynchronousOperations\Api\Data\OperationInterface;
+
+class DefaultConsumer
+{
+    /**
+     * @var JsonSerializer
+     */
+    protected $jsonSerializer;
+
+    /**
+     * @var LoggerInterface
+     */
+    protected $logger;
+
+    /**
+     * Create constructor.
+     * @param JsonSerializer $jsonSerializer
+     * @param LoggerInterface $logger
+     */
+    public function __construct(
+        JsonSerializer $jsonSerializer,
+        LoggerInterface $logger
+    ) {
+        $this->jsonSerializer = $jsonSerializer;
+        $this->logger = $logger;
+    }
+
+    /**
+     * @param OperationInterface $data
+     */
+    public function process(OperationInterface $data)
+    {
+        $time_start = microtime(true);
+
+        $data = $this->jsonSerializer->unserialize($data->getSerializedData());
+
+        $time_end = microtime(true);
+        $time = $time_end - $time_start;
+        $this->logger->info('[NicolasBejean_MySqlMq] DefaultConsumer::process | Total execution time in seconds: ' . $time);
+    }
+}

README.md

@@ -11,3 +11,14 @@ Lancer dans le terminal la commande suivante pour configurer le chemin du dépot
 Ensuite, procéder à l'installation avec la commande : 
 
 `composer require nicolasbejean/mysqlmq`
+
+## STATUS DES MESSAGES 
+
+```
+const MESSAGE_STATUS_NEW = 2;
+const MESSAGE_STATUS_IN_PROGRESS = 3;
+const MESSAGE_STATUS_COMPLETE= 4;
+const MESSAGE_STATUS_RETRY_REQUIRED = 5;
+const MESSAGE_STATUS_ERROR = 6;
+const MESSAGE_STATUS_TO_BE_DELETED = 7;
+```

composer.json

+{
+    "name": "nicolasbejean/mysqlmq",
+    "description": "",
+    "type": "magento2-module",
+    "version": "1.0.0",
+    "require": {
+        "php": "~7.1.3||~7.2.0||~7.3.0",
+        "magento/magento-composer-installer": "*",
+        "magento/framework-amqp": "100.3.5",
+        "nicolasbejean/base": ">=1.4.2"
+    },
+    "license": [
+        "OSL-3.0",
+        "AFL-3.0"
+    ],
+    "autoload": {
+        "psr-4": {
+            "NicolasBejean\\MySqlMq\\": ""
+        },
+        "files": [
+            "registration.php"
+        ]
+    }
+}

etc/communication.xml

+<?xml version="1.0"?>
+<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">
+    <topic name="nicolasbejean.mysqlmq.consumer.default" request="Magento\AsynchronousOperations\Api\Data\OperationInterface">
+        <handler name="NicolasBejeanMySqlMqConsumer" type="NicolasBejean\MySqlMq\Model\Consumer\DefaultConsumer" method="process" />
+    </topic>
+</config>

etc/di.xml

+<?xml version="1.0" encoding="utf-8" ?>
+<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework:ObjectManager/etc/config.xsd">
+    <type name="NicolasBejean\MySqlMq\Console\Command\DefaultMessage">
+        <arguments>
+            <argument name="name" xsi:type="string">nicolasbejean:mysqlmq:defaultmessage</argument>
+        </arguments>
+    </type>
+    <type name="Magento\Framework\Console\CommandListInterface">
+        <arguments>
+            <argument name="commands" xsi:type="array">
+                <item name="nicolasbejean_mysqlmq_defaultmessage" xsi:type="object">NicolasBejean\MySqlMq\Console\Command\DefaultMessage</item>
+            </argument>
+        </arguments>
+    </type>
+</config>

etc/module.xml

+<?xml version="1.0"?>
+<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework:Module/etc/module.xsd">
+    <module name="NicolasBejean_MySqlMq" setup_version="1.0.0">
+        <sequence>
+            <module name="Magento_MysqlMq"/>
+            <module name="NicolasBejean_Base"/>
+        </sequence>
+    </module>
+</config>

etc/queue_consumer.xml

+<?xml version="1.0"?>
+<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
+    <consumer name="NicolasBejeanMySqlMqConsumer"
+              queue="nicolasbejean.mysqlmq.consumer.default"
+              connection="db"
+              maxMessages="500"
+              consumerInstance="Magento\Framework\MessageQueue\Consumer"
+              handler="NicolasBejean\MySqlMq\Model\Consumer\DefaultConsumer::process"/>
+</config>

etc/queue_publisher.xml

+<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/publisher.xsd">
+    <publisher topic="nicolasbejean.mysqlmq.consumer.default">
+        <connection name="db" exchange="nicolasbejean-mysqlmq-exchange" />
+    </publisher>
+</config>

etc/queue_topology.xml

+<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">
+    <exchange name="nicolasbejean-mysqlmq-exchange" type="topic" connection="db">
+        <binding id="NicolasBejeanMySqlMqConsumer" topic="nicolasbejean.mysqlmq.consumer.default" destinationType="queue" destination="nicolasbejean.mysqlmq.consumer.default"/>
+    </exchange>
+</config>

registration.php

+<?php
+\Magento\Framework\Component\ComponentRegistrar::register(
+    \Magento\Framework\Component\ComponentRegistrar::MODULE,
+    'NicolasBejean_MySqlMq',
+    __DIR__
+);