SidekiqJob\Client

Complexity

Total Complexity

21

Size/Duplication

Total Lines	156
Duplicated Lines	8.97 %

Coupling/Cohesion

Components	1
Dependencies	3

Test Coverage

Coverage

0%

Importance

Changes	1
Bugs	0	Features	0

7 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	7	4
A	push()	7	7	1
A	schedule()	7	7	1
B	pushBulk()	0	21	7
B	atomicPush()	0	19	6
A	name()	0	4	1
A	getRedis()	0	4	1

<?php

namespace SidekiqJob;

/**
 * Class Client
 * @package SidekiqJob
 */
class Client
{
    /** Default queue name */
    const QUEUE = 'default';

    /** Namespace  */
    protected $namespace;

    /** @var \Predis\Client */
    public $redis;

    /** @var Serializer */
    protected $serializer;

    /** @var IdGenerator */
    protected $idGenerator;

    /**
     * Sidekiq job pusher init
     *
     * @param \Predis\Client $redis
     * @param string         $namespace
     * @param Serializer     $serializer
     * @param IdGenerator    $idGenerator
     */
    public function __construct(\Predis\Client $redis, $namespace = null, $serializer = null, $idGenerator = null)
    {
        $this->redis = $redis;
        $this->namespace = ($namespace === null) ? '' : (string) $namespace;
        $this->serializer = ($serializer === null) ? new Serializer() : $serializer;
        $this->idGenerator = ($idGenerator === null) ? new IdGenerator() : $idGenerator;
    }

    /**
     * Push a job
     *
     * @param string $class
     * @param array  $args
     * @param bool   $retry
     * @param string $queue
     * @return string
     */
    public function push($class, $args = [], $retry = true, $queue = self::QUEUE)
    {
        $jobId = $this->idGenerator->generate();
        $this->atomicPush($jobId, $class, $args, $queue, $retry);

        return $jobId;
    }

    /**
     * Schedule a job at a certain time
     *
     * @param float  $doAt
     * @param string $class
     * @param array  $args
     * @param bool   $retry
     * @param string $queue
     * @return string
     */
    public function schedule($doAt, $class, $args = [], $retry = true, $queue = self::QUEUE)
    {
        $jobId = $this->idGenerator->generate();
        $this->atomicPush($jobId, $class, $args, $queue, $retry, $doAt);

        return $jobId;
    }

    /**
     * Push multiple jobs to queue
     *
     * Format:
     * $jobs = [
     *  [
     *      'class' => 'SomeClass',
     *      'args' => array(),
     *      'retry' => false,
     *      'at' => microtime(true)
     *  ]
     * ];
     *
     * @param array  $jobs
     * @param string $queue
     * @return string
     * @throws exception Exception
     */
    public function pushBulk($jobs = [], $queue = self::QUEUE)
    {
        $ids = [];
        foreach ($jobs as $job) {
            if (!isset($job['class'])) {
                throw new Exception('pushBulk: each job needs a job class');
            }
            if (!isset($job['args']) || !is_array($job['args'])) {
                throw new Exception('pushBulk: each job needs args');
            }

            $retry = isset($job['retry']) ? $job['retry'] : true;
            $doAt = isset($job['at']) ? $job['at'] : null;

            $jobId = $this->idGenerator->generate();
            array_push($ids, $jobId);
            $this->atomicPush($jobId, $job['class'], $job['args'], $queue, $retry, $doAt);
        }

        return $ids;
    }

    /**
     * Push job to redis
     *
     * @param string     $jobId
     * @param string     $class
     * @param array      $args
     * @param string     $queue
     * @param bool       $retry
     * @param float|null $doAt
     * @throws exception Exception
     */
    private function atomicPush($jobId, $class, $args = [], $queue = self::QUEUE, $retry = true, $doAt = null)
    {
        if (array_values($args) !== $args) {
            throw new Exception('Associative arrays in job args are not allowed');
        }

        if (!is_null($doAt) && !is_float($doAt) && is_string($doAt)) {
            throw new Exception('at argument needs to be in a unix epoch format. Use microtime(true).');
        }

        $job = $this->serializer->serialize($jobId, $class, $args, $retry, $queue);

        if ($doAt === null) {
            $this->redis->sadd($this->name('queues'), $queue);
            $this->redis->lpush($this->name('queue', $queue), $job);
        } else {
            $this->redis->zadd($this->name('schedule'), $doAt, $job);
        }
    }

    /**
     * @param string ...$key
     * @return string
     */
    private function name()
    {
        return implode(':', array_filter(array_merge([$this->namespace], func_get_args()), 'strlen'));
    }

    /**
     * @return \Predis\Client
     */
    public function getRedis()
    {
        return $this->redis;
    }
}