Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis\Driver

Complexity

Total Complexity

9

Size/Duplication

Total Lines	72
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	3

Test Coverage

Coverage

0%

Importance

Changes

0

7 Methods

Rating	Name	Duplication	Size	Complexity
A	getConsumer()	0	8	1
A	getQueueCallback()	0	8	3
A	acceptMessage()	0	4	1
A	decodeMessage()	0	4	1
A	getProducer()	0	7	1
A	getQueueManager()	0	6	1
A	setDebug()	0	6	1

<?php

namespace Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis;

use Kaliop\QueueingBundle\Adapter\DriverInterface;
use Kaliop\QueueingBundle\Queue\QueueManagerAwareTrait;
use Symfony\Component\DependencyInjection\ContainerAwareTrait;
use Symfony\Component\DependencyInjection\ContainerAwareInterface;

class Driver implements DriverInterface, ContainerAwareInterface
{
    use ContainerAwareTrait;
    use QueueManagerAwareTrait;

    protected $debug;

    public function getConsumer($queueName)
    {
        $consumer = $this->container->get('kaliop_queueing.kinesis.consumer');

The expression $this->container->get('kaliop_queueing.kinesis.consumer'); of type object|null adds the type object to the return on line 23 which is incompatible with the return type declared by the interface Kaliop\QueueingBundle\Adapter\DriverInterface::getConsumer of type string.

        $callback = $this->getQueueCallback($queueName);
        $consumer->setQueueName($queueName);
        $consumer->setCallback($callback);
        return $consumer;
    }

    protected function getQueueCallback($queueName)
    {
        $callbacks = $this->container->getParameter('kaliop_queueing_kinesis.default.consumers');
        if (!isset($callbacks[$queueName]) || !isset($callbacks[$queueName]['callback'])) {
            throw new \UnexpectedValueException("No callback has been defined for queue '$queueName', please check config parameter 'kaliop_queueing_kinesis.default.consumers'");
        }
        return $this->container->get($callbacks[$queueName]['callback']);
    }

    public function acceptMessage($message)
    {
        return $message instanceof \Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis\Message;
    }

    /**
     * Unlike the RabbitMQ driver, we do not have to deal with a native message type from the underlying library.
     * So we just let the Producer create messages of the good type, and decoding them becomes a no-op
     *
     * @param \Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis\Message $message
     * @return \Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis\Message
     */
    public function decodeMessage($message)
    {
        return $message;
    }

    /**
     * @param string $queueName
     * @return \Kaliop\QueueingBundle\Queue\ProducerInterface
     */
    public function getProducer($queueName)
    {
        $producer = $this->container->get('kaliop_queueing.kinesis.producer');
        $producer->setQueueName($queueName);
        $producer->setDebug($this->debug);
        return $producer;
    }

    /**
     * @param string $queueName
     * @return \Kaliop\QueueingBundle\Queue\QueueManagerInterface
     */
    public function getQueueManager($queueName)
    {
        $mgr = $this->getQueueManagerInternal();
        $mgr->setQueueName($queueName);
        return $mgr;
    }

    public function setDebug($debug)
    {
        $this->debug = $debug;

        return $this;
    }
}