Publisher.php 3.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137
  1. <?php
  2. /**
  3. * Copyright © Magento, Inc. All rights reserved.
  4. * See COPYING.txt for license details.
  5. */
  6. namespace Magento\Framework\MessageQueue;
  7. use Magento\Framework\Amqp\Config as AmqpConfig;
  8. use Magento\Framework\MessageQueue\ConfigInterface as MessageQueueConfig;
  9. use Magento\Framework\MessageQueue\Publisher\ConfigInterface as PublisherConfig;
  10. /**
  11. * A MessageQueue Publisher to handle publishing a message.
  12. */
  13. class Publisher implements PublisherInterface
  14. {
  15. /**
  16. * @var ExchangeRepository
  17. */
  18. private $exchangeRepository;
  19. /**
  20. * @var EnvelopeFactory
  21. */
  22. private $envelopeFactory;
  23. /**
  24. * @var MessageEncoder
  25. */
  26. private $messageEncoder;
  27. /**
  28. * @var MessageValidator
  29. */
  30. private $messageValidator;
  31. /**
  32. * @var PublisherConfig
  33. */
  34. private $publisherConfig;
  35. /**
  36. * Help check whether Amqp is configured.
  37. *
  38. * @var AmqpConfig
  39. */
  40. private $amqpConfig;
  41. /**
  42. * Initialize dependencies.
  43. *
  44. * @param ExchangeRepository $exchangeRepository
  45. * @param EnvelopeFactory $envelopeFactory
  46. * @param MessageQueueConfig $messageQueueConfig
  47. * @param MessageEncoder $messageEncoder
  48. * @param MessageValidator $messageValidator
  49. * @internal param ExchangeInterface $exchange
  50. *
  51. * @SuppressWarnings(PHPMD.UnusedFormalParameter)
  52. */
  53. public function __construct(
  54. ExchangeRepository $exchangeRepository,
  55. EnvelopeFactory $envelopeFactory,
  56. MessageQueueConfig $messageQueueConfig,
  57. MessageEncoder $messageEncoder,
  58. MessageValidator $messageValidator
  59. ) {
  60. $this->exchangeRepository = $exchangeRepository;
  61. $this->envelopeFactory = $envelopeFactory;
  62. $this->messageEncoder = $messageEncoder;
  63. $this->messageValidator = $messageValidator;
  64. }
  65. /**
  66. * {@inheritdoc}
  67. */
  68. public function publish($topicName, $data)
  69. {
  70. $this->messageValidator->validate($topicName, $data);
  71. $data = $this->messageEncoder->encode($topicName, $data);
  72. $envelope = $this->envelopeFactory->create(
  73. [
  74. 'body' => $data,
  75. 'properties' => [
  76. 'delivery_mode' => 2,
  77. 'message_id' => md5(uniqid($topicName))
  78. ]
  79. ]
  80. );
  81. $connectionName = $this->getPublisherConfig()->getPublisher($topicName)->getConnection()->getName();
  82. $connectionName = ($connectionName === 'amqp' && !$this->isAmqpConfigured()) ? 'db' : $connectionName;
  83. $exchange = $this->exchangeRepository->getByConnectionName($connectionName);
  84. $exchange->enqueue($topicName, $envelope);
  85. return null;
  86. }
  87. /**
  88. * Check Amqp is configured.
  89. *
  90. * @return bool
  91. */
  92. private function isAmqpConfigured()
  93. {
  94. return $this->getAmqpConfig()->getValue(AmqpConfig::HOST) ? true : false;
  95. }
  96. /**
  97. * Get publisher config.
  98. *
  99. * @return PublisherConfig
  100. *
  101. * @deprecated 102.0.1
  102. */
  103. private function getPublisherConfig()
  104. {
  105. if ($this->publisherConfig === null) {
  106. $this->publisherConfig = \Magento\Framework\App\ObjectManager::getInstance()->get(PublisherConfig::class);
  107. }
  108. return $this->publisherConfig;
  109. }
  110. /**
  111. * Get Amqp config instance.
  112. *
  113. * @return AmqpConfig
  114. *
  115. * @deprecated 100.2.0 102.0.1
  116. */
  117. private function getAmqpConfig()
  118. {
  119. if ($this->amqpConfig === null) {
  120. $this->amqpConfig = \Magento\Framework\App\ObjectManager::getInstance()->get(AmqpConfig::class);
  121. }
  122. return $this->amqpConfig;
  123. }
  124. }