LogicalSteps\Async\Async

Complexity

Total Complexity

54

Size/Duplication

Total Lines	295
Duplicated Lines	14.24 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes

0

15 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	promise()	0	12	2
A	execute()	0	14	3
A	_execute()	0	4	1
A	_executeReactLoop()	0	6	1
A	_executeAmpLoop()	0	7	1
D	_run()	42	97	24
A	handlePromise()	0	20	4
B	logCallable()	0	25	6
B	logGenerator()	0	23	6
A	format()	0	4	1
A	autoloaded()	0	4	1
A	setLogger()	0	4	1
A	setLoop()	0	5	1
A	useAmpLoop()	0	4	1

<?php

namespace LogicalSteps\Async;

use Closure;
use Error;
use Generator;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use React\EventLoop\LoopInterface;
use React\Promise\Deferred;
use React\Promise\Promise;
use ReflectionGenerator;
use Throwable;

class Async implements LoggerAwareInterface
{
    const PROMISE_REACT = 'React\Promise\PromiseInterface';
    const PROMISE_GUZZLE = 'GuzzleHttp\Promise\PromiseInterface';
    const PROMISE_HTTP = 'Http\Promise\Promise';
    const PROMISE_AMP = 'Amp\Promise';
    /**
     * @var LoopInterface
     */
    protected $loop;
    /**
     * @var LoggerInterface
     */
    protected $logger;

    /**
     * @var callable that maps to _execute or _executeReactLoop ...
     */
    protected $exec;

    public function __construct()
    {
        $this->logger = new EchoLogger();
        $this->exec = [$this, '_execute'];
    }

    public function promise(Generator $flow): Promise
    {
        $deferred = new Deferred();
        $this->execute($flow, function ($error, $result) use ($deferred) {
            if ($error) {
                return $deferred->reject($error);
            }
            $deferred->resolve($result);
        });

        return $deferred->promise();
    }

    public function execute(Generator $flow, callable $callback = null)
    {
        $this->logger->info('start');
        $wrapped_callback = function ($error, $result) use ($callback) {
            if ($this->logger) {
                $this->logger->info('end');
            }
            if (is_callable($callback)) {
                $callback($error, $result);
            }
        };
        ($this->exec)($flow, $wrapped_callback);

    }

    private function _execute(Generator $flow, callable $callback = null, int $depth = 0)
    {
        $this->_run($flow, $callback, $depth);
    }

    private function _executeReactLoop(Generator $flow, callable $callback = null, int $depth = 0)
    {
        $this->loop->futureTick(function () use ($flow, $callback, $depth) {
            $this->_run($flow, $callback, $depth);
        });
    }

    private function _executeAmpLoop(Generator $flow, callable $callback = null, int $depth = 0)
    {

        ('\Amp\Loop::defer')(function () use ($flow, $callback, $depth) {
            $this->_run($flow, $callback, $depth);
        });
    }


    private function _run(Generator $flow, callable $callback = null, int $depth = 0)
    {
        try {
            if (!$flow->valid()) {
                $value = $flow->getReturn();
                if ($value instanceof Generator) {
                    $this->logGenerator($value, $depth);
                    ($this->exec)($value, $callback, $depth + 1);
                } elseif (is_callable($callback)) {
                    $callback(null, $value);
                }

                return;
            }
            $value = $flow->current();
            $args = [];
            $func = [];
            if (is_array($value) && count($value) > 1) {
                $func[] = array_shift($value);
                if (is_callable($func[0])) {
                    $func = $func[0];
                } else {
                    $func[] = array_shift($value);
                }
                $args = $value;
            } else {
                $func = $value;
            }
            if (is_callable($func)) {
                $this->logCallable($func, $args, $depth);
                $args[] = function ($error, $result) use ($flow, $callback, $depth) {

This code seems to be duplicated across your project.

                    if ($error) {
                        if ($this->logger) {
                            $this->logger->error((string)$error, compact('depth'));
                        }
                        if (is_callable($callback)) {
                            $callback($error);
                        }

                        return;
                    }
                    $flow->send($result);
                    ($this->exec)($flow, $callback, $depth);
                };
                call_user_func_array($func, $args);
            } elseif ($value instanceof Generator) {
                $this->logGenerator($value, $depth);
                ($this->exec)($value, function ($error, $result) use ($flow, $callback, $depth) {

This code seems to be duplicated across your project.

                    if ($error) {
                        if ($this->logger) {
                            $this->logger->error((string)$error);
                        }
                        if (is_callable($callback)) {
                            $callback($error);
                        }

                        return;
                    }
                    $flow->send($result);
                    ($this->exec)($flow, $callback, $depth);
                }, $depth + 1);
            } elseif (is_a($value, static::PROMISE_REACT)) {
                $this->handlePromise($flow, $callback, $depth, $value);

It seems like $callback defined by parameter $callback on line 91 can also be of type null; however, LogicalSteps\Async\Async::handlePromise() does only seem to accept callable, maybe add an additional type check?

            } elseif (is_a($value, static::PROMISE_GUZZLE) || is_a($value, static::PROMISE_HTTP)) {
                $this->handlePromise($flow, $callback, $depth, $value);

It seems like $callback defined by parameter $callback on line 91 can also be of type null; however, LogicalSteps\Async\Async::handlePromise() does only seem to accept callable, maybe add an additional type check?

                $value->wait(false);
            } elseif (is_a($value, static::PROMISE_AMP)) {
                if ($this->logger) {
                    $this->logger->info('await $promise;');
                }
                $value->onResolve(
                    function ($error, $result) use ($flow, $callback, $depth) {

This code seems to be duplicated across your project.

                        if ($error) {
                            if ($this->logger) {
                                $this->logger->error((string)$error);
                            }
                            if (is_callable($callback)) {
                                $callback($error);
                            }

                            return;
                        }
                        $flow->send($result);
                        ($this->exec)($flow, $callback, $depth);
                    }
                );
            } else {
                $flow->send($value);
                ($this->exec)($flow, $callback, $depth);
            }

        } catch
        (Throwable $t) {
            $flow->throw($t);
            ($this->exec)($flow);
        }
    }

    /**
     * Handle known promise interfaces
     *
     * @param Generator $flow
     * @param callable $callback
     * @param int $depth
     * @param \React\Promise\PromiseInterface|\GuzzleHttp\Promise\PromiseInterface $value
     */
    private function handlePromise(Generator $flow, callable $callback, int $depth, $value)
    {
        if ($this->logger) {
            $this->logger->info('await $promise;');
        }
        $value->then(
            function ($result) use ($flow, $callback, $depth) {
                $flow->send($result);
                ($this->exec)($flow, $callback, $depth);
            },
            function ($error) use ($callback, $depth) {
                if ($this->logger) {
                    $this->logger->error((string)$error, compact('depth'));
                }
                if (is_callable($callback)) {
                    $callback($error);
                }
            }
        );
    }

    private function logCallable(callable $callable, array $arguments, int $depth = 0)
    {
        if (!$this->logger) {
            return;
        }
        if (is_array($callable)) {
            $name = $callable[0];
            if (is_object($name)) {
                $name = '$' . lcfirst(get_class($name)) . '->' . $callable[1];
            } else {
                $name .= '::' . $callable[1];
            }

        } else {
            if (is_string($callable)) {
                $name = $callable;
            } elseif ($callable instanceof Closure) {
                $name = 'closure';
            } else {
                $name = 'callable';
            }
        }
        $this->logger->info('await ' . $name . $this->format($arguments), compact('depth'));

    }

    private function logGenerator(Generator $generator, int $depth = 0)
    {
        if (!$generator->valid() || !$this->logger) {
            return;
        }
        $info = new ReflectionGenerator($generator);
        $f = $info->getFunction();
        if ($name = $info->getThis()) {
            if (is_object($name)) {
                $name = '$' . lcfirst(get_class($name)) . '->' . $f->name;
            } else {
                $name .= '::' . $f->name;
            }
        } else {
            $name = $f->name;
        }
        $args = [];
        foreach ($f->getParameters() as $parameter) {
            $args[] = '$' . $parameter->name;
        }
        $this->logger->info('await ' . $name . '(' . implode(', ', $args) . ');', compact('depth'));

    }

    private function format($parameters)
    {
        return '(' . substr(json_encode($parameters), 1, -1) . ');';
    }

    /**
     * A method used to test whether this class is autoloaded.
     *
     * @return bool
     *
     * @see \LogicalSteps\Async\Test\DummyTest
     */
    public function autoloaded()
    {
        return true;
    }

    /**
     * Sets a logger instance on the object.
     *
     * @param LoggerInterface $logger
     *
     * @return void
     */
    public function setLogger(LoggerInterface $logger)
    {
        $this->logger = $logger;
    }

    /**
     * @param LoopInterface $loop
     */
    public function setLoop(LoopInterface $loop)
    {
        $this->loop = $loop;
        $this->exec = [$this, '_executeReactLoop'];
    }

    public function useAmpLoop()
    {
        $this->exec = [$this, '_executeAmpLoop'];
    }
}