MessageEncoder.php

Go to the documentation of this file.

<?php
namespace Magento\Framework\MessageQueue;

use Magento\Framework\MessageQueue\ConfigInterface as QueueConfig;
use Magento\Framework\Exception\LocalizedException;
use Magento\Framework\Phrase;
use Magento\Framework\Webapi\ServicePayloadConverterInterface;
use Magento\Framework\Communication\ConfigInterface as CommunicationConfig;

class MessageEncoder
{
    const DIRECTION_ENCODE = 'encode';
    const DIRECTION_DECODE = 'decode';

    private $dataObjectEncoder;

    private $dataObjectDecoder;

    private $jsonEncoder;

    private $jsonDecoder;

    private $communicationConfig;

    public function __construct(
        QueueConfig $queueConfig,
        \Magento\Framework\Json\EncoderInterface $jsonEncoder,
        \Magento\Framework\Json\DecoderInterface $jsonDecoder,
        \Magento\Framework\Webapi\ServiceOutputProcessor $dataObjectEncoder,
        \Magento\Framework\Webapi\ServiceInputProcessor $dataObjectDecoder
    ) {
        $this->dataObjectEncoder = $dataObjectEncoder;
        $this->dataObjectDecoder = $dataObjectDecoder;
        $this->jsonEncoder = $jsonEncoder;
        $this->jsonDecoder = $jsonDecoder;
    }

    public function encode($topic, $message, $requestType = true)
    {
        $convertedMessage = $this->convertMessage($topic, $message, self::DIRECTION_ENCODE, $requestType);
        return $this->jsonEncoder->encode($convertedMessage);
    }

    public function decode($topic, $message, $requestType = true)
    {
        try {
            $decodedMessage = $this->jsonDecoder->decode($message);
        } catch (\Exception $e) {
            throw new LocalizedException(new Phrase("Error occurred during message decoding."));
        }
        return $this->convertMessage($topic, $decodedMessage, self::DIRECTION_DECODE, $requestType);
    }

    protected function getTopicSchema($topic, $requestType)
    {
        $topicConfig = $this->getCommunicationConfig()->getTopic($topic);
        if ($topicConfig === null) {
            throw new LocalizedException(new Phrase('Specified topic "%topic" is not declared.', ['topic' => $topic]));
        }
        if ($requestType) {
            return [
                'schema_type' => $topicConfig[CommunicationConfig::TOPIC_REQUEST_TYPE],
                'schema_value' => $topicConfig[CommunicationConfig::TOPIC_REQUEST]
            ];
        } else {
            return [
                'schema_type' => isset($topicConfig[CommunicationConfig::TOPIC_RESPONSE])
                    ? CommunicationConfig::TOPIC_REQUEST_TYPE_CLASS
                    : null,
                'schema_value' => $topicConfig[CommunicationConfig::TOPIC_RESPONSE]
            ];
        }
    }

    protected function convertMessage($topic, $message, $direction, $requestType)
    {
        $topicSchema = $this->getTopicSchema($topic, $requestType);
        if ($topicSchema['schema_type'] == CommunicationConfig::TOPIC_REQUEST_TYPE_CLASS) {
            $messageDataType = $topicSchema[QueueConfig::TOPIC_SCHEMA_VALUE];
            try {
                $convertedMessage = $this->getConverter($direction)->convertValue($message, $messageDataType);
            } catch (LocalizedException $e) {
                throw new LocalizedException(
                    new Phrase(
                        'Message with topic "%topic" must be an instance of "%class".',
                        ['topic' => $topic, 'class' => $messageDataType]
                    )
                );
            }
        } else {
            $message = (array)$message;
            $isIndexedArray = array_keys($message) === range(0, count($message) - 1);
            $convertedMessage = [];
            foreach ($topicSchema['schema_value'] as $methodParameterMeta) {
                $paramName = $methodParameterMeta[CommunicationConfig::SCHEMA_METHOD_PARAM_NAME];
                $paramType = $methodParameterMeta[CommunicationConfig::SCHEMA_METHOD_PARAM_TYPE];
                if ($isIndexedArray) {
                    $paramPosition = $methodParameterMeta[CommunicationConfig::SCHEMA_METHOD_PARAM_POSITION];
                    if (isset($message[$paramPosition])) {
                        $convertedMessage[$paramName] = $this->getConverter($direction)
                            ->convertValue($message[$paramPosition], $paramType);
                    }
                } else {
                    if (isset($message[$paramName])) {
                        $convertedMessage[$paramName] = $this->getConverter($direction)
                            ->convertValue($message[$paramName], $paramType);
                    }
                }

                if ($methodParameterMeta[CommunicationConfig::SCHEMA_METHOD_PARAM_IS_REQUIRED]
                    && !isset($convertedMessage[$paramName])
                ) {
                    throw new LocalizedException(
                        new Phrase(
                            'Data item corresponding to "%param" must be specified '
                            . 'in the message with topic "%topic".',
                            [
                                'topic' => $topic,
                                'param' => $paramName
                            ]
                        )
                    );
                }
            }
        }
        return $convertedMessage;
    }

    protected function getConverter($direction)
    {
        return ($direction == self::DIRECTION_ENCODE) ? $this->dataObjectEncoder : $this->dataObjectDecoder;
    }

    private function getCommunicationConfig()
    {
        if ($this->communicationConfig === null) {
            $this->communicationConfig = \Magento\Framework\App\ObjectManager::getInstance()
                ->get(CommunicationConfig::class);
        }
        return $this->communicationConfig;
    }
}