[MessageQueue]: add support for first class queue configuration.

app/code/Magento/Ui/etc/di.xml

@@ -256,6 +256,7 @@
                 <item name="array" xsi:type="object">arrayArgumentInterpreterProxy</item>
                 <item name="boolean" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\Boolean</item>
                 <item name="number" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\Number</item>
+                <item name="int" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\Integer</item>
                 <item name="string" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\StringUtils</item>
                 <item name="null" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\NullType</item>
                 <item name="url" xsi:type="object">Magento\Ui\Config\Argument\Parser\Url</item>

app/etc/di.xml

@@ -479,6 +479,7 @@
                 <item name="array" xsi:type="object">layoutArrayArgumentReaderInterpreterProxy</item>
                 <item name="boolean" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\Boolean</item>
                 <item name="number" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\Number</item>
+                <item name="int" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\Integer</item>
                 <item name="string" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\StringUtils</item>
                 <item name="null" xsi:type="object">Magento\Framework\Data\Argument\Interpreter\NullType</item>
                 <item name="object" xsi:type="object">Magento\Framework\View\Layout\Argument\Interpreter\Passthrough</item>

lib/internal/Magento/Framework/App/ObjectManagerFactory.php

@@ -231,6 +231,7 @@ protected function createArgumentInterpreter(
                 'boolean' => new \Magento\Framework\Data\Argument\Interpreter\Boolean($booleanUtils),
                 'string' => new \Magento\Framework\Data\Argument\Interpreter\BaseStringUtils($booleanUtils),
                 'number' => new \Magento\Framework\Data\Argument\Interpreter\Number(),
+                'int' => new \Magento\Framework\Data\Argument\Interpreter\Integer(),
                 'null' => new \Magento\Framework\Data\Argument\Interpreter\NullType(),
                 'object' => new \Magento\Framework\Data\Argument\Interpreter\DataObject($booleanUtils),
                 'const' => $constInterpreter,

lib/internal/Magento/Framework/Data/Argument/Interpreter/Integer.php

@@ -0,0 +1,28 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+namespace Magento\Framework\Data\Argument\Interpreter;
+
+use Magento\Framework\Data\Argument\InterpreterInterface;
+
+/**
+ * Interpreter of numeric data, such as integer or float
+ */
+class Integer implements InterpreterInterface
+{
+    /**
+     * {@inheritdoc}
+     * @return int|float
+     * @throws \InvalidArgumentException
+     */
+    public function evaluate(array $data)
+    {
+        if (!isset($data['value']) || !is_numeric($data['value'])) {
+            throw new \InvalidArgumentException('Numeric value is expected.');
+        }
+        $result = $data['value'];
+        return (int)$result;
+    }
+}

lib/internal/Magento/Framework/Data/etc/argument/types.xsd

@@ -73,6 +73,12 @@
         </xs:complexContent>
     </xs:complexType>
 
+    <xs:complexType name="int">
+        <xs:complexContent>
+            <xs:extension base="argumentType"/>
+        </xs:complexContent>
+    </xs:complexType>
+
     <xs:complexType name="null">
         <xs:complexContent>
             <xs:restriction base="argumentType"/>

lib/internal/Magento/Framework/MessageQueue/Topology/Config/Data.php

@@ -23,4 +23,20 @@ public function __construct(
     ) {
         parent::__construct($reader, $cache, $cacheId, $serializer);
     }
+
+    public function getQueues(): array {
+        return array_filter($this->get(),function($item){
+            if($item['type'] == 'queue'){
+                return $item;
+            }
+        });
+    }
+
+    public function getExchanges(): array {
+        return array_filter($this->get(),function($item){
+            if($item['type'] != 'queue'){
+                return $item;
+            }
+        });
+    }
 }

lib/internal/Magento/Framework/MessageQueue/Topology/Config/ExchangeConfigItem.php

@@ -7,6 +7,8 @@
 
 use Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItem\BindingInterface;
 use Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItem\Binding\IteratorFactory;
+use Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItem\Binding\Iterator;
+use Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItem\BindingFactory;
 
 /**
  * {@inheritdoc}
@@ -37,7 +39,7 @@ class ExchangeConfigItem implements ExchangeConfigItemInterface
     /**
      * Exchange bindings.
      *
-     * @var BindingInterface[]
+     * @var Iterator
      */
     private $bindings;
 
@@ -69,6 +71,11 @@ class ExchangeConfigItem implements ExchangeConfigItemInterface
      */
     private $isInternal;
 
+    /**
+     * @var BindingFactory
+     */
+    public $bindingFactory;
+
     /**
      * Initialize dependencies.
      *
@@ -158,6 +165,11 @@ public function setData(array $data)
         $this->isDurable = $data['durable'];
         $this->isAutoDelete = $data['autoDelete'];
         $this->arguments = $data['arguments'];
-        $this->bindings->setData($data['bindings']);
+        if( $this->type != 'queue'){
+            $this->bindings->setData($data['bindings']);
+        }
+        else{
+            $this->bindings->setData([]);
+        }
     }
 }

lib/internal/Magento/Framework/MessageQueue/Topology/Config/ExchangeConfigItem/Iterator.php

@@ -36,7 +36,7 @@ class Iterator implements \Iterator, \ArrayAccess
      */
     public function __construct(Data $configData, ExchangeConfigItemFactory $itemFactory)
     {
-        $this->data = $configData->get();
+        $this->data = $configData->getExchanges();
         $this->object = $itemFactory->create();
         $this->rewind();
     }

lib/internal/Magento/Framework/MessageQueue/Topology/Config/QueueConfigItem/DataMapper.php

@@ -1,8 +1,10 @@
 <?php
+
 /**
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
+
 namespace Magento\Framework\MessageQueue\Topology\Config\QueueConfigItem;
 
 use Magento\Framework\MessageQueue\Topology\Config\Data;
@@ -64,16 +66,22 @@ public function getMappedData()
     {
         if (null === $this->mappedData) {
             $this->mappedData = [];
-            foreach ($this->configData->get() as $exchange) {
+
+            $queues = $this->createQueueItemsFromQueues($this->configData->getQueues());
+            foreach ($queues as $key => $value) {
+                $this->mappedData[$key] = $value;
+            }
+            foreach ($this->configData->getExchanges() as $exchange) {
                 $connection = $exchange['connection'];
                 foreach ($exchange['bindings'] as $binding) {
-                    if ($binding['destinationType'] === 'queue') {
+                    if ($binding['destinationType'] === 'queue' && !array_key_exists($binding['destination'] . '--' . $connection, $this->mappedData)) {
                         $queueItems = $this->createQueueItems($binding['destination'], $binding['topic'], $connection);
                         $this->mappedData = array_merge($this->mappedData, $queueItems);
                     }
                 }
             }
         }
+
         return $this->mappedData;
     }
 
@@ -83,13 +91,54 @@ public function getMappedData()
      * @param string $name
      * @param string $topic
      * @param string $connection
+     * @deprecated
      * @return array
      */
     private function createQueueItems($name, $topic, $connection)
+    {
+        return $this->createQueueItemsFromTopic($name, $topic, $connection);
+    }
+
+    private function createQueueItemsFromQueues(array $queues)
     {
         $output = [];
-        $synchronousTopics = [];
+        foreach ($queues as $queue) {
+            $output[$queue['name'] . '--' . $queue['connection']] =
+                $this->createQueue(
+                    $queue['name'],
+                    $queue['connection'],
+                    $queue['arguments'],
+                    $queue['durable'],
+                    $queue['autoDelete']
+                );
+        }
+        return $output;
+    }
 
+    private function createQueue($name, $connection, $arguments = [], $durable = true, $autoDelete = false)
+    {
+        return [
+            'name' => $name,
+            'connection' => $connection,
+            'durable' => isset($durable) ? $durable : true,
+            'autoDelete' => isset($autoDelete) ? $autoDelete : false,
+            'arguments' => isset($arguments) ? $arguments : [],
+        ];
+    }
+
+    /**
+     * Create queue config item.
+     *
+     * @param string $name
+     * @param string $topic
+     * @param string $connection
+     * @return array
+     */
+    private function createQueueItemsFromTopic($name, $topic, $connection)
+    {
+        $output = [];
+        $synchronousTopics = [];
+
         if (strpos($topic, '*') !== false || strpos($topic, '#') !== false) {
             $synchronousTopics = $this->matchSynchronousTopics($topic);
         } elseif ($this->isSynchronousTopic($topic)) {

lib/internal/Magento/Framework/MessageQueue/Topology/Config/Validator/DependentFields.php

@@ -1,8 +1,10 @@
 <?php
+
 /**
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
+
 namespace Magento\Framework\MessageQueue\Topology\Config\Validator;
 
 use Magento\Framework\MessageQueue\Topology\Config\ValidatorInterface;
@@ -19,9 +21,11 @@ public function validate($configData)
     {
         $errors = [];
         foreach ($configData as $name => $data) {
-            foreach ((array)$data['bindings'] as $binding) {
-                if (isset($data['type']) && $data['type'] == 'topic' && !isset($binding['topic'])) {
-                    $errors[] = 'Topic name is required for topic based exchange: ' . $name;
+            if ($data['type'] != 'queue') {
+                foreach ((array)$data['bindings'] as $binding) {
+                    if (isset($data['type']) && $data['type'] == 'topic' && !isset($binding['topic'])) {
+                        $errors[] = 'Topic name is required for topic based exchange: ' . $name;
+                    }
                 }
             }
         }

lib/internal/Magento/Framework/MessageQueue/Topology/Config/Validator/FieldsTypes.php

@@ -18,7 +18,8 @@ class FieldsTypes implements ValidatorInterface
     public function validate($configData)
     {
         foreach ($configData as $exchangeName => $exchangeConfig) {
-            $this->validateFieldsTypes($exchangeName, $exchangeConfig);
+            $exchangeConfig['type'] != 'queue' ? 
+            $this->validateFieldsTypes($exchangeName, $exchangeConfig) : $this->validateQueueTypes($exchangeName, $exchangeConfig);
         }
     }
 
@@ -128,4 +129,16 @@ private function validateBindings($exchangeName, $exchangeConfig, $bindingFields
             }
         }
     }
+
+    /**
+     * Make sure types of all fields in the queue item config are correct.
+     *
+     * @param string $queueName
+     * @param array $queueConfig
+     * @return void
+     * @throws \LogicException
+     */
+    private function validateQueueTypes($queueName, $queueConfig)
+    {
+    }
 }

lib/internal/Magento/Framework/MessageQueue/Topology/Config/Validator/Format.php

@@ -1,8 +1,10 @@
 <?php
+
 /**
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
+
 namespace Magento\Framework\MessageQueue\Topology\Config\Validator;
 
 use Magento\Framework\MessageQueue\Topology\Config\ValidatorInterface;
@@ -17,24 +19,34 @@ class Format implements ValidatorInterface
      */
     public function validate($configData)
     {
-        $requiredFields = ['name', 'type', 'connection', 'durable', 'autoDelete', 'internal', 'bindings', 'arguments'];
+        $requiredExchangeFields = ['name', 'type', 'connection', 'durable', 'autoDelete', 'internal', 'bindings', 'arguments'];
+        $requiredQueueFields = ['name', 'type', 'connection', 'durable', 'autoDelete', 'internal', 'arguments'];
+
         $requiredBindingFields = ['id', 'destinationType', 'destination', 'disabled', 'topic', 'arguments'];
         $errors = [];
         foreach ($configData as $name => $data) {
-            $diff = array_diff($requiredFields, array_keys($data));
-            foreach ($diff as $field) {
-                $errors[] = sprintf('Missing [%s] field for exchange %s.', $field, $name);
-            }
+            if ($data['type'] != 'queue') {
+                $diff = array_diff($requiredExchangeFields, array_keys($data));
+                foreach ($diff as $field) {
+                    $errors[] = sprintf('Missing [%s] field for exchange %s.', $field, $name);
+                }
 
-            if (!array_key_exists('bindings', $data) || !is_array($data['bindings'])) {
-                $errors[] = sprintf('Invalid bindings format for exchange %s.', $name);
-                continue;
-            }
+                if (!array_key_exists('bindings', $data) || !is_array($data['bindings'])) {
+                    $errors[] = sprintf('Invalid bindings format for exchange %s.', $name);
+                    continue;
+                }
 
-            foreach ($data['bindings'] as $bindingConfig) {
-                $diff = array_diff($requiredBindingFields, array_keys($bindingConfig));
+                foreach ($data['bindings'] as $bindingConfig) {
+                    $diff = array_diff($requiredBindingFields, array_keys($bindingConfig));
+                    foreach ($diff as $field) {
+                        $errors[] = sprintf('Missing [%s] field for binding %s in exchange config.', $field, $name);
+                    }
+                }
+            }
+            else{
+                $diff = array_diff($requiredQueueFields, array_keys($data));
                 foreach ($diff as $field) {
-                    $errors[] = sprintf('Missing [%s] field for binding %s in exchange config.', $field, $name);
+                    $errors[] = sprintf('Missing [%s] field for queue %s.', $field, $name);
                 }
             }
         }

lib/internal/Magento/Framework/MessageQueue/Topology/Config/Xml/Converter.php

@@ -99,6 +99,41 @@ public function convert($source)
                 'arguments' => $exchangeArguments,
             ];
         }
+
+        foreach ($source->getElementsByTagName('queue') as $queue) {
+            $name = $this->getAttributeValue($queue, 'name');
+            $connection = $this->getAttributeValue($queue, 'connection');
+
+            $bindings = [];
+            $queueArguments = [];
+            /** @var \DOMNode $node */
+            foreach ($queue->childNodes as $node) {
+                if (!in_array($node->nodeName, ['binding', 'arguments']) || $node->nodeType != XML_ELEMENT_NODE) {
+                    continue;
+                }
+                switch ($node->nodeName) {
+                    case 'binding':
+                        $bindings = $this->processBindings($node, $bindings);
+                        break;
+
+                    case 'arguments':
+                        $queueArguments = $this->processArguments($node);
+                        break;
+                }
+            }
+
+            $autoDelete = $this->getAttributeValue($queue, 'autoDelete', false);
+            $result[$name . '--' . $connection] = [
+                'name' => $name,
+                'type' => $this->getAttributeValue($queue, 'type'),
+                'connection' => $connection,
+                'durable' => $this->booleanUtils->toBoolean($this->getAttributeValue($queue, 'durable', true)),
+                'autoDelete' => $this->booleanUtils->toBoolean($autoDelete),
+                'internal' => $this->booleanUtils->toBoolean($this->getAttributeValue($queue, 'internal', false)),
+                'arguments' => $queueArguments,
+            ];
+        }
+
         return $result;
     }
 

lib/internal/Magento/Framework/MessageQueue/Topology/Config/Xml/Reader.php

@@ -22,6 +22,9 @@ class Reader extends \Magento\Framework\Config\Reader\Filesystem implements Read
         '/config/exchange/binding' => 'id',
         '/config/exchange/binding/arguments/argument' => 'name',
         '/config/exchange/binding/arguments/argument(/item)+' => 'name',
+        '/config/queue' => ['name', 'connection'],
+        '/config/queue/arguments/argument' => 'name',
+        '/config/queue/arguments/argument(/item)+' => 'name'
     ];
 
     /**