Magento 2 Documentation  2.3
Documentation for Magento 2 CMS v2.3 (December 2018)
MergedMessageProcessor.php
Go to the documentation of this file.
1 <?php
7 
12 {
16  private $messageStatusProcessor;
17 
21  public function __construct(MessageStatusProcessor $messageStatusProcessor)
22  {
23  $this->messageStatusProcessor = $messageStatusProcessor;
24  }
25 
29  public function process(
32  array $messages,
33  array $messagesToAcknowledge,
34  array $mergedMessages
35  ) {
36  $this->messageStatusProcessor->acknowledgeMessages($queue, $messagesToAcknowledge);
37  $this->dispatchMessages($queue, $configuration, $mergedMessages, $messages);
38  }
39 
48  private function dispatchMessages(
51  array $messageList,
52  array $originalMessages
53  ) {
54  $originalMessagesIds = [];
55 
56  try {
57  foreach ($messageList as $topicName => $messages) {
58  foreach ($messages as $message) {
62  $callbacks = $configuration->getHandlers($topicName);
63  $originalMessagesIds = $message->getOriginalMessagesIds();
64 
65  foreach ($callbacks as $callback) {
66  call_user_func($callback, $message->getMergedMessage());
67  }
68 
69  $originalMessages = $this->getOriginalMessages($originalMessages, $originalMessagesIds);
70  $this->messageStatusProcessor->acknowledgeMessages($queue, $originalMessages);
71  }
72  }
73  } catch (\Exception $e) {
74  $originalMessages = $this->getOriginalMessages($originalMessages, $originalMessagesIds);
75  $this->messageStatusProcessor->rejectMessages($queue, $originalMessages);
76  }
77  }
78 
86  private function getOriginalMessages(array $messages, array $messagesIds)
87  {
88  $originalMessages = [];
89 
90  foreach ($messagesIds as $messageId) {
91  if (isset($messages[$messageId])) {
92  $originalMessages[] = $messages[$messageId];
93  }
94  }
95 
96  return $originalMessages;
97  }
98 }
$queue
Definition: queue.php:21
__construct(MessageStatusProcessor $messageStatusProcessor)
$configuration
Definition: index.php:33
$message
process(QueueInterface $queue, ConsumerConfigurationInterface $configuration, array $messages, array $messagesToAcknowledge, array $mergedMessages)