OldSound\RabbitMqBundle\ReceiverExecutor\ReplyReceiverExecutor

Complexity

Total Complexity

9

Size/Duplication

Total Lines	51
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	sendReply()	0	7	1
A	execute()	0	24	6
A	support()	0	3	1
A	__construct()	0	3	1

<?php

namespace OldSound\RabbitMqBundle\ReceiverExecutor;

use OldSound\RabbitMqBundle\Declarations\RpcConsumeOptions;
use OldSound\RabbitMqBundle\Receiver\ReceiverException;
use OldSound\RabbitMqBundle\Receiver\ReceiverInterface;
use OldSound\RabbitMqBundle\Receiver\ReplyReceiverInterface;

class ReplyReceiverExecutor implements ReceiverExecutorInterface
{
    /** @var RpcConsumeOptions */
    protected $options;

    public function __construct(RpcConsumeOptions $options)
    {
        $this->options = $options;
    }

    /**
     * @param $receiver ReplyReceiverInterface
     */
    public function execute(array $messages, $receiver): array
    {
        if (!$this->support($receiver)) {
            throw new \InvalidArgumentException('TOOD');
        }

        if (count($messages) !== 1) {
            throw new \InvalidArgumentException('todo');
        }

        $message = $messages[0];

        if (!($message->get($this->options->replayToProperty) && $message->get($this->options->correlationIdProperty))) {
            throw new \InvalidArgumentException('todo'); // TODO
        }

        try {
            $reply = $receiver->execute($message);
        } catch (ReceiverException $exception) {
            return [$exception->getCode()];
        }
        $this->sendReply($message->getChannel(), $reply, $message->get($this->options->replayToProperty), $message->get($this->options->correlationIdProperty));

        return [ReceiverInterface::MSG_ACK];
    }

    protected function sendReply(\AMQPChannel $channel, $reply, $replyTo, $correlationId)
    {
        $body = $this->serializer->serialize($reply);

The property serializer does not exist on OldSound\RabbitMqBundle\ReceiverExecutor\ReplyReceiverExecutor. Did you maybe forget to declare it?

        $message = new AMQPMessage($body, ['content_type' => 'text/plain'] + $this->options->replayMessageProperties + [

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

            $this->options->correlationIdProperty => $correlationId,
        ]);
        $channel->basic_publish($message , '', $replyTo);

The method basic_publish() does not exist on AMQPChannel.

    }

    public function support($receiver): bool
    {
        return $receiver instanceof ReplyReceiverInterface;
    }
}