RedisQueue
class RedisQueue extends Queue implements Queue, ClearableQueue, IndexAwareQueue, PublishesOutboxMessages
Traits
Constants
| DEFAULT_RETRY_AFTER |
|
| DEFAULT_MIGRATION_BATCH_SIZE |
|
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 |
| protected bool | $secondaryQueueHadJob | Indicates if a secondary queue had a job available between checks of the primary queue. |
|
| protected bool|null | $isCluster | Indicates if the connection is a Redis Cluster connection. |
Methods
Determine whether the job selects outbox delivery.
Run a child attempt or handler with its own forwarding context.
Determine whether this driver needs SQL staging for the selected job.
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.
Capture the logical destination before publication-time forwarding.
Add metadata required while the original job is still available.
Capture transport options that cannot be recovered from the stored envelope.
Publish delayed Redis jobs immediately into their delayed set.
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 the payload and delay for each of the given jobs.
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 string 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.
Determine if the job should be dispatched after the open parent database transactions have committed.
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 Redis 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 unique queue names.
Get the unique queue names using an already-held raw connection.
Inspect jobs from one queue while holding one Redis connection.
Inspect jobs across every queue while holding one Redis connection.
Inspect one Redis queue using an already-held raw connection.
Inspect a formatted storage key while retaining the requested queue identity.
Get the creation timestamp of the oldest pending job, excluding delayed jobs.
Store a prepared batch and raise its queue lifecycle events.
Publish stored outbox messages with one confirmed script per destination.
Add publication-time metadata without recreating the stored job payload.
Store alternating availability and payload arguments with exact confirmation.
Push a new job onto the queue.
Push a raw payload onto the queue.
Push a new job onto the queue after a delay.
Push a raw job onto the queue after (n) seconds.
Migrate any delayed or expired jobs onto the primary queue.
Migrate the delayed jobs that are ready to the regular queue.
Retrieve the next job from the queue.
Delete a reserved job from the reserved queue and release it.
Delete all of the jobs from the queue.
Get a random ID string.
Get the queue or return the default.
Get the cluster-safe Redis key for the given queue.
Format a physical queue name as a cluster-safe Redis key.
Determine if the queue connection is a Redis Cluster connection.
Get the connection for the queue.
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.
in
StagesOutboxJobs at line 68
protected bool
shouldStageViaOutbox(object|string $job)
Determine whether this driver needs SQL staging for the selected job.
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.
at line 792
protected string|null
resolveOutboxQueue(string|null $queue)
Capture the logical destination before 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.
at line 800
protected int
availableAtForOutbox(string|null $queue, int|null $deliverAt)
Publish delayed Redis jobs immediately into their delayed set.
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.
at line 348
mixed
bulk(array $jobs, mixed $data = '', UnitEnum|string|null $queue = null)
Push an array of jobs onto the queue.
at line 396
protected array
prepareBatchJobs(array $jobs, mixed $data, string|null $queue)
Prepare the payload and delay for each of the given jobs.
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.
at line 625
protected array
createPayloadArray(array|object|string $job, string|null $queue, mixed $data = '')
Create a payload string 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.
in
Queue at line 784
protected bool
shouldDispatchAfterCommit(object|string $job)
Determine if the job should be dispatched after the open parent database transactions have committed.
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 57
__construct(Factory $redis, string $default = 'default', string|null $connection = null, int|null $retryAfter = self::DEFAULT_RETRY_AFTER, int|null $blockFor = null, bool $dispatchAfterCommit = false, int $migrationBatchSize = self::DEFAULT_MIGRATION_BATCH_SIZE)
Create a new Redis queue instance.
at line 71
int
size(UnitEnum|string|null $queue = null)
Get the size of the queue.
at line 87
int
pendingSize(UnitEnum|string|null $queue = null)
Get the number of pending jobs.
at line 95
int
delayedSize(UnitEnum|string|null $queue = null)
Get the number of delayed jobs.
at line 103
int
reservedSize(UnitEnum|string|null $queue = null)
Get the number of reserved jobs.
at line 111
int
totalSize()
Get the number of jobs across every queue.
at line 132
int
totalPendingSize()
Get the number of pending jobs across every queue.
at line 142
int
totalDelayedSize()
Get the number of delayed jobs across every queue.
at line 152
int
totalReservedSize()
Get the number of reserved jobs across every queue.
at line 164
Collection
pendingJobs(UnitEnum|string|null $queue = null)
Get the pending jobs for the given queue.
at line 174
Collection
delayedJobs(UnitEnum|string|null $queue = null)
Get the delayed jobs for the given queue.
at line 184
Collection
reservedJobs(UnitEnum|string|null $queue = null)
Get the reserved jobs for the given queue.
at line 194
Collection
allPendingJobs()
Get all pending jobs across every queue.
at line 204
Collection
allDelayedJobs()
Get all delayed jobs across every queue.
at line 214
Collection
allReservedJobs()
Get all reserved jobs across every queue.
at line 224
protected Collection
allQueueNames()
Get the unique queue names.
at line 240
protected Collection
allQueueNamesUsing(RedisConnection $connection)
Get the unique queue names using an already-held raw connection.
at line 273
protected Collection
inspectJobs(UnitEnum|string|null $queue, string $suffix = '')
Inspect jobs from one queue while holding one Redis connection.
at line 293
protected Collection
inspectAllQueues(string $suffix = '')
Inspect jobs across every queue while holding one Redis connection.
at line 308
protected Collection
inspectJobsUsing(RedisConnection $connection, string $name, string $suffix)
Inspect one Redis queue using an already-held raw connection.
at line 318
protected Collection
inspectJobsAtKey(RedisConnection $connection, string $key, string $name, string $suffix)
Inspect a formatted storage key while retaining the requested queue identity.
at line 332
int|null
creationTimeOfOldestPendingJob(UnitEnum|string|null $queue = null)
Get the creation timestamp of the oldest pending job, excluding delayed jobs.
at line 404
protected void
enqueueBatch(array $jobs, string|null $queue)
Store a prepared batch and raise its queue lifecycle events.
at line 468
OutboxBatchResult
publishOutboxMessages(array $messages)
Publish stored outbox messages with one confirmed script per destination.
at line 526
protected string
preparePayloadForOutboxPublication(string $payload, string|null $queue)
Add publication-time metadata without recreating the stored job payload.
at line 536
protected void
pushRawBatch(RedisProxy $connection, string $queueKey, array $arguments)
Store alternating availability and payload arguments with exact confirmation.
at line 552
mixed
push(object|string $job, mixed $data = '', UnitEnum|string|null $queue = null)
Push a new job onto the queue.
at line 570
mixed
pushRaw(string $payload, UnitEnum|string|null $queue = null, array $options = [])
Push a raw payload onto the queue.
at line 586
mixed
later(DateInterval|DateTimeInterface|int $delay, object|string $job, mixed $data = '', UnitEnum|string|null $queue = null)
Push a new job onto the queue after a delay.
at line 609
protected mixed
laterRaw(DateInterval|DateTimeInterface|int $delay, string $payload, string|null $queue = null)
Push a raw job onto the queue after (n) seconds.
at line 636
Job|null
pop(UnitEnum|string|null $queue = null, int $index = 0)
Pop the next job off of the queue.
at line 672
protected void
migrate(string $queue)
Migrate any delayed or expired jobs onto the primary queue.
at line 684
array
migrateExpiredJobs(string $from, string $to)
Migrate the delayed jobs that are ready to the regular queue.
at line 702
protected array
retrieveNextJob(string $queue, bool $block = true)
Retrieve the next job from the queue.
at line 731
void
deleteReserved(string $queue, RedisJob $job)
Delete a reserved job from the queue.
at line 739
void
deleteAndRelease(string $queue, RedisJob $job, DateInterval|DateTimeInterface|int $delay)
Delete a reserved job from the reserved queue and release it.
at line 756
int
clear(UnitEnum|string|null $queue = null)
Delete all of the jobs from the queue.
at line 774
protected string
getRandomId()
Get a random ID string.
at line 782
string
getQueue(UnitEnum|string|null $queue)
Get the queue or return the default.
at line 810
protected string
getQueueRedisKey(UnitEnum|string|null $queue = null)
Get the cluster-safe Redis key for the given queue.
Queue names are forwarded once before adding the storage prefix and hash tag.
at line 823
protected string
formatQueueRedisKey(string $queue)
Format a physical queue name as a cluster-safe Redis key.
Redis Cluster requires every key passed to a multi-key Lua script to live on the same hash slot. Only storage keys are hash-tagged here.
at line 833
protected bool
isClusterConnection()
Determine if the queue connection is a Redis Cluster connection.
at line 841
RedisProxy
getConnection()
Get the connection for the queue.
at line 849
Factory
getRedis()
Get the underlying Redis instance.