Ackintosh\Snidel\TaskQueue

Complexity

Total Complexity

13

Size/Duplication

Total Lines	83
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	2

Importance

Changes	1
Bugs	0	Features	0

7 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	enqueue()	0	19	3
A	dequeue()	0	18	3
A	queuedCount()	0	4	1
A	dequeuedCount()	0	4	1
A	isClosure()	0	4	2
A	__destruct()	0	7	2

<?php
namespace Ackintosh\Snidel;

use Ackintosh\Snidel\IpcKey;
use Opis\Closure\SerializableClosure;

class TaskQueue
{
    const TASK_MAX_SIZE = 1024;

    private $queuedCount = 0;
    private $dequeuedCount = 0;

    public function __construct($ownerPid)
    {
        $this->ownerPid = $ownerPid;

The property ownerPid does not exist. Did you maybe forget to declare it?

        $this->ipcKey = new IpcKey($ownerPid, 'snidel_task_' . uniqid((string) mt_rand(1, 100), true));

The property ipcKey does not exist. Did you maybe forget to declare it?

        $this->id = msg_get_queue($this->ipcKey->generate());

The property id does not exist. Did you maybe forget to declare it?

    }

    /**
     * @param   callable            $callable
     * @param   array               $args
     * @param   string              $tag
     * @return  void
     * @throws  RuntimeException
     */
    public function enqueue($callable, $args = array(), $tag = null)
    {
        $this->queuedCount++;
        if ($this->isClosure($callable)) {
            $serializedCallable = new SerializableClosure($callable);

$callable is of type callable, but the function expects a object<Closure>.

        } else {
            $serializedCallable = serialize($callable);
        }

        $data = array(
            'callable'  => $serializedCallable,
            'args'      => $args,
            'tag'       => $tag,
        );

        if (!msg_send($this->id, 1, serialize($data))) {
            throw new RuntimeException('failed to enqueue task.');
        }
    }

    public function dequeue()
    {
        $this->dequeuedCount++;
        $msgtype = $message = null;
        $success = msg_receive($this->id, 1, $msgtype, self::TASK_MAX_SIZE, $message);

        if (!$success) {
            throw new \RuntimeException('failed to dequeue task');
        }

        $data = unserialize($message);
        if ($this->isClosure($data['callable'])) {
            $data['callable'] = $data['callable']->getClosure();
        } else {
            $data['callable'] = unserialize($data['callable']);
        }
        return $data;
    }

    public function queuedCount()
    {
        return $this->queuedCount;
    }

    public function dequeuedCount()
    {
        return $this->dequeuedCount;
    }

    private function isClosure($callable)
    {
        return is_object($callable) && is_callable($callable);
    }

    public function __destruct()
    {
        if ($this->ipcKey->isOwner(getmypid())) {
            $this->ipcKey->delete();
            return msg_remove_queue($this->id);
        }
    }// @codeCoverageIgnore
}