class Worker (View source)

Traits

Properties

protected QueueManager $manager The queue manager instance.
protected Dispatcher $events The event dispatcher instance.
protected Repository $cache The cache repository implementation.
protected ExceptionHandler $exceptions The exception handler instance.
bool $shouldQuit Indicates if the worker should exit.
bool $paused Indicates if the worker is paused.

Methods

bool
causedByLostConnection(Throwable $e)

Determine if the given exception was caused by a lost connection.

void
__construct(QueueManager $manager, Dispatcher $events, ExceptionHandler $exceptions)

Create a new queue worker.

void
daemon(string $connectionName, string $queue, WorkerOptions $options)

Listen to the given queue in a loop.

void
registerTimeoutHandler(Job|null $job, WorkerOptions $options)

Register the worker timeout handler.

int
timeoutForJob(Job|null $job, WorkerOptions $options)

Get the appropriate timeout for the given job.

bool
daemonShouldRun(WorkerOptions $options, string $connectionName, string $queue)

Determine if the daemon should process on this iteration.

void
pauseWorker(WorkerOptions $options, int $lastRestart)

Pause the worker for the current loop.

stopIfNecessary(WorkerOptions $options, int $lastRestart)

Stop the process if necessary.

void
runNextJob(string $connectionName, string $queue, WorkerOptions $options)

Process the next job on the queue.

Job|null
getNextJob(Queue $connection, string $queue)

Get the next job from the queue connection.

void
runJob(Job $job, string $connectionName, WorkerOptions $options)

Process the given job.

void
stopWorkerIfLostConnection(Throwable $e)

Stop the worker if we have lost connection to a database.

void
process(string $connectionName, Job $job, WorkerOptions $options)

Process the given job from the queue.

void
handleJobException(string $connectionName, Job $job, WorkerOptions $options, Exception $e)

Handle an exception that occurred while the job was running.

void
markJobAsFailedIfAlreadyExceedsMaxAttempts(string $connectionName, Job $job, int $maxTries)

Mark the given job as failed if it has exceeded the maximum allowed attempts.

void
markJobAsFailedIfWillExceedMaxAttempts(string $connectionName, Job $job, int $maxTries, Exception $e)

Mark the given job as failed if it has exceeded the maximum allowed attempts.

void
failJob(string $connectionName, Job $job, Exception $e)

Mark the given job as failed and raise the relevant event.

void
raiseBeforeJobEvent(string $connectionName, Job $job)

Raise the before queue job event.

void
raiseAfterJobEvent(string $connectionName, Job $job)

Raise the after queue job event.

void
raiseExceptionOccurredJobEvent(string $connectionName, Job $job, Exception $e)

Raise the exception occurred queue job event.

bool
queueShouldRestart(int|null $lastRestart)

Determine if the queue worker should restart.

int|null
getTimestampOfLastQueueRestart()

Get the last queue restart timestamp, or null.

void
listenForSignals()

Enable async signals for the process.

bool
supportsAsyncSignals()

Determine if "async" signals are supported.

bool
memoryExceeded(int $memoryLimit)

Determine if the memory limit has been exceeded.

void
stop(int $status = 0)

Stop listening and bail out of the script.

void
kill(int $status = 0)

Kill the process.

void
sleep(int|float $seconds)

Sleep the script for a given number of seconds.

void
setCache(Repository $cache)

Set the cache repository implementation.

getManager()

Get the queue manager instance.

void
setManager(QueueManager $manager)

Set the queue manager instance.

Details

protected bool causedByLostConnection(Throwable $e)

Determine if the given exception was caused by a lost connection.

Parameters

Throwable $e

Return Value

bool

void __construct(QueueManager $manager, Dispatcher $events, ExceptionHandler $exceptions)

Create a new queue worker.

Parameters

QueueManager $manager
Dispatcher $events
ExceptionHandler $exceptions

Return Value

void

void daemon(string $connectionName, string $queue, WorkerOptions $options)

Listen to the given queue in a loop.

Parameters

string $connectionName
string $queue
WorkerOptions $options

Return Value

void

protected void registerTimeoutHandler(Job|null $job, WorkerOptions $options)

Register the worker timeout handler.

Parameters

Job|null $job
WorkerOptions $options

Return Value

void

protected int timeoutForJob(Job|null $job, WorkerOptions $options)

Get the appropriate timeout for the given job.

Parameters

Job|null $job
WorkerOptions $options

Return Value

int

protected bool daemonShouldRun(WorkerOptions $options, string $connectionName, string $queue)

Determine if the daemon should process on this iteration.

Parameters

WorkerOptions $options
string $connectionName
string $queue

Return Value

bool

protected void pauseWorker(WorkerOptions $options, int $lastRestart)

Pause the worker for the current loop.

Parameters

WorkerOptions $options
int $lastRestart

Return Value

void

protected stopIfNecessary(WorkerOptions $options, int $lastRestart)

Stop the process if necessary.

Parameters

WorkerOptions $options
int $lastRestart

void runNextJob(string $connectionName, string $queue, WorkerOptions $options)

Process the next job on the queue.

Parameters

string $connectionName
string $queue
WorkerOptions $options

Return Value

void

protected Job|null getNextJob(Queue $connection, string $queue)

Get the next job from the queue connection.

Parameters

Queue $connection
string $queue

Return Value

Job|null

protected void runJob(Job $job, string $connectionName, WorkerOptions $options)

Process the given job.

Parameters

Job $job
string $connectionName
WorkerOptions $options

Return Value

void

protected void stopWorkerIfLostConnection(Throwable $e)

Stop the worker if we have lost connection to a database.

Parameters

Throwable $e

Return Value

void

void process(string $connectionName, Job $job, WorkerOptions $options)

Process the given job from the queue.

Parameters

string $connectionName
Job $job
WorkerOptions $options

Return Value

void

Exceptions

Throwable

protected void handleJobException(string $connectionName, Job $job, WorkerOptions $options, Exception $e)

Handle an exception that occurred while the job was running.

Parameters

string $connectionName
Job $job
WorkerOptions $options
Exception $e

Return Value

void

Exceptions

Exception

protected void markJobAsFailedIfAlreadyExceedsMaxAttempts(string $connectionName, Job $job, int $maxTries)

Mark the given job as failed if it has exceeded the maximum allowed attempts.

This will likely be because the job previously exceeded a timeout.

Parameters

string $connectionName
Job $job
int $maxTries

Return Value

void

protected void markJobAsFailedIfWillExceedMaxAttempts(string $connectionName, Job $job, int $maxTries, Exception $e)

Mark the given job as failed if it has exceeded the maximum allowed attempts.

Parameters

string $connectionName
Job $job
int $maxTries
Exception $e

Return Value

void

protected void failJob(string $connectionName, Job $job, Exception $e)

Mark the given job as failed and raise the relevant event.

Parameters

string $connectionName
Job $job
Exception $e

Return Value

void

protected void raiseBeforeJobEvent(string $connectionName, Job $job)

Raise the before queue job event.

Parameters

string $connectionName
Job $job

Return Value

void

protected void raiseAfterJobEvent(string $connectionName, Job $job)

Raise the after queue job event.

Parameters

string $connectionName
Job $job

Return Value

void

protected void raiseExceptionOccurredJobEvent(string $connectionName, Job $job, Exception $e)

Raise the exception occurred queue job event.

Parameters

string $connectionName
Job $job
Exception $e

Return Value

void

protected bool queueShouldRestart(int|null $lastRestart)

Determine if the queue worker should restart.

Parameters

int|null $lastRestart

Return Value

bool

protected int|null getTimestampOfLastQueueRestart()

Get the last queue restart timestamp, or null.

Return Value

int|null

protected void listenForSignals()

Enable async signals for the process.

Return Value

void

protected bool supportsAsyncSignals()

Determine if "async" signals are supported.

Return Value

bool

bool memoryExceeded(int $memoryLimit)

Determine if the memory limit has been exceeded.

Parameters

int $memoryLimit

Return Value

bool

void stop(int $status = 0)

Stop listening and bail out of the script.

Parameters

int $status

Return Value

void

void kill(int $status = 0)

Kill the process.

Parameters

int $status

Return Value

void

void sleep(int|float $seconds)

Sleep the script for a given number of seconds.

Parameters

int|float $seconds

Return Value

void

void setCache(Repository $cache)

Set the cache repository implementation.

Parameters

Repository $cache

Return Value

void

QueueManager getManager()

Get the queue manager instance.

Return Value

QueueManager

void setManager(QueueManager $manager)

Set the queue manager instance.

Parameters

QueueManager $manager

Return Value

void