VXM\Async\Async

Complexity

Total Complexity

13

Size/Duplication

Total Lines	159
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Importance

Changes

0

9 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	run()	0	11	1
A	batchRun()	0	14	3
A	wait()	0	7	1
A	makeJob()	0	8	2
A	createClassJob()	0	8	1
A	addProcessListeners()	0	6	2
A	makeProcessListener()	0	8	1
A	createProcess()	0	4	1

<?php
/**
 * @link https://github.com/vuongxuongminh/laravel-async
 *
 * @copyright (c) Vuong Xuong Minh
 * @license [MIT](https://opensource.org/licenses/MIT)
 */

namespace VXM\Async;

use Closure;
use Illuminate\Contracts\Events\Dispatcher as EventDispatcher;
use Illuminate\Support\Str;
use Spatie\Async\Process\Runnable;
use VXM\Async\Runtime\ParentRuntime;

/**
 * @author Vuong Minh <vuongxuongminh@gmail.com>
 * @since  1.0.0
 */
class Async
{
    /**
     * A pool manage async processes.
     *
     * @var Pool
     */
    protected $pool;

    /**
     * Event dispatcher manage async events.
     *
     * @var EventDispatcher
     */
    protected $events;

    /**
     * Create a new Async instance.
     *
     * @param \VXM\Async\Pool $pool
     * @param \Illuminate\Contracts\Events\Dispatcher $events
     */
    public function __construct(Pool $pool, EventDispatcher $events)
    {
        $this->pool = $pool;
        $this->events = $events;
    }

    /**
     * Execute async job.
     *
     * @param callable|string|object $job need to execute.
     * @param array $events event. Have key is an event name, value is a callable triggered when event
     *                                       happen, have three events `error`, `success`, `timeout`.
     *
     * @return static
     */
    public function run($job, array $events = []): self
    {
        $process = $this->createProcess($this->makeJob($job));

$this->makeJob($job) is of type *, but the function expects a callable.

        $process->then($this->makeProcessListener('success', $process));
        $process->catch($this->makeProcessListener('error', $process));
        $process->timeout($this->makeProcessListener('timeout', $process));
        $this->addProcessListeners($events, $process);
        $this->pool->add($process);

$process is of type object<Spatie\Async\Process\Runnable>, but the function expects a callable.

        return $this;
    }

    /**
     * Batch execute async jobs.
     *
     * @param array $jobs
     * @return static
     * @see run()
     * @since 2.0.0
     */
    public function batchRun(...$jobs): self
    {
        foreach ($jobs as $job) {
            if (is_array($job)) {
                [$job, $events] = $job;

The variable $events does not seem to be defined for all execution paths leading up to this point.

            } else {
                $events = [];
            }

            $this->run($job, $events);
        }

        return $this;
    }

    /**
     * Wait until all async jobs done and return job results.
     *
     * @return array
     */
    public function wait()
    {
        $results = $this->pool->wait();
        $this->pool->flush();

        return $results;
    }

    /**
     * Make async job.
     *
     * @param $job
     *
     * @return mixed
     */
    protected function makeJob($job)
    {
        if (is_string($job)) {
            return $this->createClassJob($job);
        }

        return $job;
    }

    /**
     * Create class and method job.
     *
     * @param string $job
     *
     * @return Closure
     */
    protected function createClassJob(string $job): Closure
    {
        [$class, $method] = Str::parseCallback($job, 'handle');

The variable $class does not exist. Did you forget to declare it?

The variable $method does not exist. Did you forget to declare it?

        return function () use ($class, $method) {
            return app()->call($class.'@'.$method);
        };
    }

    /**
     * Listen events of process given.
     *
     * @param array $events
     * @param Runnable $process
     */
    protected function addProcessListeners(array $events, Runnable $process): void
    {
        foreach ($events as $event => $callable) {
            $this->events->listen("async.{$event}_{$process->getId()}", $callable);
        }
    }

    /**
     * Make a base listener for integration with [[EventDispatcher]].
     *
     * @param string $event
     * @param Runnable $process
     *
     * @return callable
     */
    protected function makeProcessListener(string $event, Runnable $process): callable
    {
        return function (...$args) use ($event, $process) {
            $event = "async.{$event}_{$process->getId()}";

Consider using a different name than the imported variable $event, or did you forget to import by reference?

            $this->events->dispatch($event, $args);
            $this->events->forget($event);
        };
    }

    /**
     * Create a new process for run a job.
     *
     * @param callable $job need to execute.
     *
     * @return Runnable process.
     */
    protected function createProcess($job): Runnable
    {
        return ParentRuntime::createProcess($job);
    }
}