Magento 2 Documentation  2.3
Documentation for Magento 2 CMS v2.3 (December 2018)
MessageQueue.php
Go to the documentation of this file.
1 <?php
7 
9 use Magento\Framework\Communication\ConfigInterface as CommunicationConfig;
10 
17 {
18  const DEFAULT_PUBLISHER = 'default';
19  const DEFAULT_CONNECTION = 'amqp';
20  const DEFAULT_EXCHANGE = 'magento';
21 
25  private $communicationReader;
26 
32  public function __construct(
33  Communication $communicationReader
34  ) {
35  $this->communicationReader = $communicationReader;
36  }
37 
44  public function read($scope = null)
45  {
46  $remoteServiceTopics = $this->communicationReader->read($scope);
47  $queueTopics = [];
48  $queueBinds = [];
49  $queueExchangeTopicToQueueMap = [];
50  foreach ($remoteServiceTopics[CommunicationConfig::TOPICS] as $topicName => $communicationConfig) {
51  $queueTopics[$topicName] = [
52  QueueConfig::TOPIC_NAME => $topicName,
53  QueueConfig::TOPIC_SCHEMA => [
54  QueueConfig::TOPIC_SCHEMA_TYPE => QueueConfig::TOPIC_SCHEMA_TYPE_METHOD,
55  QueueConfig::TOPIC_SCHEMA_VALUE => $communicationConfig[CommunicationConfig::TOPIC_REQUEST]
56  ],
57  QueueConfig::TOPIC_RESPONSE_SCHEMA => [
58  QueueConfig::TOPIC_SCHEMA_TYPE => isset($communicationConfig[CommunicationConfig::TOPIC_RESPONSE])
59  ? QueueConfig::TOPIC_SCHEMA_TYPE_OBJECT
60  : null,
61  QueueConfig::TOPIC_SCHEMA_VALUE => $communicationConfig[CommunicationConfig::TOPIC_RESPONSE]
62  ],
63  QueueConfig::TOPIC_PUBLISHER => self::DEFAULT_PUBLISHER
64  ];
65 
66  $queueName = 'queue.' . $topicName;
67  $queueBinds[$topicName . '--' . self::DEFAULT_EXCHANGE . '--' . $queueName] = [
68  QueueConfig::BIND_TOPIC => $topicName,
69  QueueConfig::BIND_EXCHANGE => self::DEFAULT_EXCHANGE,
70  QueueConfig::BIND_QUEUE => $queueName,
71  ];
72 
73  $queueExchangeTopicToQueueMap[self::DEFAULT_EXCHANGE . '--' . $topicName] = [$queueName];
74  }
75  $queuePublishers = [
76  self::DEFAULT_PUBLISHER => [
77  QueueConfig::PUBLISHER_NAME => self::DEFAULT_PUBLISHER,
78  QueueConfig::PUBLISHER_CONNECTION => self::DEFAULT_CONNECTION,
79  QueueConfig::PUBLISHER_EXCHANGE => self::DEFAULT_EXCHANGE
80  ]
81  ];
82  return [
83  QueueConfig::PUBLISHERS => $queuePublishers,
84  QueueConfig::TOPICS => $queueTopics,
85  QueueConfig::CONSUMERS => [],
86  QueueConfig::BINDS => $queueBinds,
87  QueueConfig::EXCHANGE_TOPIC_TO_QUEUES_MAP => $queueExchangeTopicToQueueMap,
88  ];
89  }
90 }