Dazzle\Channel\Channel

Complexity

Total Complexity

100

Size/Duplication

Total Lines	775
Duplicated Lines	13.68 %

Coupling/Cohesion

Components	2
Dependencies	21

Test Coverage

Coverage

70.92%

Importance

Changes

0

44 Methods

Rating	Name	Duplication	Size	Complexity
B	__construct()	0	29	2
A	__destruct()	0	20	1
A	getName()	0	4	1
A	getModel()	0	4	1
A	getRouter()	0	4	1
A	getInput()	0	4	1
A	getOutput()	0	4	1
A	createProtocol()	13	13	3
A	onStart()	0	4	1
A	onStop()	0	4	1
A	onConnect()	0	4	1
A	onDisconnect()	0	4	1
A	onInput()	0	4	1
A	onOutput()	0	4	1
A	start()	0	4	1
A	stop()	0	4	1
A	send()	9	9	4
A	push()	9	9	4
A	sendAsync()	13	13	4
A	pushAsync()	13	13	4
A	sendRequest()	13	13	4
A	pushRequest()	13	13	4
A	receive()	0	12	4
A	pull()	0	4	1
A	isStarted()	0	4	1
A	isStopped()	0	4	1
A	isConnected()	0	16	3
A	getConnected()	0	4	1
A	filterConnected()	0	4	1
A	handleSendAsync()	13	13	3
B	handlePushAsync()	0	25	4
A	handleSendRequest()	0	13	3
B	handlePushRequest()	0	35	5
A	handleReceive()	0	16	4
B	handleReceiveRequest()	0	19	5
B	handleReceiveResponse()	0	27	4
A	genID()	0	4	1
A	getTime()	0	4	1
A	getNextSuffix()	10	10	2
B	createMessageProtocol()	0	24	5
A	registerPeriodicTimers()	0	17	2
A	unregisterPeriodicTimers()	0	12	3
A	registerEvents()	0	5	1
A	unregisterEvents()	0	7	2

<?php

namespace Dazzle\Channel;

use Dazzle\Channel\Encoder\Encoder;
use Dazzle\Channel\Encoder\EncoderInterface;
use Dazzle\Channel\Extra\Request;
use Dazzle\Channel\Extra\Response;
use Dazzle\Channel\Model\ModelInterface;
use Dazzle\Channel\Protocol\Protocol;
use Dazzle\Channel\Protocol\ProtocolInterface;
use Dazzle\Channel\Record\RequestRecordStorage;
use Dazzle\Channel\Record\ResponseRecordStorage;
use Dazzle\Channel\Router\RouterCompositeInterface;
use Dazzle\Event\EventEmitter;
use Dazzle\Event\EventListener;
use Dazzle\Loop\Timer\TimerInterface;
use Dazzle\Loop\LoopAwareTrait;
use Dazzle\Loop\LoopInterface;
use Dazzle\Util\Support\GeneratorSupport;
use Dazzle\Util\Support\StringSupport;
use Dazzle\Util\Support\TimeSupport;
use Dazzle\Throwable\Exception\System\TaskIncompleteException;
use Dazzle\Throwable\Exception\Logic\InstantiationException;
use Dazzle\Throwable\Exception\LogicException;
use Dazzle\Throwable\Exception;
use Dazzle\Throwable\ThrowableProxy;

class Channel extends EventEmitter implements ChannelInterface
{
    use LoopAwareTrait;
    use RequestRecordStorage;
    use ResponseRecordStorage;

    /**
     * @var string
     */
    const TYPE_SND = 'SND';

    /**
     * @var string
     */
    const TYPE_REQ = 'REQ';

    /**
     * @var int
     */
    const BINDER = 1;

    /**
     * @var int
     */
    const CONNECTOR = 2;

    /**
     * @var int
     */
    const MODE_DEFAULT = 0;

    /**
     * @var int
     */
    const MODE_STANDARD = 0;

    /**
     * @var int
     */
    const MODE_BUFFER_ONLINE = 1;

    /**
     * @var int
     */
    const MODE_BUFFER_OFFLINE = 2;

    /**
     * @var int
     */
    const MODE_BUFFER = 3;

    /**
     * @var string
     */
    protected $name;

    /**
     * @var ModelInterface
     */
    protected $model;

    /**
     * @var RouterCompositeInterface
     */
    protected $router;

    /**
     * @var EncoderInterface
     */
    protected $encoder;

    /**
     * @var EventListener[]
     */
    protected $handlers;

    /**
     * @var string
     */
    protected $seed;

    /**
     * @var int
     */
    protected $counter;

    /**
     * @var TimerInterface
     */
    protected $reqsHelperTimer;

    /**
     * @var TimerInterface
     */
    protected $repsHelperTimer;

    /**
     * @param string $name
     * @param ModelInterface $model
     * @param RouterCompositeInterface $router
     * @param EncoderInterface $encoder
     * @param LoopInterface $loop
     * @throws InstantiationException
     */
    public function __construct($name, ModelInterface $model, RouterCompositeInterface $router, EncoderInterface $encoder, LoopInterface $loop)
    {
        parent::__construct($loop);

        try
        {
            $router->getBus('input');
            $router->getBus('output');
        }
        catch (Exception $ex)
        {
            throw new InstantiationException("Could not construct Dazzle\\Channel\\Channel due to Router wrong configuration.");
        }

        $this->name = $name;
        $this->model = $model;
        $this->router = $router;
        $this->encoder = $encoder;
        $this->loop = $loop;
        $this->handlers = [];
        $this->seed = GeneratorSupport::genId($this->name);
        $this->counter = 1e9;

The property $counter was declared of type integer, but 1000000000.0 is of type double. Maybe add a type cast?

        $this->reqsHelperTimer = null;
        $this->repsHelperTimer = null;
        $this->handledRepsTimeout = 10e3;

The property $handledRepsTimeout was declared of type integer, but 10000.0 is of type double. Maybe add a type cast?

        $this->registerEvents();
        $this->registerPeriodicTimers();
    }

    /**
     *
     */
    public function __destruct()
    {
        $this->unregisterEvents();
        $this->unregisterPeriodicTimers();

        unset($this->name);
        unset($this->model);
        unset($this->router);
        unset($this->encoder);
        unset($this->handlers);
        unset($this->seed);
        unset($this->counter);
        unset($this->reqsHelperTimer);
        unset($this->repsHelperTimer);
        unset($this->reqs);
        unset($this->reps);
        unset($this->handledReps);
        unset($this->handledRepsTimeout);
        unset($this->loop);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function getName()
    {
        return $this->name;
    }

    /**
     * @override
     * @inheritDoc
     */
    public function getModel()
    {
        return $this->model;
    }

    /**
     * @override
     * @inheritDoc
     */
    public function getRouter()
    {
        return $this->router;
    }

    /**
     * @override
     * @inheritDoc
     */
    public function getInput()
    {
        return $this->router->getBus('input');
    }

    /**
     * @override
     * @inheritDoc
     */
    public function getOutput()
    {
        return $this->router->getBus('output');
    }

    /**
     * @override
     * @inheritDoc
     */
    public function createProtocol($message = null)

This method seems to be duplicated in your project.

    {
        if ($message === null)
        {
            $message = '';
        }
        else if (!is_array($message))
        {
            $message = (string) $message;
        }

        return new Protocol('', $this->genID(), '', $this->name, $message, '', $this->getTime());

It seems like $message defined by parameter $message on line 236 can also be of type array; however, Dazzle\Channel\Protocol\Protocol::__construct() does only seem to accept string, maybe add an additional type check?

    }

    /**
     * @override
     * @inheritDoc
     */
    public function onStart(callable $handler)
    {
        return $this->on('start', $handler);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function onStop(callable $handler)
    {
        return $this->on('stop', $handler);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function onConnect(callable $handler)
    {
        return $this->on('connect', $handler);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function onDisconnect(callable $handler)
    {
        return $this->on('disconnect', $handler);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function onInput(callable $handler)
    {
        return $this->on('input', $handler);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function onOutput(callable $handler)
    {
        return $this->on('output', $handler);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function start()
    {
        $this->model->start();
    }

    /**
     * @override
     * @inheritDoc
     */
    public function stop()
    {
        $this->model->stop();
    }

    /**
     * @override
     * @inheritDoc
     */
    public function send($name, $message, $flags = Channel::MODE_DEFAULT, callable $success = null, callable $failure = null, callable $cancel = null, $timeout = 0.0)

This method seems to be duplicated in your project.

    {
        if ($success !== null || $failure !== null || $cancel !== null)
        {
            return $this->sendRequest($name, $message, $flags, $success, $failure, $cancel, $timeout);
        }

        return $this->sendAsync($name, $message, $flags);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function push($name, $message, $flags = Channel::MODE_DEFAULT, callable $success = null, callable $failure = null, callable $cancel = null, $timeout = 0.0)

This method seems to be duplicated in your project.

    {
        if ($success !== null || $failure !== null || $cancel !== null)
        {
            return $this->pushRequest($name, $message, $flags, $success, $failure, $cancel, $timeout);
        }

        return $this->pushAsync($name, $message, $flags);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function sendAsync($name, $message, $flags = Channel::MODE_DEFAULT)

This method seems to be duplicated in your project.

    {
        $protocol = $this->createMessageProtocol($message);
        $names = (array) $name;
        $handlers = [];

        foreach ($names as $name)
        {
            $handlers[] = $this->handleSendAsync($name, $protocol, $flags);
        }

        return !isset($handlers[0]) || isset($handlers[1]) ? $handlers : $handlers[0];
    }

    /**
     * @override
     * @inheritDoc
     */
    public function pushAsync($name, $message, $flags = Channel::MODE_DEFAULT)

This method seems to be duplicated in your project.

    {
        $protocol = $this->createMessageProtocol($message);
        $names = (array) $name;
        $handlers = [];

        foreach ($names as $name)
        {
            $handlers[] = $this->handlePushAsync($name, $protocol, $flags);
        }

        return !isset($handlers[0]) || isset($handlers[1]) ? $handlers : $handlers[0];
    }

    /**
     * @override
     * @inheritDoc
     */
    public function sendRequest($name, $message, $flags = Channel::MODE_DEFAULT, callable $success = null, callable $failure = null, callable $cancel = null, $timeout = 0.0)

This method seems to be duplicated in your project.

    {
        $protocol = $this->createMessageProtocol($message);
        $names = (array) $name;
        $handlers = [];

        foreach ($names as $name)
        {
            $handlers[] = $this->handleSendRequest($name, $protocol, $flags, $success, $failure, $cancel, $timeout);
        }

        return !isset($handlers[0]) || isset($handlers[1]) ? $handlers : $handlers[0];
    }

    /**
     * @override
     * @inheritDoc
     */
    public function pushRequest($name, $message, $flags = Channel::MODE_DEFAULT, callable $success = null, callable $failure = null, callable $cancel = null, $timeout = 0.0)

This method seems to be duplicated in your project.

    {
        $protocol = $this->createMessageProtocol($message);
        $names = (array) $name;
        $handlers = [];

        foreach ($names as $name)
        {
            $handlers[] = $this->handlePushRequest($name, $protocol, $flags, $success, $failure, $cancel, $timeout);
        }

        return !isset($handlers[0]) || isset($handlers[1]) ? $handlers : $handlers[0];
    }

    /**
     * @override
     * @inheritDoc
     */
    public function receive($sender, ProtocolInterface $protocol)
    {
        if ($this->handleReceiveRequest($protocol))
        {
            return;
        }

        if ($this->handleReceiveResponse($protocol) || $this->getInput()->handle($sender, $protocol))
        {
            $this->emit('input', [ $sender, $protocol ]);
        }
    }

    /**
     * @override
     * @inheritDoc
     */
    public function pull($sender, ProtocolInterface $protocol)
    {
        $this->emit('input', [ $sender, $protocol ]);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function isStarted()
    {
        return $this->model->isStarted();
    }

    /**
     * @override
     * @inheritDoc
     */
    public function isStopped()
    {
        return $this->model->isStopped();
    }

    /**
     * @override
     * @inheritDoc
     */
    public function isConnected($name)
    {
        if (is_array($name))
        {
            $statuses = [];

            foreach ($name as $singleName)
            {
                $statuses[] = $this->model->isConnected($singleName);
            }

            return $statuses;
        }

        return $this->model->isConnected($name);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function getConnected()
    {
        return $this->model->getConnected();
    }

    /**
     * @override
     * @inheritDoc
     */
    public function filterConnected($pattern)
    {
        return StringSupport::find($pattern, $this->getConnected());
    }

    /**
     * @override
     * @inheritDoc
     */
    protected function handleSendAsync($name, $message, $flags = Channel::MODE_DEFAULT)

This method seems to be duplicated in your project.

    {
        if ($message->getType() === '')
        {
            $message->setType(Channel::TYPE_SND);
        }
        if ($message->getDestination() === '')
        {
            $message->setDestination($name);
        }

        return $this->getOutput()->handle($name, $message, $flags);
    }

    /**
     * @override
     * @inheritDoc
     */
    protected function handlePushAsync($name, $message, $flags = Channel::MODE_DEFAULT)
    {
        if ($message->getType() === '')
        {
            $message->setType(Channel::TYPE_SND);
        }
        if ($message->getDestination() === '')
        {
            $message->setDestination($name);
        }

        $status = $this->model->unicast(
            $name,
            $this->encoder->with($message)->encode(),
            $flags
        );

        if ($status)
        {
            $this->resolveOrRejectResponse($message->getPid(), $message->getException());
            $this->emit('output', [ $name, $message ]);
        }

        return $status;
    }

    /**
     * @override
     * @inheritDoc
     */
    protected function handleSendRequest($name, $message, $flags = Channel::MODE_DEFAULT, callable $success = null, callable $failure = null, callable $cancel = null, $timeout = 0.0)
    {
        if ($message->getType() === '')
        {
            $message->setType(Channel::TYPE_REQ);
        }
        if ($message->getDestination() === '')
        {
            $message->setDestination($name);
        }

        return $this->getOutput()->handle($name, $message, $flags, $success, $failure, $cancel, $timeout);
    }

    /**
     * @override
     * @inheritDoc
     */
    protected function handlePushRequest($name, $message, $flags = Channel::MODE_DEFAULT, callable $success = null, callable $failure = null, callable $cancel = null, $timeout = 0.0)
    {
        if ($message->getType() === '')
        {
            $message->setType(Channel::TYPE_REQ);
        }
        if ($message->getDestination() === '')
        {
            $message->setDestination($name);
        }

        $status = $this->model->unicast(
            $name,
            $this->encoder->with($message)->encode(),
            $flags
        );

        if (!$status)
        {
            if ($cancel !== null)
            {
                $cancel(new LogicException('Request could not be sent.'));
            }

            return null;
        }

        $pid = $message->getPid();
        $request = $this->createRequest($pid, $success, $failure, $cancel, $timeout);
        $this->addRequest($pid, $request);

        $this->emit('output', [ $name, $message ]);

        return $request;
    }

    /**
     * @override
     * @inheritDoc
     */
    public function handleReceive($sender, $multipartMessage)
    {
        $protocol = $this->encoder
            ->with(new Protocol())
            ->decode(implode('', $multipartMessage));

        if ($this->handleReceiveRequest($protocol))
        {
            return;
        }

        if ($this->handleReceiveResponse($protocol) || $this->getInput()->handle($sender, $protocol))
        {
            return;
        }
    }

    /**
     * @param ProtocolInterface
     * @return bool
     */
    protected function handleReceiveRequest(ProtocolInterface $protocol)
    {
        if ($protocol->getType() === Channel::TYPE_REQ && $protocol->getDestination() === $this->name)
        {
            $pid = $protocol->getPid();
            $timestamp = $protocol->getTimestamp();
            $now = $this->getTime();

            if ($timestamp <= $now || $this->existsResponse($pid))
            {
                return true;
            }

            $timestamp -= 5e3;
            $this->addResponse($pid, $this->createResponse($pid, $protocol->getOrigin(), $timestamp, $timestamp - $now));
        }

        return false;
    }

    /**
     * @param ProtocolInterface $protocol
     * @return bool
     */
    protected function handleReceiveResponse(ProtocolInterface $protocol)
    {
        $pid = $protocol->getPid();

        if (!$this->existsRequest($pid))
        {
            return false;
        }

        $message = $protocol->getMessage();
        $exception = $protocol->getException();

        if ($exception === '')
        {
            $this->resolveRequest($pid, $message);
        }
        else if ($exception === TaskIncompleteException::class)
        {
            $this->cancelRequest($pid, new ThrowableProxy([ $exception, $message ]));
        }
        else
        {
            $this->rejectRequest($pid, new ThrowableProxy([ $exception, $message ]));
        }

        return true;
    }

    /**
     * @return string
     */
    protected function genID()
    {
        return $this->seed . $this->getNextSuffix();
    }

    /**
     * @return float
     */
    protected function getTime()
    {
        return TimeSupport::now();
    }

    /**
     * @return string
     */
    protected function getNextSuffix()

This method seems to be duplicated in your project.

    {
        if ($this->counter > 2e9)
        {
            $this->counter = 1e9;

The property $counter was declared of type integer, but 1000000000.0 is of type double. Maybe add a type cast?

            $this->seed = GeneratorSupport::genId($this->name);
        }

        return (string) $this->counter++;
    }

    /**
     * @param string|string[] $message
     * @return ProtocolInterface
     */
    protected function createMessageProtocol($message)
    {
        if (!($message instanceof ProtocolInterface))
        {
            $message = $this->createProtocol($message);
        }
        else
        {
            if ($message->getPid() === '')
            {
                $message->setPid($this->genID());
            }
            if ($message->getOrigin() === '')
            {
                $message->setOrigin($this->name);
            }
            if ($message->getTimestamp() == 0)
            {
                $message->setTimestamp($this->getTime());
            }
        }

        return $message;
    }

    /**
     *
     */
    private function registerPeriodicTimers()
    {
        $this->reqsHelperTimer = $this->getLoop()->addPeriodicTimer(0.1, function() {
            $this->expireRequests();
        });
        $this->repsHelperTimer = $this->getLoop()->addPeriodicTimer(0.1, function() {
            $this->expireResponses();
            $unfinished = $this->unfinishedResponses();

            foreach ($unfinished as $response)
            {
                $protocol = new Protocol('', $response->getPid(), '', $response->getAlias(), '', '', $this->getTime());
                $response = new Response($this, $protocol, new TaskIncompleteException("Task unfinished."));
                $response->call();
            }
        });
    }

    /**
     *
     */
    private function unregisterPeriodicTimers()
    {
        if ($this->reqsHelperTimer !== null)
        {
            $this->reqsHelperTimer->cancel();
        }

        if ($this->repsHelperTimer !== null)
        {
            $this->repsHelperTimer->cancel();
        }
    }

    /**
     *
     */
    private function registerEvents()
    {
        $this->handlers = $this->model->copyEvents($this, [ 'start', 'stop', 'connect', 'disconnect' ]);
        $this->handlers[] = $this->model->on('recv', [ $this, 'handleReceive' ]);
    }

    /**
     *
     */
    private function unregisterEvents()
    {
        foreach ($this->handlers as $handler)
        {
            $handler->cancel();
        }
    }
}