Cmobi\RabbitmqBundle\Rpc\RpcClient

Complexity

Total Complexity

12

Size/Duplication

Total Lines	118
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes	3
Bugs	1	Features	0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	7	1
A	onResponse()	0	6	2
B	call()	0	30	2
A	setMessage()	0	4	1
A	getMessage()	0	4	1
A	setChannel()	0	4	1
A	getChannel()	0	4	1
A	getQueueName()	0	4	1
A	getConnection()	0	4	1
A	setConnection()	0	4	1

<?php

namespace Cmobi\RabbitmqBundle\Rpc;

use Cmobi\RabbitmqBundle\ConnectionManagerInterface;
use PhpAmqpLib\Channel\AbstractChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

abstract class RpcClient
{
    private $queue;
    private $connection;
    private $body;
    private $channel;
    private $callbackQueue;
    private $response;
    private $correlationId;

    public function __construct($queueName, ConnectionManagerInterface $manager)
    {
        $this->body = '';
        $this->queue = $queueName;
        $this->connection = $manager->getConnection();
        $this->channel = $this->connection->channel();
    }

    /**
     * @param AMQPMessage $rep
     */
    public function onResponse(AMQPMessage $rep)
    {
        if($rep->get('correlation_id') == $this->correlationId) {
            $this->response = $rep->body;
        }
    }

    /**
     * @return null|string
     */
    public function call()
    {
        list($callbackQueue, ,) = $this->getChannel()->queue_declare(
            '', false, false, false, true
        );
        $this->callbackQueue = $callbackQueue;
        $this->getChannel()->basic_consume(
            $this->callbackQueue, '', false, false, false, false,
            [$this, 'onResponse']
        );
        $this->response = null;
        $this->correlationId = uniqid();

        $msg = new AMQPMessage(
            $this->getMessage(),
            [
                'correlation_id' => $this->correlationId,
                'reply_to' => $this->callbackQueue
            ]
        );
        $this->getChannel()->basic_publish($msg, '', $this->getQueueName());

        while(!$this->response) {
            $this->getChannel()->wait();
        }
        $this->getChannel()->close();
        $this->getConnection()->close();
        
        return $this->response;
    }

    /**
     * @param string $body
     */
    public function setMessage($body)
    {
        $this->body = (string)$body;
    }

    /**
     * @return string
     */
    public function getMessage()
    {
        return $this->body;
    }

    /**
     * @param AbstractChannel $channel
     */
    public function setChannel(AbstractChannel $channel)
    {
        $this->channel = $channel;
    }

    /**
     * @return \PhpAmqpLib\Channel\AMQPChannel
     */
    public function getChannel()
    {
        return $this->channel;
    }

    /**
     * @return string
     */
    public function getQueueName()
    {
        return $this->queue;
    }

    /**
     * @return AMQPStreamConnection
     */
    public function getConnection()
    {
        return $this->connection;
    }

    /**
     * @param \PhpAmqpLib\Connection\AMQPStreamConnection $connection
     */
    public function setConnection(AMQPStreamConnection $connection)
    {
        $this->connection = $connection;
    }
}