Worker
class Worker
Traits
Constants
| EXIT_SUCCESS |
|
| EXIT_ERROR |
|
| EXIT_MEMORY_LIMIT |
|
| RESTART_SIGNAL_CACHE_KEY |
The cache key for the restart signal. IMPORTANT: Uses Laravel's key for cross-framework queue interoperability. |
| protected HANDLED_SIGNALS |
Signals installed when the worker daemon starts. |
Properties
| protected string|null | $name | The name of the worker. |
|
| protected Repository|null | $cache | The cache repository implementation. |
|
| protected callable | $isDownForMaintenance | The callback used to determine if the application is in maintenance mode. |
|
| protected array | $runningJobs | The running jobs. |
|
| protected array | $timeoutJobIds | The timeout job IDs. |
|
| protected int|null | $monitorId | The job monitor's ID. |
|
| protected Timer | $timer | The timer that owns the timeout monitor. |
|
| protected bool | $monitorLocked | Indicates if the job monitor is checking for timeout jobs. |
|
| protected int | $jobsProcessed | The number of jobs completed by the worker. |
|
| protected float|null | $lastJobProcessedAt | The timestamp of the last completed job. |
|
| protected WorkerStopReason|null | $stopReason | The terminal reason set by an asynchronous worker failure. |
|
| bool | $shouldQuit | Indicates if the worker should exit. |
|
| bool | $paused | Indicates if the worker is paused. |
|
| static protected callable[] | $popCallbacks | The callbacks used to pop jobs from queues. |
|
| static int|null | $memoryExceededExitCode | The custom exit code to be used when memory is exceeded. |
|
| static int|null | $timeoutExceededExitCode | The custom exit code to be used when a job times out. |
|
| static bool | $reportJobExceptions | Indicates if the worker should report job exceptions. |
|
| static bool | $stopOnLostConnection | Indicates if the worker should stop when a lost connection is detected. |
|
| static bool | $restartable | Indicates if the worker should check for the restart signal in the cache. |
|
| static bool | $pausable | Indicates if the worker should check for the paused signal in the cache. |
Methods
Determine if the given exception was caused by a lost connection.
Create a new queue worker.
Listen to the given queue in a loop.
Scanning the running jobs and terminate the timeout jobs.
Determine if there are any timeout jobs.
Determine if the daemon should process on this iteration.
Pause the worker for the current loop.
Determine the exit code to stop the process if necessary.
Process the next job on the queue.
Determine which of the given queues are currently paused.
Run the callback with the worker option context active.
Process the given job from the queue.
Fail a job whose payload cannot be consumed.
Handle an exception that occurred while the job was running.
Mark the given job as failed if it has exceeded the maximum allowed attempts.
Mark the given job as failed if it has exceeded the maximum allowed attempts.
Mark the given job as failed if it has exceeded the maximum allowed attempts.
Mark the given job as failed if the exception handler determines it should not be retried.
Mark the given job as failed if it should fail on timeouts.
Raise an event indicating the worker is starting.
Raise the before job has been popped.
Raise the exception occurred queue job event.
Determine if the queue worker should restart.
Get the last queue restart timestamp, or null.
Enable async signals for the process.
Handle a worker interruption signal.
Handle a worker pause signal.
Handle a worker resume signal.
Pass the signal to the running jobs.
Determine if "async" signals are supported.
Determine if the memory limit has been exceeded.
Get the current memory usage in megabytes.
Get the current monotonic time.
Stop listening and bail out of the script.
Kill the process.
Terminate the current worker process immediately.
Sleep the script for a given number of seconds.
Register a callback to be executed to pick jobs.
Get the queue manager instance.
Flush all static state.
Details
in
DetectsLostConnections at line 16
protected bool
causedByLostConnection(Throwable $e)
Determine if the given exception was caused by a lost connection.
at line 197
__construct(Factory $manager, Dispatcher $events, ExceptionHandler $exceptions, callable $isDownForMaintenance, Timer|null $timer = null)
Create a new queue worker.
at line 211
int
daemon(string $connectionName, string $queue, WorkerOptions $options)
Listen to the given queue in a loop.
at line 340
protected void
waitForRunningJobs(Concurrent $concurrent)
Wait for every admitted job coroutine to finish.
at line 350
protected void
monitorTimeoutJobs(WorkerOptions $options)
Monitor the jobs for timeout.
at line 385
protected void
terminateTimeoutJobs(WorkerOptions $options)
Scanning the running jobs and terminate the timeout jobs.
at line 400
protected bool
hasTimeoutJobs()
Determine if there are any timeout jobs.
at line 405
protected void
handleTimeoutJob(Job $job, WorkerOptions $options)
No description
at line 435
protected int
timeoutForJob(Job|null $job, WorkerOptions $options)
Get the appropriate timeout for the given job.
at line 443
protected bool
daemonShouldRun(WorkerOptions $options, string $connectionName, string $queue)
Determine if the daemon should process on this iteration.
at line 454
protected array|null
pauseWorker(WorkerOptions $options, int|null $lastRestart, float|int $startTime, int $jobsAdmitted)
Pause the worker for the current loop.
at line 474
protected array|null
stopIfNecessary(WorkerOptions $options, int|null $lastRestart, float|int $startTime, int $jobsAdmitted, mixed $job = null, bool $checkQueueEmpty = true, bool $hasRunningJobs = false)
Determine the exit code to stop the process if necessary.
at line 503
null
runNextJob(string $connectionName, string $queue, WorkerOptions $options)
Process the next job on the queue.
at line 525
protected Job|null
getNextJob(Queue $connection, string $queue)
Get the next job from the queue connection.
at line 571
protected array
getPausedQueues(string $connectionName, array $queues)
Determine which of the given queues are currently paused.
at line 590
protected null
runJob(Job $job, string $connectionName, WorkerOptions $options)
Process the given job.
at line 610
protected mixed
withCoroutineContext(WorkerOptions $options, callable $callback)
Run the callback with the worker option context active.
at line 641
protected void
stopWorkerIfLostConnection(Throwable $e)
Stop the worker if we have lost connection to a database.
at line 653
void
process(string $connectionName, Job $job, WorkerOptions $options)
Process the given job from the queue.
at line 722
protected void
handleInvalidPayload(string $connectionName, Job $job, InvalidPayloadException $e)
Fail a job whose payload cannot be consumed.
at line 731
protected string
registerCoroutineJob(Job $job, WorkerOptions $options)
Register a coroutine job to running jobs.
at line 748
protected void
handleJobException(string $connectionName, Job $job, WorkerOptions $options, Throwable $e)
Handle an exception that occurred while the job was running.
at line 808
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.
at line 830
protected void
markJobAsFailedIfWillExceedMaxAttempts(string $connectionName, Job $job, int $maxTries, Throwable $e)
Mark the given job as failed if it has exceeded the maximum allowed attempts.
at line 846
protected void
markJobAsFailedIfWillExceedMaxExceptions(string $connectionName, Job $job, Throwable $e)
Mark the given job as failed if it has exceeded the maximum allowed attempts.
at line 872
protected void
markJobAsFailedIfItShouldntBeRetried(string $connectionName, Job $job, Throwable $e)
Mark the given job as failed if the exception handler determines it should not be retried.
at line 883
protected void
markJobAsFailedIfItShouldFailOnTimeout(string $connectionName, Job $job, Throwable $e)
Mark the given job as failed if it should fail on timeouts.
at line 893
protected void
failJob(Job $job, Throwable $e)
Mark the given job as failed and raise the relevant event.
at line 901
protected int
calculateBackoff(Job $job, WorkerOptions $options)
Calculate the backoff for the given job.
at line 915
protected void
raiseWorkerStartingEvent(string $connectionName, string $queue, WorkerOptions $options)
Raise an event indicating the worker is starting.
at line 923
protected void
raiseBeforeJobPopEvent(string $connectionName, string|null $queue = null)
Raise the before job has been popped.
at line 933
protected void
raiseAfterJobPopEvent(string $connectionName, Job|null $job)
Raise the after job has been popped.
at line 946
protected void
raiseBeforeJobEvent(string $connectionName, Job|null $job)
Raise the before queue job event.
at line 957
protected void
raiseAfterJobEvent(string $connectionName, Job $job)
Raise the after queue job event.
at line 968
protected void
raiseExceptionOccurredJobEvent(string $connectionName, Job|null $job, Throwable $e)
Raise the exception occurred queue job event.
at line 980
protected bool
queueShouldRestart(int|null $lastRestart)
Determine if the queue worker should restart.
at line 992
protected int|null
getTimestampOfLastQueueRestart()
Get the last queue restart timestamp, or null.
at line 1010
protected void
listenForSignals(string $connectionName, string $queue, WorkerOptions $options)
Enable async signals for the process.
at line 1034
protected void
handleInterruptionSignal(int $signal, string $connectionName, string $queue, WorkerOptions $options)
Handle a worker interruption signal.
at line 1046
protected void
handlePauseSignal(string $connectionName, string $queue, WorkerOptions $options)
Handle a worker pause signal.
at line 1056
protected void
handleResumeSignal(string $connectionName, string $queue, WorkerOptions $options)
Handle a worker resume signal.
at line 1066
protected void
notifyJobsOfSignal(int $signal)
Pass the signal to the running jobs.
at line 1095
protected bool
supportsAsyncSignals()
Determine if "async" signals are supported.
at line 1103
bool
memoryExceeded(float $memoryLimit)
Determine if the memory limit has been exceeded.
at line 1111
protected float
currentMemoryUsage()
Get the current memory usage in megabytes.
at line 1119
protected float
currentTime()
Get the current monotonic time.
at line 1127
int
stop(int $status = 0, WorkerOptions|null $options = null, WorkerStopReason|null $reason = null)
Stop listening and bail out of the script.
at line 1144
never
kill(int $status = 0, WorkerOptions|null $options = null, WorkerStopReason|null $reason = null)
Kill the process.
at line 1164
protected never
terminateProcess(int $status)
Terminate the current worker process immediately.
at line 1176
protected MaxAttemptsExceededException
maxAttemptsExceededException(Job $job)
Create an instance of MaxAttemptsExceededException.
at line 1184
protected TimeoutExceededException
timeoutExceededException(Job|null $job)
Create an instance of TimeoutExceededException.
at line 1192
void
sleep(float|int $seconds)
Sleep the script for a given number of seconds.
at line 1203
Worker
setCache(Repository $cache)
Set the cache repository implementation.
Boot-only. The repository persists on the singleton worker for the worker lifetime and is read by every subsequent pause or restart check.
at line 1216
Worker
setName(string $name)
Set the name of the worker.
Boot-only. The name persists on the singleton worker for the worker lifetime and selects the static pop callback used by every job pop.
at line 1230
static void
popUsing(string $workerName, callable|null $callback)
Register a callback to be executed to pick jobs.
Boot-only. The callback persists in a static property for the worker process lifetime and runs on every job pop for the named worker. Passing null removes the callback.
at line 1242
Factory
getManager()
Get the queue manager instance.
at line 1253
void
setManager(Factory $manager)
Set the queue manager instance.
Tests only. Swaps the manager reference on the singleton worker; runtime use races across coroutines and changes every concurrent queue lookup.
at line 1261
static void
flushState()
Flush all static state.