Charcoal\Queue\AbstractQueueManager

Complexity

Total Complexity

39

Size/Duplication

Total Lines	442
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	3

Importance

Changes

0

22 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	17	4
A	setData()	0	8	3
A	setQueueId()	0	5	1
A	queueId()	0	4	1
A	setRate()	0	5	1
A	rate()	0	4	1
A	setLimit()	0	5	1
A	limit()	0	4	1
A	setChunkSize()	0	5	1
A	chunkSize()	0	4	1
A	setItemCallback()	0	5	1
A	setItemSuccessCallback()	0	5	1
A	setItemFailureCallback()	0	5	1
A	setProcessedCallback()	0	5	1
A	processQueue()	0	23	5
B	processItems()	0	26	6
A	loadQueueItems()	0	43	4
A	totalQueuedItems()	0	34	2
A	totalChunks()	0	4	1
	queueItemProto()	0	1	?
A	queueItemFactory()	0	4	1
A	setQueueItemFactory()	0	5	1

<?php

namespace Charcoal\Queue;

use Exception;

// From PSR-3
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;

// From 'charcoal-core'
use Charcoal\Loader\CollectionLoader;

// From 'charcoal-factory'
use Charcoal\Factory\FactoryInterface;

/**
 * Abstract Queue Manager
 *
 * The queue manager is used to load queued items and batch-process them.
 *
 * ## Loading queued items
 *
 * If a "queue_id" is specified, only the item for this specific queue will be loaded.
 * Otherwise, all unprocessed queue items will be processed.
 *
 * ## Type of queue items
 *
 * The type of queue items can be set in extended concrete class with the
 * `queue_item_proto()` method. This method should return a QueueItemInterface instance.
 *
 * ## Callbacks
 *
 * There are 4 available callback methods that can be set:
 *
 * - `item_callback`
 *   - Called after an item has been processed.
 *   - Arguments: `QueueModelInterface $item`
 * - `item_success_callback`
 * - `item_failure_callback`
 * - `processed_callback`
 *   - Called when the entire queue has been processed
 *   - Arguments: `array $success`, `array $failures`
 */
abstract class AbstractQueueManager implements
    QueueManagerInterface,
    LoggerAwareInterface
{
    use LoggerAwareTrait;

    /**
     * The queue processing rate (throttle), in items per second.
     *
     * @var integer
     */
    private $rate = 0;

    /**
     * The batch limit.
     *
     * @var integer
     */
    private $limit = 0;

    /**
     * The chunk size to batch the queue with.
     *
     * @var integer|null
     */
    private $chunkSize = null;

    /**
     * The queue ID.
     *
     * If set, then it will load only the items from this queue.
     * If NULL, load *all* queued items.
     *
     * @var mixed
     */
    private $queueId;

    /**
     * Items that were successfully processed
     *
     * @var array
     */
    private $successItems = [];

    /**
     * Item that failed to process
     *
     * @var array
     */
    private $failedItems = [];

    /**
     * Items that were skipped during the processing
     *
     * @var array
     */
    private $skippedItems = [];

    /**
     * The callback routine when an item is processed (whether resolved or rejected).
     *
     * @var callable $itemCallback
     */
    private $itemCallback;

    /**
     * The callback routine when the item is resolved.
     *
     * @var callable $itemSuccessCallback
     */
    private $itemSuccessCallback;

    /**
     * The callback routine when the item is rejected.
     *
     * @var callable $itemFailureCallback
     */
    private $itemFailureCallback;

    /**
     * The callback routine when the queue is processed.
     *
     * @var callable $processedCallback
     */
    private $processedCallback;

    /**
     * @var FactoryInterface $queueItemFactory
     */
    private $queueItemFactory;

    /**
     * Construct new queue manager.
     *
     * @param array $data Dependencies and settings.
     */
    public function __construct(array $data = [])
    {
        $this->setLogger($data['logger']);
        $this->setQueueItemFactory($data['queue_item_factory']);

        if (isset($data['rate'])) {
            $this->rate = intval($data['rate']);
        }

        if (isset($data['limit'])) {
            $this->limit = intval($data['limit']);
        }

        if (isset($data['chunkSize'])) {
            $this->chunkSize = intval($data['chunkSize']);
        }
    }

    /**
     * Set the manager's data.
     *
     * @param array $data The queue data to set.
     * @return AbstractQueueManager Chainable
     */
    public function setData(array $data)
    {
        if (isset($data['queue_id']) && $data['queue_id']) {
            $this->setQueueId($data['queue_id']);
        }

        return $this;
    }

    /**
     * Set the queue's ID.
     *
     * @param mixed $id The unique queue identifier.
     * @return self
     */
    public function setQueueId($id)
    {
        $this->queueId = $id;
        return $this;
    }

    /**
     * Get the queue's ID.
     *
     * @return mixed
     */
    public function queueId()
    {
        return $this->queueId;
    }

    /**
     * @param integer $rate The throttling / processing rate, in items per second.
     * @return self
     */
    public function setRate($rate)
    {
        $this->rate = intval($rate);
        return $this;
    }

    /**
     * @return integer
     */
    public function rate()
    {
        return $this->rate;
    }

    /**
     * @param integer $limit The maximum number of items to load.
     * @return self
     */
    public function setLimit($limit)
    {
        $this->limit = intval($limit);
        return $this;
    }

    /**
     * @return integer
     */
    public function limit()
    {
        return $this->limit;
    }

    /**
     * @param integer $chunkSize The size of the chunk of items to process at the same time in the queue.
     * @return self
     */
    public function setChunkSize($chunkSize)
    {
        $this->chunkSize = intval($chunkSize);
        return $this;
    }

    /**
     * @return integer
     */
    public function chunkSize()
    {
        return $this->chunkSize;
    }

    /**
     * Set the callback routine when an item is processed.
     *
     * @param callable $callback A item callback routine.
     * @return self
     */
    public function setItemCallback(callable $callback)
    {
        $this->itemCallback = $callback;
        return $this;
    }

    /**
     * Set the callback routine when the item is resolved.
     *
     * @param callable $callback A item callback routine.
     * @return self
     */
    public function setItemSuccessCallback(callable $callback)
    {
        $this->itemSuccessCallback = $callback;
        return $this;
    }

    /**
     * Set the callback routine when the item is rejected.
     *
     * @param callable $callback A item callback routine.
     * @return self
     */
    public function setItemFailureCallback(callable $callback)
    {
        $this->itemSuccessCallback = $callback;
        return $this;
    }

    /**
     * Set the callback routine when the queue is processed.
     *
     * @param callable $callback A queue callback routine.
     * @return self
     */
    public function setProcessedCallback(callable $callback)
    {
        $this->processedCallback = $callback;
        return $this;
    }

    /**
     * Process the queue.
     *
     * It can be process in a single batch or in multiple chunks to reduce memory
     * If no callback is passed and a self::$processedCallback is set, the latter is used.
     *
     * @param  callable $callback An optional alternative callback routine executed
     *                            after all queue items are processed.
     * @return boolean  Success / Failure
     */
    public function processQueue(callable $callback = null)
    {
        if (!is_callable($callback)) {
            $callback = $this->processedCallback;
        }

        if (!is_null($this->chunkSize())) {
            $totalChunks = $this->totalChunks();
            for ($i = 0; $i <= $totalChunks; $i++) {
                $queuedItems = $this->loadQueueItems();
                $this->processItems($queuedItems);
            }
        } else {
            $queuedItems = $this->loadQueueItems();
            $this->processItems($queuedItems);
        }

        if (is_callable($callback)) {
            $callback($this->successItems, $this->failedItems, $this->skippedItems);
        }

        return true;
    }

    /**
     * @param mixed $queuedItems The items to process.
     * @return void
     */
    private function processItems($queuedItems)
    {
        foreach ($queuedItems as $q) {
            try {
                $res = $q->process($this->itemCallback, $this->itemSuccessCallback, $this->itemFailureCallback);
                if ($res === true) {
                    $this->successItems[] = $q;
                } elseif ($res === false) {
                    $this->failedItems[] = $q;
                } else {
                    $this->skippedItems[] = $q;
                }
            } catch (Exception $e) {
                $this->logger->error(
                    sprintf('Could not process a queue item: %s', $e->getMessage())
                );
                $this->failedItems[] = $q;
                continue;
            }

            // Throttle according to processing rate.
            if ($this->rate > 0) {
                usleep(1000000 / $this->rate);
            }
        }
    }

    /**
     * Retrieve the items of the current queue.
     *
     * @return \Charcoal\Model\Collection|array
     */
    public function loadQueueItems()
    {
        $loader = new CollectionLoader([
            'logger'  => $this->logger,
            'factory' => $this->queueItemFactory(),
        ]);
        $loader->setModel($this->queueItemProto());
        $loader->addFilter([
            'property' => 'processed',
            'value'    => 0,
        ]);
        $loader->addFilter([
            'property' => 'processing_date',
            'operator' => '<',
            'value'    => date('Y-m-d H:i:s'),
        ]);

        $queueId = $this->queueId();
        if ($queueId) {
            $loader->addFilter([
                'property' => 'queue_id',
                'value'    => $queueId,
            ]);
        }

        $loader->addOrder([
            'property' => 'queued_date',
            'mode'     => 'asc',
        ]);

        if (!is_null($this->chunkSize())) {
            $loader->setNumPerPage($this->chunkSize());
        }

        if ($this->limit() > 0) {
            $loader->setNumPerPage($this->limit());
            $loader->setPage(0);
        }

        $queued = $loader->load();

        return $queued;
    }

    /**
     * Retrieve the total of queued items.
     *
     * @return integer
     */
    public function totalQueuedItems()
    {
        $loader = new CollectionLoader([
            'logger'  => $this->logger,
            'factory' => $this->queueItemFactory(),
        ]);
        $loader->setModel($this->queueItemProto());
        $loader->addFilter([
            'property' => 'processed',
            'value'    => 0,
        ]);
        $loader->addFilter([
            'property' => 'processing_date',
            'operator' => '<',
            'value'    => date('Y-m-d H:i:s'),
        ]);

        $queueId = $this->queueId();
        if ($queueId) {
            $loader->addFilter([
                'property' => 'queue_id',
                'value'    => $queueId,
            ]);
        }

        $loader->addOrder([
            'property' => 'queued_date',
            'mode'     => 'asc',
        ]);

        $total = $loader->loadCount();

        return $total;
    }

    /**
     * Retrieve the number of chunks to process.
     *
     * @return integer
     */
    public function totalChunks()
    {
        return (int)ceil($this->totalQueuedItems() / $this->chunkSize());
    }

    /**
     * Retrieve the queue item's model.
     *
     * @return QueueItemInterface
     */
    abstract public function queueItemProto();

    /**
     * @return FactoryInterface
     */
    protected function queueItemFactory()
    {
        return $this->queueItemFactory;
    }

    /**
     * @param FactoryInterface $factory The factory used to create queue items.
     * @return self
     */
    private function setQueueItemFactory(FactoryInterface $factory)
    {
        $this->queueItemFactory = $factory;
        return $this;
    }
}