migrate the retry jobs of a queue
@param string $queue
@return array
21
protected function migrateRetryJobs($queue)
    {
        $luaScript = <<<LUA
-- Get all of the jobs with an expired "score"...
local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])

-- If we have values in the array, we will remove them from the first queue
-- and add them onto the destination queue in chunks of 100, which moves
-- all of the appropriate jobs onto the destination queue very safely.
if(next(val) ~= nil) then
    redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)

    for i = 1, #val, 100 do
        redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))
    end
end

return val
LUA;

        return $this->client->eval(
            $luaScript,
            2,
            $this->getRetryZsetNameWithPrefix($queue),
            $this->getQueueNameWithPrefix($queue),
            time()
        );
    }
291239
public function pop($queue = null)
    {
        $queue = $this->getQueue($queue);

        $response = $this->sqs->receiveMessage(
            array('QueueUrl' => $queue, 'AttributeNames' => array('ApproximateReceiveCount'))
        );

        if (count($response['Messages']) > 0)
        {
            return new SqsJob($this->container, $this->sqs, $queue, $response['Messages'][0]);
        }
    }
462287
public function job()
    {
        $job = $this->resqueInstance->redis->get(self::WORKER_NAME . ':' . $this);
        if (!$job) {
            return [];
        } else {
            return json_decode($job, true);
        }
    }
1239456
protected function getNextMessage($queue)
    {
        $queue = (array) $queue;

        foreach ($queue as $queueName) {
            if ($message = $this->manager->pop($queueName)) {
                return $message;
            }
        }

        return null;
    }
1329045
private static function removeList($queue)
    {
        $counter = self::size($queue);
        $result = self::redis()->del('queue:' . $queue);

        return ($result == 1) ? $counter : 0;
    }
1136962
protected function retrieveNextJob($queue, $block = true)
    {
        $nextJob = $this->getConnection()->eval(
            LuaScripts::pop(), 3, $queue, $queue.':reserved', $queue.':notify',
            $this->availableAt($this->retryAfter)
        );

        if (empty($nextJob)) {
            return [null, null];
        }

        [$job, $reserved] = $nextJob;

        if (! $job && ! is_null($this->blockFor) && $block &&
            $this->getConnection()->blpop([$queue.':notify'], $this->blockFor)) {
            return $this->retrieveNextJob($queue, false);
        }

        return [$job, $reserved];
    }
617041
protected function getNextJob($connection, $queue)
    {
        try {
            if (is_null($queue)) {
                return $connection->pop();
            }

            foreach (explode(',', $queue) as $queue) {
                if (! is_null($job = $connection->pop($queue))) {
                    return $job;
                }
            }
        }
        catch (Exception $e) {
            $this->handleException($e);
        }
        catch (Throwable $e) {
            $this->handleException(new FatalThrowableError($e));
        }
    }
343887
public function reset($queue)
    {
        if (!function_exists('posix_getsid')) {
            throw JobConfigurationException::create('Function posix_getsid don\'t exists');
        }

        $configurations = $this->configurationManager->findPotentialDeadJobs($queue);

        foreach ($configurations as $configuration) {
            if ($this->restoreManager->reset($configuration)) {
                return true;
            }
        }

        return false;
    }
1043241
public static function dequeue($queue, $items = [])
    {
        if (count($items) > 0) {
            return self::removeItems($queue, $items);
        }

        return self::removeList($queue);
    }
1243116
public function reloadErrorJobs()
    {
        $redis = $this->getRedisChildClient();
        $count = 0;
        while ($data = $redis->rpoplpush($this->errorKey, $this->queueKey)) {
            $count++;
        }

        return $count;
    }
469844
public function retry(RetrySchedulerInterface $retryScheduler): array
    {
        if (!$this->isRetryable()) {
            throw new class('This job cannot be retried anymore.') extends \BadMethodCallException implements ExceptionInterface {};
        }

        $cloned = clone $this;
        $cloned->retries++;

        return [$cloned, $retryScheduler->scheduleNextRetry($cloned, $cloned->retries)];
    }
1199148
public function pop($queue = null)
    {
        $queue = $queue ?: $this->queue;

        $now = new \DateTimeImmutable('now');

        $query = $this->db->getQuery(true);

        $query->select('*')
            ->from($query->quoteName($this->table))
            ->where('queue = %q', $queue)
            ->where('visibility <= %q', $now->format('Y-m-d H:i:s'))
            ->orWhere(
                function (Query $query) use ($now) {
                    $query->where('reserved IS NULL')
                        ->where('reserved < %q', $now->modify('-' . $this->timeout . 'seconds')->format('Y-m-d H:i:s'));
                }
            );

        $trans = $this->db->getTransaction()->start();

        try {
            $data = $this->db->setQuery($query . ' FOR UPDATE')->loadOne('assoc');

            if (!$data) {
                return null;
            }

            $data['attempts']++;

            $values = ['reserved' => $now->format('Y-m-d H:i:s'), 'attempts' => $data['attempts']];

            $this->db->getWriter()->updateBatch($this->table, $values, ['id' => $data['id']]);

            $trans->commit();
        } catch (\Throwable $t) {
            $trans->rollback();
        }

        $message = new QueueMessage();

        $message->setId($data['id']);
        $message->setAttempts($data['attempts']);
        $message->setBody(json_decode($data['body'], true));
        $message->setRawBody($data['body']);
        $message->setQueueName($queue);

        return $message;
    }
52100
protected function blockingPop($queue)
    {
        $rawBody = $this->getConnection()->blpop($queue, $this->blockFor);

        if (! empty($rawBody)) {
            $payload = json_decode($rawBody[1], true);

            $payload['attempts']++;

            $reserved = json_encode($payload);

            $this->getConnection()->zadd(
                $queue.':reserved', [
                $reserved => $this->availableAt($this->retryAfter),
                ]
            );

            return [$rawBody[1], $reserved];
        }

        return [null, null];
    }
691459
protected function getFailedJobs()
	{
		$results = [];

		foreach ($this->laravel['queue.failer']->all() as $failed) {
			$results[] = $this->parseFailedJob((array) $failed);
		}

		return array_filter($results);
	}
138099
public function pop($queue = null)
    {
        $queue = $this->getQueue($queue);

        return $this->database->transaction(function () use ($queue)
        {
            if (! is_null($job = $this->getNextAvailableJob($queue))) {
                $this->markJobAsReserved($job->id, $job->attempts);

                return new DatabaseJob(
                    $this->container, $this, $job, $queue
                );
            }
        });
    }
587919
public function get($queueName)
    {
        if (! is_null($this->expire) )
        {
            $this->migrateAllExpiredJobs($queueName);
        }

        $objectReids = $this->__handler;
        $objectReids->loadScripts('queueGet');

        $queueData = $objectReids->queueGet($queueName, $queueName.':reserved', time() + $this->expire);

        if( ! is_null($queueData) )
        {
            return new system_queue_message_redis($queueData);
        }

        return false;
    }
302931
public function sizeDelayed($queue)
    {
        return $this->redis->zcard(self::redisKey($this->getQueue($queue), 'delayed'));
    }