将一个job的状态迁移为buried(通过kick命令唤醒)

@see kickJob

@param Job $job      需要迁移的Job.
@param int $priority 优先级.

@return true
507286
public function bury(Job $job, $priority = PheanstalkInterface::DEFAULT_PRIORITY)
    {
        $this->client->bury($job, $priority);

        return true;
    }
617880
public function enqueue(Job $job): void
    {
        $queue = $job->getQueue();

        $this->client->sadd('queue', [$queue]);

        $this->client->rpush(
            sprintf('queue:%s', $queue),
            [
                json_encode($job->getPayload()),
            ]
        );
    }
76853
private function pushExecutedJob(Job $job)
    {
        $this->executed_jobs[] = $job;
        $command = $job->getCommand();
        $args = $job->getArguments();
        // If callable, log the string Closure
        if (is_callable($command)) {
            $command = \is_string($command) ? $command : 'Closure';
        }
        $this->addSchedulerVerboseOutput("<green>Success</green>: <white>{$command} {$args}</white>");

        return $job;
    }
1301191
public function workingOn(Job &$job)
    {
        $this->job = $job;
        $job->setWorker($this);

        Event::fire(Event::WORKER_WORKING_ON, array($this, $job));

        $this->redis->hmset(self::redisKey($this), array(
            'job_id'      => $job->getId(),
            'job_started' => microtime(true)
        ));
    }
61379
protected function updateJob(array $job, int $status): bool
    {
        $set = [
            'status' => $status,
        ];

        if ($status >= JobInterface::STATUS_DONE) {
            $set['ended'] = new UTCDateTime();
        }

        $result = $this->db->{$this->scheduler->getJobQueue()}->updateMany([
            '_id' => $job['_id'],
            '$isolated' => true,
        ], [
            '$set' => $set,
        ]);

        return $result->isAcknowledged();
    }
526454
public function perform(Resque_Job $job)
	{
		try {
			Resque_Event::trigger('afterFork', $job);
			$job->perform();
		}
		catch(Exception $e) {
			$this->logger->log(Psr\Log\LogLevel::CRITICAL, '{job} has failed {stack}', array('job' => $job, 'stack' => $e));
			$job->fail($e);
			return;
		}

		$job->updateStatus(Resque_Job_Status::STATUS_COMPLETE);
		$this->logger->log(Psr\Log\LogLevel::NOTICE, '{job} has finished', array('job' => $job));
	}
283820
public function push( $job, $data = '', $queue = null )
	{
		$queueJob = $this->resolveJob( $this->createPayload( $job, $data, $queue ) );

		try
		{
			$this->raiseBeforeJobEvent( $queueJob );

			$queueJob->fire();

			$this->raiseAfterJobEvent( $queueJob );
		}
		catch ( Exception $e )
		{
			$this->raiseExceptionOccurredJobEvent( $queueJob, $e );

			$this->handleFailedJob( $queueJob );

			throw $e;
		}
		catch ( Throwable $e )
		{
			$this->raiseExceptionOccurredJobEvent( $queueJob, $e );

			$this->handleFailedJob( $queueJob );

			throw $e;
		}

		return 0;
	}
472284
public function push($job, $data = '', $queue = null)
	{
		$queueJob = $this->resolveJob($this->createPayload($job, $data, $queue));

		try
		{
			$queueJob->fire();
		}
		catch(Exception $e)
		{
			$this->handleFailedJob($queueJob);

			throw $e;
		}

		return 0;
	}
645278
public function queuePush(Job $job)
    {
        if ($this->isSingle()) {
            $queue_config = $this->getQueueConfig();
            if (isset($queue_config['driver'])) {
                /** @var BaseQueue $driver */
                $driver = $queue_config['driver'];
                $driver::component()->push($job);
            }
        }
    }
245089
private function executeJob(Job $job)
    {
        $this->currentJobId = $job->getId();
        $job->setLogger($this->logger);
        $this->logger->notice("Starting job #" . $this->currentJobId);
        
        if (!function_exists('pcntl_fork')) {
            $this->runJob($job);
            return;
        }

        $pid = pcntl_fork();
        if ($pid) {
            pcntl_wait($status);
        } else {
            $this->runJob($job);
            die();
        }
    }
860759
public function push( Job $job, $delay = 0 ) {
		$result = $this->database->insert( $this->jobs_table, array(
			'job'          => serialize( $job ),
			'available_at' => $this->datetime( $delay ),
			'created_at'   => $this->datetime(),
		) );

		if ( ! $result ) {
			return false;
		}

		return $this->database->insert_id;
	}
557134
public function kickJob($id)
    {
        $payload = $this->getPayload($id);

        if ($payload[self::JOB_PAYLOAD_STATE] !== self::STATE_BURIED) {
            throw new QueueException(' Cannot kick state unsupported!' . $payload[self::JOB_PAYLOAD_STATE]);
        }

        $keyReady = $this->getKeyReady($payload[self::JOB_PAYLOAD_QUEUE]);
        $keyBuried = $this->getKeyBuried($payload[self::JOB_PAYLOAD_QUEUE]);
        $redisJobNamespace = $this->getRedisJobNamespace();

        $result = $this->getRedis()->transaction(
            function ($tx) use ($id, $payload, $keyBuried, $keyReady, $redisJobNamespace) {
                /** @var $tx \Predis\Client */
                $tx->zrem($keyBuried, $id);
                $payload = self::buildPayload($payload[self::JOB_PAYLOAD_QUEUE], self::STATE_READY);
                $tx->hset($redisJobNamespace, $id, $payload);
                $tx->rpush($keyReady, $id);
            }
        );

        return !in_array(false, $result);
    }
393220
public function put($job, $options = null)
    {
        if ($job instanceof JobInterface) {
            if ($job->isRunOnce()) {
                $cacheKey = self::CACHE_JOB . get_class($job);
                if ($this->getCache()->get($cacheKey) === null) {
                    $expire = $job->getDelay() + $job->getTtr() + 60;
                    $this->getCache()->save($cacheKey, 1, $expire);
                } else {
                    return true;
                }
            }
            $options = [
                'delay' => $job->getDelay(),
                'ttr' => $job->getTtr(),
                'priority' => $job->getPriority()
            ];
        }
        $this->getLogger()->info('Add job ' . json_encode($job));
        return parent::put($job, $options);
    }
129859
public function release(
        Job $job,
        $priority = null,
        $delay = null
    )
    {
        $priority = $priority ?? PheanstalkInterface::DEFAULT_PRIORITY;
        $delay = $delay ?? PheanstalkInterface::DEFAULT_DELAY;

        return $this->client->release($job, $priority, $delay);
    }
252527
private function pushExecutedJob(Job $job)
    {
        $this->executedJobs[] = $job;

        $compiled = $job->compile();

        // If callable, log the string Closure
        if (is_callable($compiled)) {
            $compiled = 'Closure';
        }

        $this->addSchedulerVerboseOutput("Executing {$compiled}");

        return $job;
    }
180580
public function markJobDone(QueuedJob $job) {
		$fields = [
			'completed' => $this->getDateTime(),
		];
		$job = $this->patchEntity($job, $fields);

		return (bool)$this->save($job);
	}
146433
public function reschedule(Job $job, \DateTime $date, $priority = PheanstalkInterface::DEFAULT_PRIORITY)
    {
        if ($date < new \DateTime()) {
            throw new \InvalidArgumentException(
                sprintf('You cannot reschedule a job in the past (got %s, and the current date is %s)', $date->format(DATE_ISO8601), date(DATE_ISO8601))
            );
        }

        $this->pheanstalk->release($job, $priority, $date->getTimestamp() - time());

        $this->logJob($job->getId(), sprintf('Rescheduled job for %s', $date->format('Y-m-d H:i:s')));
    }