MGDigital\BusQue\QueueWorker

Complexity

Total Complexity

10

Size/Duplication

Total Lines	45
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	6

Test Coverage

Coverage

89.29%

Importance

Changes	5
Bugs	0	Features	1

3 Methods

Rating	Name	Duplication	Size	Complexity
B	work()	0	11	6
A	__construct()	0	4	1
A	iterate()	0	22	3

<?php

namespace MGDigital\BusQue;

use MGDigital\BusQue\Exception\TimeoutException;
use Psr\Log\LogLevel;

class QueueWorker
{

    private $implementation;

    public function __construct(Implementation $implementation)
    {
        $this->implementation = $implementation;
    }

    public function work(string $queueName, int $n = null, int $time = null)
    {
        $stopwatchStart = time();
        while ($n === null || $n > 0) {
            $this->iterate($queueName, $time);
            $n === null || $n--;
            if ($time !== null && (time() - $stopwatchStart >= $time)) {
                break;
            }
        }
    }

    private function iterate(string $queueName, int $time = null)
    {
        try {
            $received = $this->implementation->getQueueDriver()
                ->awaitCommand($queueName, $time);
        } catch (TimeoutException $e) {
            return;
        }
        $command = $this->implementation->getCommandSerializer()
            ->unserialize($received->getSerialized());
        $this->implementation->getLogger()->debug('Command received', compact('command'));
        try {
            $this->implementation->getCommandBusAdapter()
                ->handle($command, true);
            $this->implementation->getLogger()->info('Command handled', compact('command'));
        } catch (\Throwable $exception) {
            $this->implementation->getLogger()->error('Command failed', compact('command', 'exception'));
        } finally {
            $this->implementation->getQueueDriver()
                ->completeCommand($received->getQueueName(), $received->getId());
        }
    }
}