OldSound\RabbitMqBundle\RPC\RpcClient

Complexity

Total Complexity

12

Size/Duplication

Total Lines	126
Duplicated Lines	0 %

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	declareRepliesQueue()	0	9	1
A	batchExecute()	0	6	2
A	createAnonQueueDeclaration()	0	10	1
A	getReplies()	0	35	5
A	__construct()	0	7	1
A	addRequest()	0	22	2

<?php

namespace OldSound\RabbitMqBundle\RPC;

use OldSound\RabbitMqBundle\Declarations\BindingDeclaration;
use OldSound\RabbitMqBundle\Declarations\Declarator;
use OldSound\RabbitMqBundle\Declarations\ConsumeOptions;
use OldSound\RabbitMqBundle\Declarations\QueueDeclaration;
use OldSound\RabbitMqBundle\ExecuteCallbackStrategy\BatchExecuteCallbackStrategy;

The type OldSound\RabbitMqBundle\ExecuteCallbackStrategy\BatchExecuteCallbackStrategy was not found. Maybe you did not declare it correctly or list all dependencies?

use OldSound\RabbitMqBundle\Producer\ProducerInterface;
use OldSound\RabbitMqBundle\RabbitMq\Exception\RpcResponseException;

The type OldSound\RabbitMqBundle\RabbitMq\Exception\RpcResponseException was not found. Maybe you did not declare it correctly or list all dependencies?

use OldSound\RabbitMqBundle\Serializer\JsonMessageBodySerializer;
use OldSound\RabbitMqBundle\Serializer\MessageBodySerializerInterface;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;

The type Symfony\Component\Messenger\Transport\Serialization\SerializerInterface was not found. Maybe you did not declare it correctly or list all dependencies?

use Symfony\Component\Serializer\Serializer;

class RpcClient implements BatchReceiverInterface

The type OldSound\RabbitMqBundle\RPC\BatchReceiverInterface was not found. Maybe you did not declare it correctly or list all dependencies?

{
    const PROPERTY_REPLAY_TO = 'reply_to';
    const PROPERTY_CORRELATION_ID = 'correlation_id';

    /** @var AMQPChannel */
    private $channel;
    /** @var int */
    private $expiration;

    /** @var QueueDeclaration */
    private $anonRepliesQueue;
    /** @var MessageBodySerializerInterface[] */
    private $serializers;

    /** @var int */
    private $requests = 0;
    /** @var AMQPMessage[] */
    private $messages;
    private $replies = [];

The private property $replies is not used, and could be removed.

    public function __construct(
        AMQPChannel $channel,
        int $expiration = 10000
    ) {
        $this->channel = $channel;
        $this->serializer = $serializer ?? new JsonMessageBodySerializer();

The variable $serializer seems to never exist and therefore isset should always be false.

The property serializer does not exist. Although not strictly required by PHP, it is generally a best practice to declare properties explicitly.

        $this->expiration = $expiration;
    }

    public function declareRepliesQueue($repliesQueueName = null): RpcClient
    {
        $this->anonRepliesQueue = $this->createAnonQueueDeclaration();
        $this->anonRepliesQueue->name = $repliesQueueName;
        $declarator = new Declarator($this->channel);
        [$queueName] = $declarator->declareQueues([$this->anonRepliesQueue]);
        $this->anonRepliesQueue->name = $queueName;

        return $this;
    }

    // TODO public move
    private function createAnonQueueDeclaration(): QueueDeclaration
    {
        $anonQueueDeclaration = new QueueDeclaration();
        $anonQueueDeclaration->passive = false;
        $anonQueueDeclaration->durable = false;
        $anonQueueDeclaration->exclusive = true;
        $anonQueueDeclaration->autoDelete = true;
        $anonQueueDeclaration->nowait = false;

        return $anonQueueDeclaration;
    }

    public function addRequest($msgBody, $rpcQueue, MessageBodySerializerInterface $serializer = null)
    {
        if (!$this->anonRepliesQueue) {
            throw new \LogicException('no init anonRepliesQueue');
        }

        $correlationId = $this->requests;
        $this->serializers[$correlationId] = $serializer;

        $serializer = $serializer ?? new JsonMessageBodySerializer();

        $replyToQueue = $this->anonRepliesQueue->name; // 'amq.rabbitmq.reply-to';
        $msg = new AMQPMessage($serializer->serialize($msgBody, 'json'), [

The call to OldSound\RabbitMqBundle\Serializer\JsonMessageBodySerializer::serialize() has too many arguments starting with 'json'.

The call to OldSound\RabbitMqBundle\Serializer\MessageBodySerializerInterface::serialize() has too many arguments starting with 'json'.

            self::PROPERTY_REPLAY_TO => $replyToQueue,
            self::PROPERTY_CORRELATION_ID => $correlationId,
            'content_type' => 'text/plain',
            'delivery_mode' => ProducerInterface::DELIVERY_MODE_NON_PERSISTENT,
            'expiration' => $this->expiration,
        ]);

        $this->channel->basic_publish($msg, '', $rpcQueue);
        $this->requests++;
    }

    public function batchExecute(array $messages)
    {
        if ($this->messages !== null) {
            throw new \LogicException('Rpc client consming should be called once by batch count limit');
        }
        $this->messages = $messages;
    }

    /**
     * @param $name
     * @param MessageBodySerializerInterface $serializer
     * @return array|AMQPMessage[]
     */
    public function getReplies($name): array
    {
        if (0 === $this->requests) {
            throw new \LogicException('request empty');
        }

        $consumer = new Consumer($this->channel);

The type OldSound\RabbitMqBundle\RPC\Consumer was not found. Maybe you did not declare it correctly or list all dependencies?

        $consuming = new ConsumeOptions();
        $consuming->exclusive = true;
        $consuming->qosPrefetchCount = $this->requests;
        $consuming->queue = $this->anonRepliesQueue->name;
        $consuming->receiver = $this;
        $consumer->consumeQueue($consuming, new BatchExecuteCallbackStrategy($this->requests));

        try {
            $consumer->consume($this->requests);
        } finally {
            // TODO $this->getChannel()->basic_cancel($consumer_tag);
        }

        $replices = [];
        foreach($this->messages as $message) {
            /** @var AMQPMessage $message */
            if (!$message->has('correlation_id')) {
                $this->logger->error('unexpected message. rpc replies have no correlation_id ');
                continue;
            }

            $correlationId = $message->get('correlation_id');
            $serializer = $this->serializers[$correlationId];
            $reply = $serializer ? $serializer->deserialize($message->body) : $message;
            $replices[$correlationId] = $reply;
        }
        ksort($replices);
        return $replices;
    }
}