PHP Oro\Component\MessageQueue\Consumption Context Code Examples

                                                    /**
     * @inheritdoc
     */
    public function onStart(Context $context)
    {
        $context->setLogger($this->logger);
        $this->logger->debug('Set logger to the context');
    }
                                            

                                                    public function onPreReceived(Context $context): void
    {
        if ($context->getMessageProcessorName()) {
            // Message processor is already set.
            return;
        }

        $topicName = $context->getMessage()?->getProperty(Config::PARAMETER_TOPIC_NAME) ?? '';
        $topicMeta = $this->topicMetaRegistry->getTopicMeta($topicName);

        $transportQueueName = $context->getMessage()?->getProperty(Config::PARAMETER_QUEUE_NAME) ?? '';
        $destinationMeta = $this->destinationMetaRegistry->getDestinationMetaByTransportQueueName($transportQueueName);

        $messageProcessorName = $topicMeta->getMessageProcessorName($destinationMeta->getQueueName());

        if (empty($messageProcessorName)) {
            // Falls back to noop message processor if message is not claimed by a message processor.
            $messageProcessorName = $this->noopMessageProcessorName;

            $context->getLogger()->warning(
                sprintf(
                    'Message processor for "%s" topic name in "%s" queue was not found, falling back to "%s"',
                    $topicName,
                    $destinationMeta->getQueueName(),
                    $this->noopMessageProcessorName
                )
            );
        } else {
            $context->getLogger()->debug(
                sprintf(
                    'Found "%s" message processor for topic "%s" in queue "%s"',
                    $messageProcessorName,
                    $topicName,
                    $destinationMeta->getQueueName()
                )
            );
        }

        $context->setMessageProcessorName($messageProcessorName);
    }
                                            

                                                    /**
     * @inheritdoc
     */
    public function onPreReceived(Context $context)
    {
        $message = $context->getMessage();
        if (!$message->isRedelivered()) {
            return;
        }

        if ($context->getStatus()) {
            // There is no sense in proceeding as message status is already known.
            $context->getLogger()->debug(
                'Skipping extension as message status is already set.',
                ['messageId' => $message->getMessageId(), 'status' => $context->getStatus()]
            );

            return;
        }

        $properties = $message->getProperties();
        if (!isset($properties[self::PROPERTY_REDELIVER_COUNT])) {
            $properties[self::PROPERTY_REDELIVER_COUNT] = 1;
        } else {
            $properties[self::PROPERTY_REDELIVER_COUNT]++;
        }

        $delayedMessage = new Message();
        $delayedMessage->setBody($message->getBody());
        $delayedMessage->setHeaders($message->getHeaders());
        $delayedMessage->setProperties($properties);
        $delayedMessage->setDelay($this->delay);
        $delayedMessage->setMessageId($message->getMessageId());

        $queue = $context->getSession()->createQueue($context->getQueueName());

        $this->driver->send($queue, $delayedMessage);
        $context->getLogger()->debug('Send delayed message');

        $context->setStatus(MessageProcessorInterface::REJECT);
        $context->getLogger()->debug('Reject redelivered original message by setting reject status to context.');
    }
                                            

                                                    private function interruptExecution(Context $context, string $reason): void
    {
        $context->getLogger()->info(
            'Execution interrupted: ' . $reason,
            ['context' => $context]
        );

        $context->setExecutionInterrupted(true);
        $context->setInterruptedReason($reason);
    }
                                            

                                                    private function interruptExecution(Context $context, string $reason): void
    {
        $context->getLogger()->info(
            'Execution interrupted: ' . $reason,
            ['context' => $context]
        );

        $context->setExecutionInterrupted(true);
        $context->setInterruptedReason($reason);
    }