Equip\Queue\Worker

Complexity

Total Complexity

12

Size/Duplication

Total Lines	135
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Test Coverage

Coverage

100%

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	11	1
A	consume()	0	8	2
B	tick()	0	27	6
A	bindSignals()	0	14	1
A	shutdown()	0	4	1
A	drain()	0	4	1

<?php

namespace Equip\Queue;

use Equip\Queue\Driver\DriverInterface;
use Exception;
use League\Tactician\CommandBus;

class Worker
{
    /**
     * @var DriverInterface
     */
    private $driver;

    /**
     * @var Queue
     */
    private $queue;

    /**
     * @var Event
     */
    private $event;

    /**
     * @var CommandBus
     */
    private $command_bus;

    /**
     * Will shutdown the worker on the next tick
     *
     * @var bool
     */
    private $shutdown = false;

    /**
     * Will shutdown the worker when the queue is drained
     *
     * @var bool
     */
    private $drain = false;

    public function __construct(
        DriverInterface $driver,
        Queue $queue,
        Event $event,
        CommandBus $command_bus
    ) {
        $this->driver = $driver;
        $this->queue = $queue;
        $this->event = $event;
        $this->command_bus = $command_bus;
    }

    /**
     * Consumes messages off of the queue
     *
     * @codeCoverageIgnore
     *
     * @param string $queue
     */
    public function consume($queue)
    {
        $this->bindSignals();

        while ($this->tick($queue)) {
            pcntl_signal_dispatch();
        }
    }

    /**
     * Handles fetching messages from the queue
     *
     * @param string $queue
     *
     * @return bool
     */
    protected function tick($queue)
    {
        if ($this->shutdown) {
            $this->event->shutdown();
            return false;
        }

        $message = $this->driver->dequeue($queue);
        if (empty($message) && $this->drain) {
            $this->event->drained();
            return false;
        } elseif (empty($message)) {
            return true;
        }

        $command = unserialize($message);
        try {
            $this->event->acknowledge($command);
            $this->command_bus->handle($command);
            $this->event->finish($command);
        } catch (Exception $exception) {
            $this->queue->add(sprintf('%s-failed', $queue), $command);
            $this->event->reject($command, $exception);
        }

        return true;
    }

    /**
     * Handles binding POSIX signals appropriately
     *
     * @codeCoverageIgnore
     */
    private function bindSignals()
    {
        // Shutdown the listener
        array_map(function ($signal) {
            pcntl_signal($signal, [$this, 'shutdown']);
        }, [
            SIGTERM,
            SIGINT,
            SIGQUIT,
        ]);

        // Drain the queue
        pcntl_signal(SIGHUP, [$this, 'drain']);
    }

    /**
     * Set the worker to shutdown on the next tick
     */
    private function shutdown()
    {
        $this->shutdown = true;
    }

    /**
     * Set the worker to shutdown when the queue is drained
     */
    private function drain()
    {
        $this->drain = true;
    }
}