class QueueManager implements Factory, Monitor mixin Queue

Traits

Hosts must declare a protected array $poolableDrivers containing their default poolable drivers.

Properties

protected Closure> $releaseCallbacks from  HasPoolProxy
protected array $connections

The array of resolved queue connections.

protected array $connectors

The array of resolved queue connectors.

protected array $poolableDrivers

The array of drivers which will be wrapped as pool proxies.

protected QueuePoolProxy>> $poolProxyClasses

The pool proxy classes for drivers with supplemental queue capabilities.

Methods

mixed
createPoolProxy(string $driver, Closure $createCallback, PoolDefinition $definition, string $proxyClass)

Create a pool proxy for an immutable definition.

poolDefinition(string $resource, array $poolConfig, array $fingerprintSource)

Build a namespaced pool definition for a pooled resource.

poolFactory()

Get the shared object-pool factory.

setReleaseCallback(string $driver, Closure $callback)

Set the release callback for a pooled driver.

Closure|null
getReleaseCallback(string $driver)

Get the release callback for a pooled driver.

addPoolableDriver(string $driver)

Add a driver to the poolable-driver list.

removePoolableDriver(string $driver)

Remove a driver from the poolable-driver list.

array
getPoolableDrivers()

Get the poolable-driver list.

setPoolableDrivers(array $poolableDrivers)

Set the poolable-driver list.

int
secondsUntil(DateInterval|DateTimeInterface|int $delay)

Get the number of seconds until the given DateTime.

int
availableAt(DateInterval|DateTimeInterface|int|null $delay = 0)

Get the "available at" UNIX timestamp.

parseDateInterval(DateInterval|DateTimeInterface|int|null $delay)

If the given value is an interval, convert it to a DateTime instance.

int
currentTime()

Get the current system time as a UNIX timestamp.

string
runTimeForHumans(float $startTime, float|null $endTime = null)

Given a start time, format the total run time for human readability.

string|null
resolveConnectionFromQueueRoute(object $queueable, UnitEnum|string|null $queue = null)

Resolve the default connection name for a given queueable instance.

string|null
resolveQueueFromQueueRoute(object $queueable)

Resolve the default queue name for a given queueable instance.

queueRoutes()

Get the queue routes manager instance.

queueRoutesContainer()

Get the container that owns the queue routes.

__construct(Container $app)

Create a new queue manager instance.

void
before(mixed $callback)

Register an event listener for the before job event.

void
after(mixed $callback)

Register an event listener for the after job event.

void
exceptionOccurred(mixed $callback)

Register an event listener for the exception occurred job event.

void
looping(mixed $callback)

Register an event listener for the daemon queue loop.

void
failing(mixed $callback)

Register an event listener for the failed job event.

void
starting(mixed $callback)

Register an event listener for the daemon queue starting.

void
stopping(mixed $callback)

Register an event listener for the daemon queue stopping.

void
whenJobRunsLongerThan(DateTimeInterface|CarbonInterval|float|int $threshold, callable $handler)

Register a callback to be invoked when a job runs for longer than a given amount of time.

void
route(array|string $class, UnitEnum|string|null $queue = null, UnitEnum|string|null $connection = null)

Register the default queue route for a given class.

void
forward(array|string|UnitEnum $queue, UnitEnum|string|null $to = null, UnitEnum|string|null $connection = null)

Forward the given queue to another queue and/or connection.

void
pause(UnitEnum|string $queue, UnitEnum|string|null $connection = null)

Pause a queue by its name and connection.

void
pauseFor(UnitEnum|string $queue, DateInterval|DateTimeInterface|int $ttl, UnitEnum|string|null $connection = null)

Pause a queue by its name and connection for a given amount of time.

void
pauseAll()

Pause job processing for all queues on all connections.

void
resume(UnitEnum|string $queue, UnitEnum|string|null $connection = null)

Resume a paused queue by its name and connection.

void
resumeAll()

Resume job processing for all queues on all connections.

bool
isPaused(UnitEnum|string $queue, UnitEnum|string|null $connection = null)

Determine if a queue is paused.

array
getPausedQueues(array $queues, string|null $connection = null)

Determine which of the given queues are currently paused.

void
withoutInterruptionPolling()

Indicate that queue workers should not poll for restart or pause signals.

bool
connected(UnitEnum|string|null $name = null)

Determine if the driver is connected.

connection(UnitEnum|string|null $name = null)

Resolve a queue connection instance.

resolve(string $name)

Resolve a queue connection.

void
ensureConnectionsAreAcyclic(array $config, string $driver, array $path = [])

Reject circular definitions owned by the resolved failover connector.

getConnector(string $driver)

Get the connector for a given driver.

void
extend(string $driver, Closure $resolver)

Add a queue connection resolver.

void
addConnector(string $driver, Closure $resolver)

Add a queue connection resolver.

array|null
getConfig(string $name)

Get the queue connection configuration.

string
getDefaultDriver()

Get the name of the default queue connection.

void
setDefaultDriver(UnitEnum|string $name)

Set the name of the default queue connection.

string
getName(string|null $connection = null)

Get the full name for the given connection.

void
purge(string|null $name = null)

Disconnect a queue connection and close its shared resource pool.

getApplication()

Get the application instance used by the manager.

setApplication(Container $app)

Set the application instance used by the manager.

void
createPayloadUsing(callable|null $callback)

Register a callback to be executed when creating job payloads.

mixed
__call(string $method, array $parameters)

Dynamically pass calls to the default connection.

Details

in HasPoolProxy at line 28
protected mixed createPoolProxy(string $driver, Closure $createCallback, PoolDefinition $definition, string $proxyClass)

Create a pool proxy for an immutable definition.

Parameters

string $driver
Closure $createCallback
PoolDefinition $definition
string $proxyClass

Return Value

mixed

in HasPoolProxy at line 49
protected PoolDefinition poolDefinition(string $resource, array $poolConfig, array $fingerprintSource)

Build a namespaced pool definition for a pooled resource.

Parameters

string $resource
array $poolConfig
array $fingerprintSource

Return Value

PoolDefinition

at line 571
protected Factory poolFactory()

Get the shared object-pool factory.

Return Value

Factory

in HasPoolProxy at line 75
HasPoolProxy setReleaseCallback(string $driver, Closure $callback)

Set the release callback for a pooled driver.

The callback persists on the shared manager for the worker lifetime and is captured by every subsequently created proxy for the driver.

Parameters

string $driver
Closure $callback

Return Value

HasPoolProxy

in HasPoolProxy at line 86
Closure|null getReleaseCallback(string $driver)

Get the release callback for a pooled driver.

Parameters

string $driver

Return Value

Closure|null

in HasPoolProxy at line 98
HasPoolProxy addPoolableDriver(string $driver)

Add a driver to the poolable-driver list.

The list persists on the manager for the worker lifetime and is consulted on subsequent driver creation. Per-request use races across coroutines and does not affect already-cached drivers.

Parameters

string $driver

Return Value

HasPoolProxy

in HasPoolProxy at line 115
HasPoolProxy removePoolableDriver(string $driver)

Remove a driver from the poolable-driver list.

The list persists on the manager for the worker lifetime and is consulted on subsequent driver creation. Per-request use races across coroutines and does not affect already-cached drivers.

Parameters

string $driver

Return Value

HasPoolProxy

in HasPoolProxy at line 133
array getPoolableDrivers()

Get the poolable-driver list.

Return Value

array

in HasPoolProxy at line 145
HasPoolProxy setPoolableDrivers(array $poolableDrivers)

Set the poolable-driver list.

The list persists on the manager for the worker lifetime and is consulted on subsequent driver creation. Per-request use races across coroutines and does not affect already-cached drivers.

Parameters

array $poolableDrivers

Return Value

HasPoolProxy

in InteractsWithTime at line 17
protected int secondsUntil(DateInterval|DateTimeInterface|int $delay)

Get the number of seconds until the given DateTime.

Parameters

DateInterval|DateTimeInterface|int $delay

Return Value

int

in InteractsWithTime at line 29
protected int availableAt(DateInterval|DateTimeInterface|int|null $delay = 0)

Get the "available at" UNIX timestamp.

Parameters

DateInterval|DateTimeInterface|int|null $delay

Return Value

int

in InteractsWithTime at line 47
protected DateTimeInterface|int parseDateInterval(DateInterval|DateTimeInterface|int|null $delay)

If the given value is an interval, convert it to a DateTime instance.

Parameters

DateInterval|DateTimeInterface|int|null $delay

Return Value

DateTimeInterface|int

in InteractsWithTime at line 63
protected int currentTime()

Get the current system time as a UNIX timestamp.

Return Value

int

in InteractsWithTime at line 71
protected string runTimeForHumans(float $startTime, float|null $endTime = null)

Given a start time, format the total run time for human readability.

Parameters

float $startTime
float|null $endTime

Return Value

string

string|null resolveConnectionFromQueueRoute(object $queueable, UnitEnum|string|null $queue = null)

Resolve the default connection name for a given queueable instance.

Parameters

object $queueable
UnitEnum|string|null $queue

the caller-selected queue, overriding the queueable's queue when resolving a forwarded connection

Return Value

string|null

string|null resolveQueueFromQueueRoute(object $queueable)

Resolve the default queue name for a given queueable instance.

Parameters

object $queueable

Return Value

string|null

protected QueueRoutes queueRoutes()

Get the queue routes manager instance.

Return Value

QueueRoutes

at line 237
protected Container queueRoutesContainer()

Get the container that owns the queue routes.

Return Value

Container

at line 84
__construct(Container $app)

Create a new queue manager instance.

Parameters

Container $app

at line 95
void before(mixed $callback)

Register an event listener for the before job event.

The listener persists on the singleton event dispatcher for the worker lifetime and runs for every subsequent matching queue event.

Parameters

mixed $callback

Return Value

void

at line 108
void after(mixed $callback)

Register an event listener for the after job event.

The listener persists on the singleton event dispatcher for the worker lifetime and runs for every subsequent matching queue event.

Parameters

mixed $callback

Return Value

void

at line 121
void exceptionOccurred(mixed $callback)

Register an event listener for the exception occurred job event.

The listener persists on the singleton event dispatcher for the worker lifetime and runs for every subsequent matching queue event.

Parameters

mixed $callback

Return Value

void

at line 134
void looping(mixed $callback)

Register an event listener for the daemon queue loop.

The listener persists on the singleton event dispatcher for the worker lifetime and runs for every subsequent matching queue event.

Parameters

mixed $callback

Return Value

void

at line 147
void failing(mixed $callback)

Register an event listener for the failed job event.

The listener persists on the singleton event dispatcher for the worker lifetime and runs for every subsequent matching queue event.

Parameters

mixed $callback

Return Value

void

at line 160
void starting(mixed $callback)

Register an event listener for the daemon queue starting.

The listener persists on the singleton event dispatcher for the worker lifetime and runs for every subsequent matching queue event.

Parameters

mixed $callback

Return Value

void

at line 173
void stopping(mixed $callback)

Register an event listener for the daemon queue stopping.

The listener persists on the singleton event dispatcher for the worker lifetime and runs for every subsequent matching queue event.

Parameters

mixed $callback

Return Value

void

at line 186
void whenJobRunsLongerThan(DateTimeInterface|CarbonInterval|float|int $threshold, callable $handler)

Register a callback to be invoked when a job runs for longer than a given amount of time.

The listener persists on the singleton event dispatcher for the worker lifetime and runs for every subsequent matching queue event.

Parameters

DateTimeInterface|CarbonInterval|float|int $threshold
callable $handler

Return Value

void

at line 214
void route(array|string $class, UnitEnum|string|null $queue = null, UnitEnum|string|null $connection = null)

Register the default queue route for a given class.

The route persists on the singleton QueueRoutes registry for the worker lifetime and affects every subsequent dispatch of that class.

Parameters

array|string $class
UnitEnum|string|null $queue
UnitEnum|string|null $connection

Return Value

void

at line 228
void forward(array|string|UnitEnum $queue, UnitEnum|string|null $to = null, UnitEnum|string|null $connection = null)

Forward the given queue to another queue and/or connection.

Forwards persist on the singleton QueueRoutes registry for the worker lifetime and affect every subsequent dispatch and queue operation.

Parameters

array|string|UnitEnum $queue
UnitEnum|string|null $to
UnitEnum|string|null $connection

Return Value

void

at line 245
void pause(UnitEnum|string $queue, UnitEnum|string|null $connection = null)

Pause a queue by its name and connection.

Parameters

UnitEnum|string $queue
UnitEnum|string|null $connection

Return Value

void

at line 267
void pauseFor(UnitEnum|string $queue, DateInterval|DateTimeInterface|int $ttl, UnitEnum|string|null $connection = null)

Pause a queue by its name and connection for a given amount of time.

Parameters

UnitEnum|string $queue
DateInterval|DateTimeInterface|int $ttl
UnitEnum|string|null $connection

Return Value

void

at line 289
void pauseAll()

Pause job processing for all queues on all connections.

Return Value

void

at line 307
void resume(UnitEnum|string $queue, UnitEnum|string|null $connection = null)

Resume a paused queue by its name and connection.

Parameters

UnitEnum|string $queue
UnitEnum|string|null $connection

Return Value

void

at line 331
void resumeAll()

Resume job processing for all queues on all connections.

Queues paused individually are not affected.

Return Value

void

at line 349
bool isPaused(UnitEnum|string $queue, UnitEnum|string|null $connection = null)

Determine if a queue is paused.

Parameters

UnitEnum|string $queue
UnitEnum|string|null $connection

Return Value

bool

at line 365
array getPausedQueues(array $queues, string|null $connection = null)

Determine which of the given queues are currently paused.

Parameters

array $queues
string|null $connection

Return Value

array

at line 398
void withoutInterruptionPolling()

Indicate that queue workers should not poll for restart or pause signals.

This prevents the workers from hitting the application cache to determine if they need to pause or restart.

Mutates process-global worker flags; runtime use races across coroutines and changes every concurrent worker pause or restart check.

Return Value

void

at line 408
bool connected(UnitEnum|string|null $name = null)

Determine if the driver is connected.

Parameters

UnitEnum|string|null $name

Return Value

bool

at line 424
Queue connection(UnitEnum|string|null $name = null)

Resolve a queue connection instance.

Parameters

UnitEnum|string|null $name

Return Value

Queue

at line 449
protected Queue resolve(string $name)

Resolve a queue connection.

Parameters

string $name

Return Value

Queue

Exceptions

InvalidArgumentException

at line 490
protected void ensureConnectionsAreAcyclic(array $config, string $driver, array $path = [])

Reject circular definitions owned by the resolved failover connector.

Parameters

array $config
string $driver
array $path

Return Value

void

at line 523
protected ConnectorInterface getConnector(string $driver)

Get the connector for a given driver.

Parameters

string $driver

Return Value

ConnectorInterface

Exceptions

InvalidArgumentException

at line 538
void extend(string $driver, Closure $resolver)

Add a queue connection resolver.

The resolver persists in the singleton's connectors array for the worker lifetime and applies to every subsequent connection.

Parameters

string $driver
Closure $resolver

Return Value

void

at line 550
void addConnector(string $driver, Closure $resolver)

Add a queue connection resolver.

The resolver persists in the singleton's connectors array for the worker lifetime and applies to every subsequent connection.

Parameters

string $driver
Closure $resolver

Return Value

void

at line 559
protected array|null getConfig(string $name)

Get the queue connection configuration.

Parameters

string $name

Return Value

array|null

at line 579
string getDefaultDriver()

Get the name of the default queue connection.

Return Value

string

at line 589
void setDefaultDriver(UnitEnum|string $name)

Set the name of the default queue connection.

Mutates process-global config; per-request use races across coroutines.

Parameters

UnitEnum|string $name

Return Value

void

at line 600
string getName(string|null $connection = null)

Get the full name for the given connection.

Parameters

string|null $connection

Return Value

string

at line 613
void purge(string|null $name = null)

Disconnect a queue connection and close its shared resource pool.

Use this to recover broken pooled resources. Other connections sharing the pool transparently acquire a fresh pool on their next operation.

Parameters

string|null $name

Return Value

void

at line 649
Container getApplication()

Get the application instance used by the manager.

Return Value

Container

at line 663
QueueManager setApplication(Container $app)

Set the application instance used by the manager.

Swaps the container reference on the singleton and on every cached queue connection; per-request use races across coroutines and breaks every concurrent dispatch.

Parameters

Container $app

Return Value

QueueManager

at line 689
void createPayloadUsing(callable|null $callback)

Register a callback to be executed when creating job payloads.

The callback persists in a static property for the worker lifetime and runs on every subsequent payload creation across all coroutines. Passing null clears the registry.

Parameters

callable|null $callback

Return Value

void

at line 698
mixed __call(string $method, array $parameters)

Dynamically pass calls to the default connection.

Parameters

string $method
array $parameters

Return Value

mixed