Tarantool\JobQueue\Runner\Amp\GenericTask

Complexity

Total Complexity

8

Size/Duplication

Total Lines	51
Duplicated Lines	0 %

Test Coverage

Coverage

0%

Importance

Changes	2
Bugs	0	Features	0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	getExecutor()	0	10	2
A	__construct()	0	5	1
A	run()	0	6	1
A	createExecutor()	0	16	4

<?php

namespace Tarantool\JobQueue\Runner\Amp;

use Amp\Parallel\Worker\Environment;
use Amp\Parallel\Worker\Task;
use Tarantool\JobQueue\Executor\Executor;
use Tarantool\JobQueue\Executor\ExecutorChain;
use Tarantool\JobQueue\Executor\ProcessExecutor;
use Tarantool\JobQueue\JobBuilder\JobOptions;
use Tarantool\Queue\Queue;
use Tarantool\Queue\Task as QueueTask;

class GenericTask implements Task
{
    private const EXECUTOR_KEY = self::class.'-executor';

    private $task;
    private $queue;
    private $executorsConfigFile;

    public function __construct(QueueTask $task, Queue $queue, string $executorsConfigFile = null)
    {
        $this->task = $task;
        $this->queue = $queue;
        $this->executorsConfigFile = $executorsConfigFile;
    }

    public function run(Environment $environment): void
    {
        $data = $this->task->getData();
        $payload = $data[JobOptions::PAYLOAD] ?? null;
        $executor = $this->getExecutor($environment);
        $executor->execute($payload, $this->queue);
    }

    private function getExecutor(Environment $environment): Executor
    {
        $executor = $environment->get(self::EXECUTOR_KEY);

        if (!$executor) {
            $executor = $this->createExecutor();
            $environment->set(self::EXECUTOR_KEY, $executor);
        }

        return $executor;
    }

    private function createExecutor(): Executor
    {
        if (!$this->executorsConfigFile) {
            return new ProcessExecutor();
        }

        $executors = require $this->executorsConfigFile;
        if ($executors instanceof Executor) {
            return $executors;
        }

        if (!is_array($executors)) {
            throw new \RuntimeException(sprintf('%s must return an Executor object or an array of Executor objects.', $this->executorsConfigFile));
        }

        return new ExecutorChain($executors);
    }
}