Documentation

Worker
in package
implements ArrayAccess, Countable, IteratorAggregate

Queue worker class

Tags
category

Pop

author

Nick Sagona, III nick@popphp.org

copyright

Copyright (c) 2009-2026 Nick Sagona, III

license

https://www.popphp.org/license New BSD License

version
3.0.0

Table of Contents

Interfaces

ArrayAccess
Countable
IteratorAggregate

Properties

$application  : Application|null
Application object
$events  : Manager|null
Event manager, for worker-level lifecycle observability hooks (worker.work_loop.*, worker.run_loop.*). If not set, and this Worker was constructed with an Application that has its own event manager, that Application's manager is used instead - see triggerEvent(). If neither is available, event firing is a silent no-op.
$name  : string|null
Optional operator-facing label for this worker, surfaced in its registry record
$queues  : array<string|int, mixed>
Queues
$registry  : WorkerRegistry|null
Worker registry, for observability. Null (the default) means no registry work happens at all and behavior is identical to a build without this feature.
$stopped  : bool
Whether a graceful shutdown has been requested, via stop() directly or via a caught SIGTERM/SIGINT (see installSignalHandlers()).
$weights  : array<string|int, mixed>
Queue weights, keyed by queue name. Higher services first. Not named "priority" - that word is already used elsewhere in this codebase for the unrelated FIFO/FILO adapter job-ordering setting.

Methods

__construct()  : mixed
Constructor
__get()  : Queue|null
Get a queue
__isset()  : bool
Determine if a queue is registered with the worker object
__set()  : void
Register a queue with the worker
__unset()  : void
Unset a queue with the worker
addQueue()  : Worker
Add queue
addQueues()  : Worker
Add queues
application()  : Application|null
Get the application (alias)
clear()  : Worker
Clear jobs from queue
clearAll()  : Worker
Clear all jobs from queues
clearAllFailed()  : Worker
Clear all failed jobs from queues
clearAllTasks()  : Worker
Clear all tasks from queues
clearFailed()  : Worker
Clear failed jobs from queue
clearTasks()  : Worker
Clear tasks from queue
count()  : int
Return count
create()  : Worker
Create queue worker worker
events()  : Manager|null
Get event manager (alias)
getApplication()  : Application|null
Get the application
getEvents()  : Manager|null
Get event manager
getIterator()  : ArrayIterator
Get iterator
getName()  : string|null
Get this worker's operator-facing label
getQueue()  : Queue|null
Get queue
getQueues()  : array<string|int, mixed>
Get queues
getRegistry()  : WorkerRegistry|null
Get the worker registry
getWeight()  : int
Get a queue's weight (0 if never set)
hasApplication()  : bool
Has application
hasEvents()  : bool
Has event manager
hasName()  : bool
Has an operator-facing label
hasQueue()  : bool
Has queue
hasRegistry()  : bool
Has a worker registry
isStopped()  : bool
Whether a graceful shutdown has been requested
offsetExists()  : bool
Determine if a queue is registered with the worker object
offsetGet()  : Queue|null
Get a queue
offsetSet()  : void
Set a queue with the worker
offsetUnset()  : void
Unset a queue from the worker
registry()  : WorkerRegistry|null
Get the worker registry (alias)
run()  : array<string|int, mixed>
Run next scheduled task in queue
runAll()  : array<string|int, mixed>
Run next scheduled task across all queues, fairly: every queue gets one shared evaluation pass immediately, then - only if at least one queue has a sub-minute task - up to 59 more shared passes (one per second, sleeping once per tick, not once per queue per tick), mirroring Queue::run()'s own pass-1-then-tick-loop shape one level up. No single queue's tick-loop work can block another queue's evaluation on the same tick.
runLoop()  : void
Run scheduled tasks across all registered queues, forever, until stopped. Calls runAll() every iteration; sleeps $sleepSeconds only when a full pass finds nothing due anywhere, looping again immediately otherwise. runAll() already blocks appropriately on its own whenever a sub-minute task exists, so this backoff sleep only ever triggers on the coarse-only-or-nothing-scheduled case, which returns near-instantly and would otherwise busy-loop. A stop signal arriving mid-runAll() (during its own internal sub-minute tick loop) isn't noticed until that call returns - see the design spec's Non-goals for why this latency is accepted rather than fixed here.
setEvents()  : Worker
Set event manager
setName()  : Worker
Set this worker's operator-facing label
setRegistry()  : Worker
Set the worker registry, and wire its tracking onto this worker's queues
stop()  : Worker
Request a graceful shutdown of a running workLoop()/runLoop() call.
work()  : AbstractJob|null
Work next job. Pass a queue name to work that specific queue (exactly today's behavior). Pass nothing to try every registered queue in weight order (highest first), returning the first job successfully claimed - the highest-weight-first worker model, since workAll() fans out to every queue regardless of weight and doesn't need this.
workAll()  : array<string|int, mixed>
Work next job across in all queues
workLoop()  : void
Work jobs across all registered queues, forever, until stopped.
deregisterIfOwned()  : void
Deregister this process, but only if the given flag says this call owned the registration
ensureRegistered()  : bool
Register this process with the registry if nothing has already, returning whether THIS call performed the registration.
getQueuesByWeight()  : array<string|int, mixed>
Get queues ordered by weight, highest first. PHP's sort functions are stable since 8.0, so queues with equal weight (including the default-zero case when no weight was ever set) keep their original insertion order automatically.
heartbeat()  : void
Refresh this worker's heartbeat, if it has a registry. A no-op otherwise, and a no-op inside the registry when not registered.
installSignalHandlers()  : void
Install SIGTERM/SIGINT handlers that request a graceful stop(), rather than letting the OS terminate the process immediately. A no-op when ext-pcntl isn't loaded - workLoop()/runLoop() still run correctly without it, just without OS-signal-based shutdown available (only stop() can end them in that case).
triggerEvent()  : void
Trigger a worker-level lifecycle event. Uses this Worker's own event manager if one is set via setEvents(); otherwise falls back to the event manager of the Application this Worker was constructed with, if it has one; otherwise does nothing. Never throws on its own account - if the resolved manager's trigger() call throws (e.g. a listener's own code throws), that exception propagates to the caller exactly as any other uncaught exception would.

Properties

$events

Event manager, for worker-level lifecycle observability hooks (worker.work_loop.*, worker.run_loop.*). If not set, and this Worker was constructed with an Application that has its own event manager, that Application's manager is used instead - see triggerEvent(). If neither is available, event firing is a silent no-op.

protected Manager|null $events = null

$name

Optional operator-facing label for this worker, surfaced in its registry record

protected string|null $name = null

$queues

Queues

protected array<string|int, mixed> $queues = []

$registry

Worker registry, for observability. Null (the default) means no registry work happens at all and behavior is identical to a build without this feature.

protected WorkerRegistry|null $registry = null

$stopped

Whether a graceful shutdown has been requested, via stop() directly or via a caught SIGTERM/SIGINT (see installSignalHandlers()).

protected bool $stopped = false

workLoop()/runLoop() each reset this to false at their own start.

$weights

Queue weights, keyed by queue name. Higher services first. Not named "priority" - that word is already used elsewhere in this codebase for the unrelated FIFO/FILO adapter job-ordering setting.

protected array<string|int, mixed> $weights = []

Methods

__construct()

Constructor

public __construct([mixed $queues = null ][, Application|null $application = null ]) : mixed

Instantiate the queue worker object.

Parameters
$queues : mixed = null
$application : Application|null = null

__get()

Get a queue

public __get(string $name) : Queue|null
Parameters
$name : string
Return values
Queue|null

__isset()

Determine if a queue is registered with the worker object

public __isset(string $name) : bool
Parameters
$name : string
Return values
bool

__set()

Register a queue with the worker

public __set(string $name, mixed $value) : void
Parameters
$name : string
$value : mixed

__unset()

Unset a queue with the worker

public __unset(string $name) : void
Parameters
$name : string

addQueue()

Add queue

public addQueue(Queue $queue[, int $weight = 0 ]) : Worker
Parameters
$queue : Queue
$weight : int = 0
Return values
Worker

addQueues()

Add queues

public addQueues(array<string|int, mixed> $queues) : Worker
Parameters
$queues : array<string|int, mixed>
Return values
Worker

clear()

Clear jobs from queue

public clear(string $queueName) : Worker
Parameters
$queueName : string
Return values
Worker

clearAllFailed()

Clear all failed jobs from queues

public clearAllFailed() : Worker
Return values
Worker

clearAllTasks()

Clear all tasks from queues

public clearAllTasks() : Worker
Return values
Worker

clearFailed()

Clear failed jobs from queue

public clearFailed(string $queueName) : Worker
Parameters
$queueName : string
Return values
Worker

clearTasks()

Clear tasks from queue

public clearTasks(string $queueName) : Worker
Parameters
$queueName : string
Return values
Worker

count()

Return count

public count() : int
Return values
int

create()

Create queue worker worker

public static create([mixed $queues = null ][, Application|null $application = null ]) : Worker
Parameters
$queues : mixed = null
$application : Application|null = null
Return values
Worker

getIterator()

Get iterator

public getIterator() : ArrayIterator
Return values
ArrayIterator

getName()

Get this worker's operator-facing label

public getName() : string|null
Return values
string|null

getQueue()

Get queue

public getQueue(string $queue) : Queue|null
Parameters
$queue : string
Return values
Queue|null

getQueues()

Get queues

public getQueues() : array<string|int, mixed>
Return values
array<string|int, mixed>

getWeight()

Get a queue's weight (0 if never set)

public getWeight(string $queueName) : int
Parameters
$queueName : string
Return values
int

hasApplication()

Has application

public hasApplication() : bool
Return values
bool

hasEvents()

Has event manager

public hasEvents() : bool
Return values
bool

hasName()

Has an operator-facing label

public hasName() : bool
Return values
bool

hasQueue()

Has queue

public hasQueue(string $queue) : bool
Parameters
$queue : string
Return values
bool

hasRegistry()

Has a worker registry

public hasRegistry() : bool
Return values
bool

isStopped()

Whether a graceful shutdown has been requested

public isStopped() : bool
Return values
bool

offsetExists()

Determine if a queue is registered with the worker object

public offsetExists(mixed $offset) : bool
Parameters
$offset : mixed
Return values
bool

offsetGet()

Get a queue

public offsetGet(mixed $offset) : Queue|null
Parameters
$offset : mixed
Return values
Queue|null

offsetSet()

Set a queue with the worker

public offsetSet(mixed $offset, mixed $value) : void
Parameters
$offset : mixed
$value : mixed

offsetUnset()

Unset a queue from the worker

public offsetUnset(string $offset) : void
Parameters
$offset : string

run()

Run next scheduled task in queue

public run(string $queueName) : array<string|int, mixed>
Parameters
$queueName : string
Return values
array<string|int, mixed>

runAll()

Run next scheduled task across all queues, fairly: every queue gets one shared evaluation pass immediately, then - only if at least one queue has a sub-minute task - up to 59 more shared passes (one per second, sleeping once per tick, not once per queue per tick), mirroring Queue::run()'s own pass-1-then-tick-loop shape one level up. No single queue's tick-loop work can block another queue's evaluation on the same tick.

public runAll() : array<string|int, mixed>
Return values
array<string|int, mixed>

runLoop()

Run scheduled tasks across all registered queues, forever, until stopped. Calls runAll() every iteration; sleeps $sleepSeconds only when a full pass finds nothing due anywhere, looping again immediately otherwise. runAll() already blocks appropriately on its own whenever a sub-minute task exists, so this backoff sleep only ever triggers on the coarse-only-or-nothing-scheduled case, which returns near-instantly and would otherwise busy-loop. A stop signal arriving mid-runAll() (during its own internal sub-minute tick loop) isn't noticed until that call returns - see the design spec's Non-goals for why this latency is accepted rather than fixed here.

public runLoop([int $sleepSeconds = 1 ]) : void

A negative $sleepSeconds is silently clamped to 0.

Parameters
$sleepSeconds : int = 1

setName()

Set this worker's operator-facing label

public setName(string|null $name) : Worker
Parameters
$name : string|null
Return values
Worker

stop()

Request a graceful shutdown of a running workLoop()/runLoop() call.

public stop() : Worker

Takes effect at that loop's next iteration boundary - never mid-job or mid-task-evaluation.

Return values
Worker

work()

Work next job. Pass a queue name to work that specific queue (exactly today's behavior). Pass nothing to try every registered queue in weight order (highest first), returning the first job successfully claimed - the highest-weight-first worker model, since workAll() fans out to every queue regardless of weight and doesn't need this.

public work([string|null $queueName = null ]) : AbstractJob|null
Parameters
$queueName : string|null = null
Return values
AbstractJob|null

workAll()

Work next job across in all queues

public workAll() : array<string|int, mixed>
Return values
array<string|int, mixed>

workLoop()

Work jobs across all registered queues, forever, until stopped.

public workLoop([int $sleepSeconds = 1 ]) : void

Calls workAll() every iteration; sleeps $sleepSeconds only when a full pass finds nothing anywhere (every queue returned null), looping again immediately otherwise. Stoppable via stop() directly or, when ext-pcntl is loaded, via SIGTERM/SIGINT - either way, the current iteration's job is never torn down mid-execution and its remaining code always runs to completion - though a blocking call inside that code (e.g. sleep()) can itself be interrupted early if a signal lands during it; see README's Daemon mode section. A negative $sleepSeconds is silently clamped to 0.

Parameters
$sleepSeconds : int = 1

deregisterIfOwned()

Deregister this process, but only if the given flag says this call owned the registration

protected deregisterIfOwned(bool $owned) : void
Parameters
$owned : bool

ensureRegistered()

Register this process with the registry if nothing has already, returning whether THIS call performed the registration.

protected ensureRegistered(string $mode) : bool

That return value is the ownership rule that lets one mechanism serve both deployment models: a single-pass work()/run() that registered deregisters itself on the way out, while the same call made from inside a daemon loop finds the loop's registration already in place, takes no ownership, and leaves the daemon's record alone.

Parameters
$mode : string
Return values
bool

getQueuesByWeight()

Get queues ordered by weight, highest first. PHP's sort functions are stable since 8.0, so queues with equal weight (including the default-zero case when no weight was ever set) keep their original insertion order automatically.

protected getQueuesByWeight() : array<string|int, mixed>
Return values
array<string|int, mixed>

heartbeat()

Refresh this worker's heartbeat, if it has a registry. A no-op otherwise, and a no-op inside the registry when not registered.

protected heartbeat() : void

installSignalHandlers()

Install SIGTERM/SIGINT handlers that request a graceful stop(), rather than letting the OS terminate the process immediately. A no-op when ext-pcntl isn't loaded - workLoop()/runLoop() still run correctly without it, just without OS-signal-based shutdown available (only stop() can end them in that case).

protected installSignalHandlers() : void

Unlike Queue::runWithTimeout()'s SIGALRM handler next door, these handlers are never restored to SIG_DFL once installed - not even after workLoop()/runLoop() returns. This is a deliberate, accepted tradeoff for this pass rather than an oversight: a second SIGTERM/SIGINT sent after a loop has already ended is simply inert (caught by this handler and ignored) instead of terminating the process via the OS default.

triggerEvent()

Trigger a worker-level lifecycle event. Uses this Worker's own event manager if one is set via setEvents(); otherwise falls back to the event manager of the Application this Worker was constructed with, if it has one; otherwise does nothing. Never throws on its own account - if the resolved manager's trigger() call throws (e.g. a listener's own code throws), that exception propagates to the caller exactly as any other uncaught exception would.

protected triggerEvent(string $name, array<string|int, mixed> $params) : void
Parameters
$name : string
$params : array<string|int, mixed> 

        

        
    




    

    

                                

                

                                
    

        On this page

        
    


                

                            

            

    

        

            
Search results