Dazzle\Redis\Dispatcher

Complexity

Total Complexity

28

Size/Duplication

Total Lines	234
Duplicated Lines	0 %

Coupling/Cohesion

Components	2
Dependencies	8

Importance

Changes

0

13 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	12	1
A	__destruct()	0	11	2
A	createClient()	0	15	3
A	getDriver()	0	4	1
A	isEnding()	0	4	2
A	appendRequest()	0	4	1
A	watch()	0	17	4
A	handleConnect()	0	5	1
A	handleData()	0	20	4
A	handleRequest()	0	4	1
A	handleResponse()	0	18	4
A	handleDisconnect()	0	12	2
A	handleClose()	0	12	2

<?php

namespace Dazzle\Redis;

use Dazzle\Socket\Socket;
use Dazzle\Promise\Deferred;
use Dazzle\Loop\LoopInterface;
use Dazzle\Redis\Driver\Driver;
use Dazzle\Socket\SocketInterface;
use Dazzle\Event\AsyncEventEmitter;
use Dazzle\Redis\Driver\DriverInterface;
use Dazzle\Throwable\Exception\Runtime\ExecutionException;
use Dazzle\Throwable\Exception\Runtime\UnderflowException;
use Error;
use Exception;

use Clue\Redis\Protocol\Model\ErrorReply;
use Clue\Redis\Protocol\Model\ModelInterface;
use Clue\Redis\Protocol\Parser\ParserException;

class Dispatcher extends AsyncEventEmitter
{
    /**
     * Stream
     * @var SocketInterface
     */
    private $stream;

    /**
     * Deferred requests
     * @var array
     */
    private $reqs;

    /**
     * Driver of RESP protocol
     * @var DriverInterface
     */
    private $protocol;

    /**
     * Flag of ending
     * @var bool
     */
    private $ending;

    /**
     * Flag of closed
     * @var bool
     */
    private $closed;

    /**
     * Constructor
     * @param LoopInterface $loop
     */
    public function __construct(LoopInterface $loop)
    {
        parent::__construct($loop);
        $this->reqs = [];
        $this->ending = false;
        $this->closed = false;
        $this->protocol = new Driver();
        $this->on('connect', [$this, 'handleConnect']);
        $this->on('response',[$this, 'handleResponse']);
        $this->on('disconnect',[$this, 'handleDisconnect']);
        $this->on('close', [$this, 'handleClose']);
    }

    /**
     * Destructor
     */
    public function __destruct()
    {
        if ($this->ending != true)

It seems like you are loosely comparing two booleans. Considering using the strict comparison !== instead.

        {
            $this->handleDisconnect();        
        }
        else 
        {
            $this->handleClose();        
        }
    }

     /**
     * Create socket client with connection to Redis database.
     *
     * @param string $endpoint
     * @return SocketInterface
     * @throws ExecutionException
     */
    protected function createClient($endpoint)
    {
        $ex = null;

        try
        {
            return new Socket($endpoint, $this->loop);

It seems like $this->loop can be null; however, __construct() does not accept null, maybe add an additional type check?

        }
        catch (Error $ex)

Consider adding a comment why this CATCH block is empty.

        {}
        catch (Exception $ex)

Consider adding a comment why this CATCH block is empty.

        {}

        throw new ExecutionException('Redis connection socket could not be created!', 0, $ex);
    }

    /**
     * Get RESP Driver
     * @return DriverInterface
     */
    public function getDriver()
    {
        return $this->protocol;
    }

    /**
     * Get dispatcher ending flag
     * @return bool
     */
    public function isEnding()
    {
        return $this->ending? true: false;
    }

    /**
     * Append request
     * @param $req
     */
    public function appendRequest($req)
    {
        $this->reqs[] = $req;
    }

    /**
     * Watch and dispatch streaming
     * @param $endpoint
     */
    public function watch($endpoint)
    {
        if ($this->stream !== null) {
            return;
        }

        try {
            $this->stream = $this->createClient($endpoint);
        } catch (\Exception $e) {
            $this->emit('error', [$e]);
        }

        if ($this->stream->isOpen()) {
            $this->emit('connect', [$this]);
        }

    }

    /**
     * @internal
     */
    public function handleConnect()
    {
        $this->stream->on('data', [$this, 'handleData']);
        $this->emit('request');
    }

    /**
     * @internal
     * @param $_
     * @param $data
     */
    public function handleData($_, $data)
    {
        try {
            $models = $this->protocol->parseResponse($data);
        } catch (ParserException $error) {
            $this->emit('error', [$error]);

            return;
        }

        foreach ($models as $data) {
            try {
                $this->emit('response', [$data]);
            } catch (UnderflowException $error) {
                $this->emit('error', [$error]);

                break;
            }
        }
    }

    /**
     * @internal
     * @param $payload
     */
    public function handleRequest($payload)
    {
        $this->stream->write($payload);
    }

    /**
     * @internal
     */
    public function handleResponse(ModelInterface $message)
    {
        if (!$this->reqs) {

The expression $this->reqs of type array is implicitly converted to a boolean; are you sure this is intended? If so, consider using empty($expr) instead to make it clear that you intend to check for an array without elements.

            throw new UnderflowException('Unexpected reply received, no matching request found');
        }
        /* @var Deferred $req */
        $req = array_shift($this->reqs);

        if ($message instanceof ErrorReply) {
            $req->reject($message);
        } else {
            $req->resolve($message->getValueNative());
        }

        if (count($this->reqs) <= 0) {
            $this->emit('disconnect');
        }
    }

    /**
     * @internal
     */
    public function handleDisconnect()
    {
        $this->ending = true;
        // reject all remaining requests in the queue
        while($this->reqs) {

The expression $this->reqs of type array is implicitly converted to a boolean; are you sure this is intended? If so, consider using ! empty($expr) instead to make it clear that you intend to check for an array without elements.

            $req = array_shift($this->reqs);
            /* @var $req Deferred */
            $req->reject(new RuntimeException('Connection closing'));
        }
        $this->stream->close();
        $this->emit('close');
    }

    /**
     * @internal
     */
    public function handleClose()
    {
        if ($this->closed) {
            return;
        }
        $this->removeListeners('connect');
        $this->removeListeners('response');
        $this->removeListeners('disconnect');
        $this->removeListeners('close');
        $this->removeListeners('error');
        $this->closed = true;
    }
}