Disque\Client

Complexity

Total Complexity

12

Size/Duplication

Total Lines	152
Duplicated Lines	0 %

Coupling/Cohesion

Components	3
Dependencies	23

Test Coverage

Coverage

0%

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
B	__construct()	0	30	2
A	setConnectionManager()	0	7	2
A	getConnectionManager()	0	4	1
A	isConnected()	0	4	1
A	connect()	0	4	1
A	__call()	0	12	2
A	registerCommand()	0	5	1
A	queue()	0	7	2

<?php
namespace Disque;

use Disque\Command;
use Disque\Command\CommandInterface;
use Disque\Command\InvalidCommandException;
use Disque\Connection\Manager;
use Disque\Connection\ManagerInterface;
use Disque\Queue\Queue;

/**
 * @method int ackJob(string... $ids)
 * @method string addJob(string $queue, string $payload, array $options = [])
 * @method int delJob(string... $ids)
 * @method int dequeue(string... $ids)
 * @method int enqueue(string... $ids)
 * @method int fastAck(string... $ids)
 * @method array getJob(string... $queues, array $options = [])
 * @method array hello()
 * @method string info()
 * @method array jscan(array $options = [])
 * @method int nack(string... $ids)
 * @method int qlen(string $queue)
 * @method array qpeek(string $queue, int $count)
 * @method array qscan(array $options = [])
 * @method array show(string $id)
 * @method int working(string $id)
 */
class Client
{
    /**
     * Connection manager
     *
     * @var ManagerInterface
     */
    protected $connectionManager;

    /**
     * Command handlers
     *
     * @var array
     */
    protected $commandHandlers = [];

    /**
     * List of built queues
     *
     * @var array
     */
    private $queues;

    /**
     * A list of credentials to Disque servers
     *
     * @var Disque\Connection\Credentials[]
     */
    private $servers;

    /**
     * Create a new Client
     *
     * @param Disque\Connection\Credentials[] $servers
     */
    public function __construct(array $servers = [])
    {
        foreach ([
            new Command\AckJob(),
            new Command\AddJob(),
            new Command\DelJob(),
            new Command\Dequeue(),
            new Command\Enqueue(),
            new Command\FastAck(),
            new Command\GetJob(),
            new Command\Hello(),
            new Command\Info(),
            new Command\JScan(),
            new Command\Nack(),
            new Command\Pause(),
            new Command\QLen(),
            new Command\QPeek(),
            new Command\QScan(),
            new Command\QStat(),
            new Command\Show(),
            new Command\Working()
        ] as $command) {
            $this->registerCommand($command);
        }

        $this->servers = $servers;

        $connectionManager = new Manager();
        $this->setConnectionManager($connectionManager);
    }

    /**
     * Set a connection manager
     *
     * @param ManagerInterface $manager
     */
    public function setConnectionManager(ManagerInterface $manager)
    {
        $this->connectionManager = $manager;
        foreach ($this->servers as $server) {
            $this->connectionManager->addServer($server);
        }
    }

    /**
     * Get the connection manager
     *
     * @return ManagerInterface Connection manager
     */
    public function getConnectionManager()
    {
        return $this->connectionManager;
    }

    /**
     * Tells if connection is established
     *
     * @return bool Success
     */
    public function isConnected()
    {
        return $this->connectionManager->isConnected();
    }

    /**
     * Connect to Disque
     *
     * @return Disque\Connection\Node\Node Connected node information
     * @throws Disque\Connection\ConnectionException
     */
    public function connect()
    {
        return $this->connectionManager->connect();
    }

    /**
     * @throws InvalidCommandException
     */
    public function __call($command, array $arguments)
    {
        $command = strtoupper($command);
        if (!isset($this->commandHandlers[$command])) {
            throw new InvalidCommandException($command);
        }

        $command = $this->commandHandlers[$command];
        $command->setArguments($arguments);
        $result = $this->connectionManager->execute($command);
        return $command->parse($result);
    }

    /**
     * Register a command handler
     *
     * @param CommandInterface $commandHandler Command
     * @return void
     */
    public function registerCommand(CommandInterface $commandHandler)
    {
        $command = strtoupper($commandHandler->getCommand());
        $this->commandHandlers[$command] = $commandHandler;
    }

    /**
     * Get a queue
     *
     * @param string $name Queue name
     * @return Queue Queue
     */
    public function queue($name)
    {
        if (!isset($this->queues[$name])) {
            $this->queues[$name] = new Queue($this, $name);
        }
        return $this->queues[$name];
    }
}