FailoverQueue
class FailoverQueue extends Queue implements Queue, IndexAwareQueue
Traits
Constants
| protected FAILING_QUEUES_CONTEXT_PREFIX |
Context key prefix for the queues which failed on the last action. Entries pair a weak owner with its failure list. A dead or different owner means no history, since released object IDs may be reused within a coroutine. Stored in coroutine Context instead of an instance property because this queue is cached in QueueManager::$connections. Instance state would leak across concurrent dispatches. |
Properties
| protected bool | $dispatchViaOutbox | Whether this connection stages jobs through the outbox by default. |
from StagesOutboxJobs |
| protected Container | $container | The IoC container instance. |
from Queue |
| protected string | $connectionName | The connection name for the queue. |
from Queue |
| protected array | $config | The original configuration for the queue. |
from Queue |
| protected bool | $dispatchAfterCommit | Indicates that jobs should be dispatched after the open parent database transactions have committed. |
from Queue |
| protected null|Queue): mixed): mixed | $afterCommitDispatcher | Dispatch an after-commit operation through its live queue owner. |
from Queue |
| static protected callable[] | $createPayloadCallbacks | The create payload callbacks. |
from Queue |
Methods
Determine whether the job selects outbox delivery.
Run a child attempt or handler with its own forwarding context.
Preserve failover attempts instead of capturing an undecided destination.
Select an explicit transaction for a local or failover after-commit fallback.
Stage a single job through its normal guarded enqueue lifecycle.
Stage selected jobs in owner groups and return the direct remainder.
Build a byte-preserving row from a finalized payload.
Expand the logical destination without applying publication-time forwarding.
Add metadata required while the original job is still available.
Capture transport options that cannot be recovered from the stored envelope.
Determine when a stored message may first be published.
Get the number of seconds until the given DateTime.
Get the "available at" UNIX timestamp.
If the given value is an interval, convert it to a DateTime instance.
Given a start time, format the total run time for human readability.
Get a configuration value from an attribute, falling back to a property.
Extract the value from an attribute instance.
Get an instance of the given attribute class from the target class or its parents.
Determine if a property declared on a child class overrides an inherited attribute.
Resolve the default connection name for a given queueable instance.
Resolve the default queue name for a given queueable instance.
Get the container that owns the queue routes.
Push a new job onto the queue.
Push a new job onto a specific queue after (n) seconds.
Push an array of jobs onto the queue.
Prepare payloads and delays for a set-based enqueue operation.
Run an operation with its own log context, then restore the caller.
Create a payload string from the given job and data.
Create a payload array from the given job and data.
Create a payload for an object-based queue handler.
Get the maximum number of attempts for an object-based queue handler.
Get the expiration timestamp for an object-based queue handler.
Create a typical, string based queue payload array.
Register a callback to be executed when creating job payloads.
Create the given payload using any registered payload hooks.
Set the dispatcher for after-commit queue operations.
Enqueue a job using the given callback.
Defer an enqueue operation until the applicable transaction commits.
Defer a batch enqueue operation until the applicable transaction commits.
Run deferred work with its dispatch locks and optional deferredDispatchFailed job hook.
Enqueue a job immediately using the given callback.
Prepare a finalized payload for storage without changing lifecycle event correlation.
Handle a payload after the queue has accepted it.
Notify every batch member and return the first ordinary notification failure.
Notify every batch member of a failure or cancellation, reporting listener errors so they cannot replace it.
Report a failure that must not replace the primary error, without swallowing cancellation.
Finalize an encoded payload immediately before queueing.
Partition jobs by whether they should be deferred until the active transaction commits.
Degrade selected outbox delivery to an after-commit failover attempt.
Register a transaction rollback callback that releases the job's locks.
Create a callback that releases the job's locks.
Mark a job's dispatch locks as accepted by the queue.
Raise the job queueing event.
Raise the job queueing failed event, or the canceled event when cancellation stopped the job.
Raise the job queueing canceled event.
Raise the job queued event.
Normalize an enum queue name without applying defaults or forwarding.
Create a new failover queue instance.
Get the size of the queue.
Get the number of pending jobs.
Get the number of delayed jobs.
Get the number of reserved jobs.
Get the number of jobs across every queue.
Get the number of pending jobs across every queue.
Get the number of delayed jobs across every queue.
Get the number of reserved jobs across every queue.
Get the pending jobs for the given queue.
Get the delayed jobs for the given queue.
Get the reserved jobs for the given queue.
Get all pending jobs across every queue.
Get all delayed jobs across every queue.
Get all reserved jobs across every queue.
Get the creation timestamp of the oldest pending job, excluding delayed jobs.
Push a new job onto the queue.
Push a raw payload onto the queue.
Push a new job onto the queue after (n) seconds.
Resolve forwards owned by this failover connection.
Attempt the given method on all connections.
Forward an operation to the first available child connection.
Defer the failover attempt until every applicable transaction commits.
Details
in
StagesOutboxJobs at line 32
protected bool
shouldDispatchViaOutbox(object|string $job)
Determine whether the job selects outbox delivery.
in
StagesOutboxJobs at line 44
protected mixed
withOutboxForwarding(array|null $forwarded, callable $callback)
Run a child attempt or handler with its own forwarding context.
at line 232
protected bool
shouldStageViaOutbox(object|string $job)
Preserve failover attempts instead of capturing an undecided destination.
in
StagesOutboxJobs at line 76
protected string|null
resolveOutboxCommitConnection(DatabaseTransactionsManager $transactions, object|string $job)
Select an explicit transaction for a local or failover after-commit fallback.
in
StagesOutboxJobs at line 97
protected string
enqueueViaOutbox(object|string $job, string $payload, string|null $queue, DateInterval|DateTimeInterface|int|null $delay)
Stage a single job through its normal guarded enqueue lifecycle.
in
StagesOutboxJobs at line 124
protected array
stageOutboxJobs(array $jobs, mixed $data, string|null $queue)
Stage selected jobs in owner groups and return the direct remainder.
in
StagesOutboxJobs at line 207
protected array
prepareOutboxRow(object|string $job, string $payload, string|null $queue, DateInterval|DateTimeInterface|int|null $delay)
Build a byte-preserving row from a finalized payload.
in
StagesOutboxJobs at line 245
protected string|null
resolveOutboxQueue(string|null $queue)
Expand the logical destination without applying publication-time forwarding.
in
StagesOutboxJobs at line 253
protected string
preparePayloadForOutbox(object|string $job, string $payload)
Add metadata required while the original job is still available.
in
StagesOutboxJobs at line 261
protected array
prepareOutboxOptions(object|string $job, string|null $queue, string $payload)
Capture transport options that cannot be recovered from the stored envelope.
in
StagesOutboxJobs at line 269
protected int
availableAtForOutbox(string|null $queue, int|null $deliverAt)
Determine when a stored message may first be published.
in
InteractsWithTime at line 17
protected int
secondsUntil(DateInterval|DateTimeInterface|int $delay)
Get the number of seconds until the given DateTime.
in
InteractsWithTime at line 29
protected int
availableAt(DateInterval|DateTimeInterface|int|null $delay = 0)
Get the "available at" UNIX timestamp.
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.
in
InteractsWithTime at line 63
protected int
currentTime()
Get the current system time as a UNIX timestamp.
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.
in
ReadsClassAttributes at line 17
protected mixed
getAttributeValue(object $target, string $attributeClass, string|null $property = null, mixed $default = null)
Get a configuration value from an attribute, falling back to a property.
in
ReadsClassAttributes at line 46
protected mixed
extractAttributeValue(object $instance)
Extract the value from an attribute instance.
in
ReadsClassAttributes at line 58
protected object|null
getAttributeInstance(object $target, string $attributeClass, ReflectionClass|null $declaringClass = null)
Get an instance of the given attribute class from the target class or its parents.
in
ReadsClassAttributes at line 74
protected bool
propertyOverridesAttribute(object $target, ReflectionClass $reflection, string|null $property, ReflectionClass $attributeDeclaringClass)
Determine if a property declared on a child class overrides an inherited attribute.
in
ResolvesQueueRoutes at line 19
string|null
resolveConnectionFromQueueRoute(object $queueable, UnitEnum|string|null $queue = null)
Resolve the default connection name for a given queueable instance.
in
ResolvesQueueRoutes at line 27
string|null
resolveQueueFromQueueRoute(object $queueable)
Resolve the default queue name for a given queueable instance.
in
ResolvesQueueRoutes at line 35
protected QueueRoutes
queueRoutes()
Get the queue routes manager instance.
in
ResolvesQueueRoutes at line 48
protected Container
queueRoutesContainer()
Get the container that owns the queue routes.
in
Queue at line 98
mixed
pushOn(UnitEnum|string|null $queue, object|string $job, mixed $data = '')
Push a new job onto the queue.
in
Queue at line 106
mixed
laterOn(UnitEnum|string|null $queue, DateInterval|DateTimeInterface|int $delay, object|string $job, mixed $data = '')
Push a new job onto a specific queue after (n) seconds.
in
Queue at line 114
mixed
bulk(array $jobs, mixed $data = '', UnitEnum|string|null $queue = null)
Push an array of jobs onto the queue.
in
Queue at line 137
protected array
prepareBatchJobs(array $jobs, mixed $data, string|null $queue)
Prepare payloads and delays for a set-based enqueue operation.
in
Queue at line 156
protected mixed
getJobDelay(object|string $job)
Get the delay configured on the given job.
in
Queue at line 170
protected mixed
withLogContext(Repository|null $context, Closure $callback)
Run an operation with its own log context, then restore the caller.
in
Queue at line 193
protected string
createPayload(array|object|string $job, string|null $queue, mixed $data = '', DateInterval|DateTimeInterface|int|null $delay = null)
Create a payload string from the given job and data.
in
Queue at line 226
protected array
createPayloadArray(array|object|string $job, string|null $queue, mixed $data = '')
Create a payload array from the given job and data.
in
Queue at line 238
protected array
createObjectPayload(object $job, string|null $queue)
Create a payload for an object-based queue handler.
in
Queue at line 309
protected string
getDisplayName(object $job)
Get the display name for the given job.
in
Queue at line 319
mixed
getJobTries(mixed $job)
Get the maximum number of attempts for an object-based queue handler.
in
Queue at line 333
mixed
getJobBackoff(mixed $job)
Get the backoff for an object-based queue handler.
in
Queue at line 353
mixed
getJobExpiration(mixed $job)
Get the expiration timestamp for an object-based queue handler.
in
Queue at line 369
protected bool
jobShouldBeEncrypted(object $job)
Determine if the job should be encrypted.
in
Queue at line 381
protected array
createStringPayload(array|string $job, string|null $queue, mixed $data)
Create a typical, string based queue payload array.
in
Queue at line 406
static 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.
in
Queue at line 419
protected array
withCreatePayloadHooks(string|null $queue, array $payload)
Create the given payload using any registered payload hooks.
in
Queue at line 440
Queue
setAfterCommitDispatcher(Closure|null $dispatcher)
Set the dispatcher for after-commit queue operations.
Queue connection instances are cached on QueueManager; replacing the dispatcher on a shared connection changes every subsequent after-commit enqueue through it.
in
Queue at line 456
protected mixed
enqueueUsing(object|string $job, string $payload, string|null $queue, DateInterval|DateTimeInterface|int|null $delay, callable $callback)
Enqueue a job using the given callback.
The callback receives the queue that owns the operation first so deferred pooled dispatch never retains a borrowed queue.
in
Queue at line 495
protected void
deferEnqueueAfterCommit(DatabaseTransactionsManager $transactions, object|string $job, Closure $enqueue)
Defer an enqueue operation until the applicable transaction commits.
in
Queue at line 509
protected void
deferBatchEnqueueAfterCommit(DatabaseTransactionsManager $transactions, array $jobs, Closure $enqueue, string|null $connection = null)
Defer a batch enqueue operation until the applicable transaction commits.
in
Queue at line 552
static protected mixed
runDeferredEnqueue(array $jobs, Closure $operation)
Run deferred work with its dispatch locks and optional deferredDispatchFailed job hook.
in
Queue at line 589
protected mixed
enqueueNow(object|string $job, string $payload, string|null $queue, DateInterval|DateTimeInterface|int|null $delay, callable $callback)
Enqueue a job immediately using the given callback.
in
Queue at line 631
protected string
preparePayloadForPush(object|string $job, string $payload, string|null $queue)
Prepare a finalized payload for storage without changing lifecycle event correlation.
in
Queue at line 639
protected void
handlePayloadPushed(string $payload, string|null $queue)
Handle a payload after the queue has accepted it.
in
Queue at line 656
protected Throwable|null
notifyBatchMembers(array $members, Closure ...$notifications)
Notify every batch member and return the first ordinary notification failure.
A canceled notification does not skip the remaining members; the first cancellation is rethrown once every member has been notified, and the ordinary failure it displaces is reported.
in
Queue at line 693
protected void
notifyBatchMembersOfFailure(array $members, Closure $notification)
Notify every batch member of a failure or cancellation, reporting listener errors so they cannot replace it.
in
Queue at line 703
protected void
reportSecondaryFailure(Throwable $exception)
Report a failure that must not replace the primary error, without swallowing cancellation.
in
Queue at line 730
protected string
finalizePayloadForQueueing(string|null $queue, object|string $job, string $payload, DateInterval|DateTimeInterface|int|null $delay)
Finalize an encoded payload immediately before queueing.
in
Queue at line 765
protected array
partitionJobsByAfterCommit(array $jobs, DatabaseTransactionsManager|null $transactions)
Partition jobs by whether they should be deferred until the active transaction commits.
at line 240
protected bool
shouldDispatchAfterCommit(object|string $job)
Degrade selected outbox delivery to an after-commit failover attempt.
in
Queue at line 806
protected void
addJobRollbackCallback(DatabaseTransactionsManager $transactions, object|string $job, string|null $connection = null)
Register a transaction rollback callback that releases the job's locks.
in
Queue at line 818
protected Closure|null
createJobRollbackCallback(object|string $job)
Create a callback that releases the job's locks.
in
Queue at line 831
protected void
acceptDispatchLocks(object|string $job)
Mark a job's dispatch locks as accepted by the queue.
in
Queue at line 843
protected void
raiseJobQueueingEvent(string|null $queue, object|string $job, string $payload, DateInterval|DateTimeInterface|int|null $delay)
Raise the job queueing event.
in
Queue at line 864
protected void
raiseJobQueueingFailedEvent(string|null $queue, object|string $job, string $payload, DateInterval|DateTimeInterface|int|null $delay, Throwable $exception, bool $sendAttempted = true)
Raise the job queueing failed event, or the canceled event when cancellation stopped the job.
in
Queue at line 908
protected void
raiseJobQueueingCanceledEvent(string|null $queue, object|string $job, string $payload, DateInterval|DateTimeInterface|int|null $delay, Throwable $cancellation, bool $sendAttempted, bool $queued = false, Throwable|null $exception = null)
Raise the job queueing canceled event.
The cancellation is already propagating, so a listener failure is reported instead of replacing it.
in
Queue at line 951
protected void
raiseJobQueuedEvent(string|null $queue, mixed $jobId, object|string $job, string $payload, DateInterval|DateTimeInterface|int|null $delay)
Raise the job queued event.
in
Queue at line 978
protected string|null
normalizeQueue(UnitEnum|string|null $queue)
Normalize an enum queue name without applying defaults or forwarding.
in
Queue at line 986
protected string
resolveQueue(string $queue)
Get the routed queue name for the given queue.
in
Queue at line 1002
string
getConnectionName()
Get the connection name for the queue.
in
Queue at line 1016
Queue
setConnectionName(string $name)
Set the connection name for the queue.
Queue connection instances are cached on QueueManager; runtime use races across coroutines and changes every concurrent dispatch through this connection.
in
Queue at line 1027
array
getConfig()
Get the queue configuration array.
in
Queue at line 1041
Queue
setConfig(array $config)
Set the queue configuration array.
Queue connection instances are cached on QueueManager; runtime use races across coroutines and changes every concurrent dispatch through this connection.
in
Queue at line 1053
Container
getContainer()
Get the container instance being used by the connection.
in
Queue at line 1065
Queue
setContainer(Container $container)
Set the IoC container instance.
Queue connection instances are cached on QueueManager; runtime use races across coroutines and changes every concurrent dispatch through this connection.
in
Queue at line 1076
static void
flushState()
Flush all static state.
at line 45
__construct(QueueManager $manager, Dispatcher $events, array $connections, bool $dispatchAfterCommit = false)
Create a new failover queue instance.
at line 56
int
size(UnitEnum|string|null $queue = null)
Get the size of the queue.
at line 64
int
pendingSize(UnitEnum|string|null $queue = null)
Get the number of pending jobs.
at line 72
int
delayedSize(UnitEnum|string|null $queue = null)
Get the number of delayed jobs.
at line 80
int
reservedSize(UnitEnum|string|null $queue = null)
Get the number of reserved jobs.
at line 88
int
totalSize()
Get the number of jobs across every queue.
at line 96
int
totalPendingSize()
Get the number of pending jobs across every queue.
at line 104
int
totalDelayedSize()
Get the number of delayed jobs across every queue.
at line 112
int
totalReservedSize()
Get the number of reserved jobs across every queue.
at line 120
Collection
pendingJobs(UnitEnum|string|null $queue = null)
Get the pending jobs for the given queue.
at line 128
Collection
delayedJobs(UnitEnum|string|null $queue = null)
Get the delayed jobs for the given queue.
at line 136
Collection
reservedJobs(UnitEnum|string|null $queue = null)
Get the reserved jobs for the given queue.
at line 144
Collection
allPendingJobs()
Get all pending jobs across every queue.
at line 152
Collection
allDelayedJobs()
Get all delayed jobs across every queue.
at line 160
Collection
allReservedJobs()
Get all reserved jobs across every queue.
at line 168
int|null
creationTimeOfOldestPendingJob(UnitEnum|string|null $queue = null)
Get the creation timestamp of the oldest pending job, excluding delayed jobs.
at line 178
mixed
push(object|string $job, mixed $data = '', UnitEnum|string|null $queue = null)
Push a new job onto the queue.
at line 188
mixed
pushRaw(string $payload, UnitEnum|string|null $queue = null, array $options = [])
Push a raw payload onto the queue.
at line 198
mixed
later(DateInterval|DateTimeInterface|int $delay, object|string $job, mixed $data = '', UnitEnum|string|null $queue = null)
Push a new job onto the queue after (n) seconds.
at line 208
Job|null
pop(UnitEnum|string|null $queue = null, int $index = 0)
Pop the next job off of the queue.
at line 221
protected string|null
resolveForwardedQueue(UnitEnum|string|null $queue)
Resolve forwards owned by this failover connection.
at line 250
protected mixed
attemptOnAllConnections(string $method, array $arguments, object|string|null $job = null)
Attempt the given method on all connections.
at line 271
protected mixed
attemptOnConnections(string $method, array $arguments, object|string|null $job, bool $afterCommitHandled = false)
Forward an operation to the first available child connection.
at line 328
protected bool
deferUntilAllTransactionsCommit(DatabaseTransactionsManager $transactions, string $method, array $arguments, object|string $job)
Defer the failover attempt until every applicable transaction commits.