Phive\Queue\RedisQueue

Complexity

Total Complexity

10

Size/Duplication

Total Lines	90
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	3

Test Coverage

Coverage

100%

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	4	1
A	push()	0	11	2
A	pop()	0	15	3
A	count()	0	7	1
A	clear()	0	5	1
A	assertResult()	0	6	2

<?php

/*
 * This file is part of the Phive Queue package.
 *
 * (c) Eugene Leonovich <gen.work@gmail.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Phive\Queue;

/**
 * RedisQueue requires redis server 2.6 or higher.
 */
class RedisQueue implements Queue
{
    const SCRIPT_PUSH = <<<'LUA'
        local id = redis.call('INCR', KEYS[2])
        return redis.call('ZADD', KEYS[1], ARGV[2], id..':'..ARGV[1])
LUA;

    const SCRIPT_POP = <<<'LUA'
        local items = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, 1)
        if 0 == #items then return -1 end
        redis.call('ZREM', KEYS[1], items[1])
        return string.sub(items[1], string.find(items[1], ':') + 1)
LUA;

    /**
     * @var \Redis
     */
    private $redis;

    public function __construct(\Redis $redis)
    {
        $this->redis = $redis;
    }

    /**
     * {@inheritdoc}
     */
    public function push($item, $eta = null)
    {
        $eta = QueueUtils::normalizeEta($eta);

        if (\Redis::SERIALIZER_NONE !== $this->redis->getOption(\Redis::OPT_SERIALIZER)) {
            $item = $this->redis->_serialize($item);

The method _serialize() does not exist on Redis. Did you maybe mean _unserialize()?

        }

        $result = $this->redis->evaluate(self::SCRIPT_PUSH, ['items', 'seq', $item, $eta], 2);
        $this->assertResult($result);
    }

    /**
     * {@inheritdoc}
     */
    public function pop()
    {
        $result = $this->redis->evaluate(self::SCRIPT_POP, ['items', time()], 1);
        $this->assertResult($result);

        if (-1 === $result) {
            throw new NoItemAvailableException($this);
        }

        if (\Redis::SERIALIZER_NONE !== $this->redis->getOption(\Redis::OPT_SERIALIZER)) {
            return $this->redis->_unserialize($result);
        }

        return $result;
    }

    /**
     * {@inheritdoc}
     */
    public function count()
    {
        $result = $this->redis->zCard('items');
        $this->assertResult($result);

        return $result;
    }

    /**
     * {@inheritdoc}
     */
    public function clear()
    {
        $result = $this->redis->del(['items', 'seq']);
        $this->assertResult($result);
    }

    /**
     * @param mixed $result
     *
     * @throws QueueException
     */
    protected function assertResult($result)
    {
        if (false === $result) {
            throw new QueueException($this, $this->redis->getLastError());
        }
    }
}