messageEncoder = $messageEncoder;
$this->queueRepository = $queueRepository;
$this->mergerFactory = $mergerFactory;
$this->interval = $interval;
$this->batchSize = $batchSize;
$this->resource = $resource;
$this->configuration = $configuration;
$this->messageProcessorLoader = $messageProcessorLoader
?: \Magento\Framework\App\ObjectManager::getInstance()->get(MessageProcessorLoader::class);
$this->consumerConfig = $consumerConfig ?: ObjectManager::getInstance()->get(ConsumerConfig::class);
}
/**
* @inheritdoc
*/
public function process($maxNumberOfMessages = null)
{
$queueName = $this->configuration->getQueueName();
$consumerName = $this->configuration->getConsumerName();
$connectionName = $this->getConsumerConfig()->getConsumer($consumerName)->getConnection();
$queue = $this->queueRepository->get($connectionName, $queueName);
$merger = $this->mergerFactory->create($consumerName);
$consumerName = $this->configuration->getConsumerName();
$connectionName = $this->consumerConfig->getConsumer($consumerName)->getConnection();
if ($connectionName === 'stomp') {
$queue->subscribeQueue();
}
if (!isset($maxNumberOfMessages)) {
$this->runDaemonMode($queue, $merger);
} else {
$this->run($queue, $merger, $maxNumberOfMessages);
}
}
/**
* Run process in a daemon mode.
*
* @param QueueInterface $queue
* @param MergerInterface $merger
* @return void
*/
private function runDaemonMode(QueueInterface $queue, MergerInterface $merger)
{
$transactionCallback = $this->getTransactionCallback($queue, $merger);
while (true) {
$messages = $this->batchSize > 0
? $this->getMessages($queue, $this->batchSize)
: $this->getAllMessages($queue);
$transactionCallback($messages);
sleep($this->interval);
}
}
/**
* Run short running process.
*
* @param QueueInterface $queue
* @param MergerInterface $merger
* @param int $maxNumberOfMessages
* @return void
*/
private function run(QueueInterface $queue, MergerInterface $merger, $maxNumberOfMessages)
{
$count = ($maxNumberOfMessages
? $maxNumberOfMessages
: $this->configuration->getMaxMessages()) ?: 1;
$transactionCallback = $this->getTransactionCallback($queue, $merger);
if ($this->batchSize) {
while ($count > 0) {
$messages = $this->getMessages($queue, $count > $this->batchSize ? $this->batchSize : $count);
$transactionCallback($messages);
$count -= $this->batchSize;
}
} else {
$messages = $this->getMessages($queue, $count);
$transactionCallback($messages);
}
}
/**
* Get all messages from a queue.
*
* @param QueueInterface $queue
* @return EnvelopeInterface[]
*/
private function getAllMessages(QueueInterface $queue)
{
$messages = [];
while ($message = $queue->dequeue()) {
$messages[] = $message;
}
return $messages;
}
/**
* Get $count messages from a queue.
*
* @param QueueInterface $queue
* @param int $count
* @return EnvelopeInterface[]
*/
private function getMessages(QueueInterface $queue, $count)
{
$messages = [];
for ($i = $count; $i > 0; $i--) {
$message = $queue->dequeue();
if ($message === null) {
break;
}
$messages[] = $message;
}
return $messages;
}
/**
* Decode provided messages.
*
* @param EnvelopeInterface[] $messages
* @return object[]
*/
private function decodeMessages(array $messages)
{
$decodedMessages = [];
foreach ($messages as $messageId => $message) {
$properties = $message->getProperties();
$topicName = $properties['topic_name'];
$decodedMessages[$topicName][$messageId] = $this->messageEncoder->decode($topicName, $message->getBody());
}
return $decodedMessages;
}
/**
* Get transaction callback.
*
* @param QueueInterface $queue
* @param MergerInterface $merger
* @return \Closure
*/
private function getTransactionCallback(QueueInterface $queue, MergerInterface $merger)
{
return function (array $messages) use ($queue, $merger) {
list($messages, $messagesToAcknowledge) = $this->lockMessages($messages);
$decodedMessages = $this->decodeMessages($messages);
$mergedMessages = $merger->merge($decodedMessages);
$messageProcessor = $this->messageProcessorLoader->load($mergedMessages);
$messageProcessor->process(
$queue,
$this->configuration,
$messages,
$messagesToAcknowledge,
$mergedMessages
);
};
}
/**
* Create lock for the messages.
*
* @param array $messages
* @return array
*/
private function lockMessages(array $messages)
{
$toProcess = [];
$toAcknowledge = [];
foreach ($messages as $message) {
try {
$this->getMessageController()->lock($message, $this->configuration->getConsumerName());
$toProcess[] = $message;
} catch (MessageLockException $exception) {
$toAcknowledge[] = $message;
}
}
return [$toProcess, $toAcknowledge];
}
/**
* Get consumer config.
*
* This getter serves as a workaround to add this dependency to this class without breaking constructor structure
*
* @return ConsumerConfig
*
* @deprecated 103.0.0 Use constructor injection instead.
* @see \Magento\Framework\MessageQueue\Consumer\ConfigInterface
*/
private function getConsumerConfig()
{
if ($this->consumerConfig === null) {
$this->consumerConfig = \Magento\Framework\App\ObjectManager::getInstance()->get(ConsumerConfig::class);
}
return $this->consumerConfig;
}
/**
* Get message controller.
*
* This getter serves as a workaround to add this dependency to this class without breaking constructor structure
*
* @return MessageController
*
* @deprecated 103.0.0 Use constructor injection instead.
* @see \Magento\Framework\MessageQueue\MessageController
*/
private function getMessageController()
{
if ($this->messageController === null) {
$this->messageController = \Magento\Framework\App\ObjectManager::getInstance()
->get(\Magento\Framework\MessageQueue\MessageController::class);
}
return $this->messageController;
}
}