Listen to the given queue.

@param  string  $connection
@param  string  $queue
@param  int     $delay
@param  int     $sleep
@param  int     $maxTries
@return array
344000
public function runNextJob($connection, $queue = null, $delay = 0, $sleep = 3, $maxTries = 0)
    {
        $job = $this->getNextJob(
            $this->manager->connection($connection), $queue
        );

        // If we're able to pull a job off of the stack, we will process it and
        // then immediately return back out. If there is no job on the queue
        // we will "sleep" the worker for the specified number of seconds.

        if (! is_null($job)) {
            return $this->runJob($job, $connection, $maxTries, $delay);
        }

        $this->sleep($sleep);

        return array('job' => null, 'failed' => false);
    }
485119
public function onConnection(ConnectionInterface $connection)
    {
        $this->log('New connection from ' . $connection->getRemoteAddress());

        $buffer = '';
        $connection->on('data', function ($data) use (&$buffer, &$connection) {
            $buffer .= $data;

            if (strpos($buffer, Client::END_OF_COMMAND)) {
                $chunks = explode(Client::END_OF_COMMAND, $buffer);
                $count = count($chunks);
                $buffer = $chunks[$count - 1];

                for ($i = 0; $i < $count - 1; $i++) {
                    $command = json_decode($chunks[$i], true);
                    $this->runCommand($command, $connection);
                }
            }
        });
    }
124539
public function later($delay, Queue $queue)
    {
        $connection = property_exists($this, 'connection') ? $this->connection : null;

        $queueName = property_exists($this, 'queue') ? $this->queue : null;

        return $queue->connection($connection)->laterOn(
            $queueName ?: null, $delay, new SendQueuedMailable($this)
        );
    }
186908
public function set(string $data, string $queue = 'default', $delay = 0)
    {
        return $this->beanstalkTask->putInTube($queue, $data, null, $delay);
    }
1326697
protected function runNextJobForDaemon($connectionName, $queue, $delay, $sleep, $maxTries)
	{
		try
		{
			$this->pop($connectionName, $queue, $delay, $sleep, $maxTries);
		}
		catch (Exception $e)
		{
			if ($this->exceptions) $this->exceptions->report($e);
		}
	}
442398
private function registerClient(Connection $connection, $timeout)
    {
        $deferred = new Deferred();

        $responseTimer = $this->loop->addTimer(
            $timeout,
            function () use ($deferred, $timeout) {
                $exception = new \RuntimeException(sprintf('No response after %d seconds.', $timeout));
                $deferred->reject($exception);
            }
        );

        $this->startFlow(new OutgoingConnectFlow($connection, $this->identifierGenerator), true)
            ->always(function () use ($responseTimer) {
                $this->loop->cancelTimer($responseTimer);
            })->then(function (Connection $connection) use ($deferred) {
                $this->timer[] = $this->loop->addPeriodicTimer(
                    floor($connection->getKeepAlive() * 0.75),
                    function () {
                        $this->startFlow(new OutgoingPingFlow());
                    }
                );

                $deferred->resolve($connection);
            })->otherwise(function (\Exception $e) use ($deferred) {
                $deferred->reject($e);
            });

        return $deferred->promise();
    }
354881
protected function runWorker($connection, $queue, $delay, $memory, $daemon = false)
    {
        $this->worker->setDaemonExceptionHandler(
            $this->container['Nova\Foundation\Exceptions\HandlerInterface']
        );

        $sleep = $this->option('sleep');
        $tries = $this->option('tries');

        if (! $daemon) {
            return $this->worker->runNextJob($connection, $queue, $delay, $sleep, $tries);
        }

        $this->worker->setCache(
            $this->container['cache']->driver()
        );

        return $this->worker->daemon($connection, $queue, $delay, $memory, $sleep, $tries);
    }
1139233
public function laterOn($queue, $delay, $job, $data = '')
    {
        return $this->later($delay, $job, $data, $queue);
    }
1305218
public function delayed($delay, Job $job)
    {
        return $this->connector->delayed($this->queue, $delay, $job);
    }
170465
public function log($connection, $queue, $payload)
	{
		$failed_at = new DateTime;

		$this->getTable()->insert(compact('connection', 'queue', 'payload', 'failed_at'));
	}
1360684
public function pop( $connectionName, $queue = null, $delay = 0, $sleep = 3, $maxTries = 0 )
	{
		try
		{
			$connection = $this->manager->connection( $connectionName );

			$job = $this->getNextJob( $connection, $queue );

			// If we're able to pull a job off of the stack, we will process it and
			// then immediately return back out. If there is no job on the queue
			// we will "sleep" the worker for the specified number of seconds.
			if ( !is_null( $job ) )
				return $this->process( $this->manager->getName( $connectionName ), $job, $maxTries, $delay );
		}
		catch ( Exception $e )
		{
			if ( $this->exceptions )
				$this->exceptions->report( $e );
		}

		$this->sleep( $sleep );

		return ['job' => null, 'failed' => false];
	}
351682
public function later($delay, $job, $data = '', $queue = null)
    {
        $payload = $this->createPayload($job, $data);

        $delay = $this->getSeconds($delay);

        $this->getConnection()->zAdd($this->getQueue($queue) . ':delayed', $this->getTime() + $delay, $payload);

        return Arr::get(json_decode($payload, true), 'id');
    }
839719
public function log($connection, $queue, $payload, $exception)
    {
        $failed_at = Date::now();

        $exception = (string) $exception;

        return $this->getTable()->insertGetId(compact(
            'connection', 'queue', 'payload', 'exception', 'failed_at'
        ));
    }
79682
public function log($connection, $queue, $payload)
    {
        $failed_at = date(DateTime::MYSQL_DATETIME);
        $this->getDb()->insertAsDict($this->table, compact('connection', 'queue', 'payload', 'failed_at'));
    }
404330
protected static function build( $connection ) {
		$connections = static::connections();

		if ( empty( $connections[ $connection ] ) ) {
			throw new ConnectionNotFoundException();
		}

		static::$instances[ $connection ] = new Queue( $connections[ $connection ] );

		return static::$instances[ $connection ];
	}
686685
public static function use($connection)
    {
        if (! is_null($config = config("database.redis.clusters.{$connection}.0"))) {
            config(["database.redis.{$connection}" => $config]);
        } elseif (is_null($config) && is_null($config = config("database.redis.{$connection}"))) {
            throw new Exception("Redis connection [{$connection}] has not been configured.");
        }

        config(['database.redis.horizon' => array_merge($config, [
            'options' => ['prefix' => config('horizon.prefix') ?: 'horizon:'],
        ])]);
    }
1203231
protected function getNextJob($connection, $queue)
    {
        try {
            foreach (explode(',', $queue) as $queue) {
                if (!is_null($job = $connection->pop($queue))) {
                    return $job;
                }
            }
        } catch (Exception $e) {
            $this->report($e);

            $this->stopWorkerIfLostConnection($e);

            $this->sleep(1);
        } catch (Throwable $e) {
            $e = new FatalThrowableError($e);
            $this->report($e);

            $this->stopWorkerIfLostConnection($e);

            $this->sleep(1);
        }
    }