BeanstalkdQueue
class BeanstalkdQueue extends Queue implements Queue
Traits
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.
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.
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 bulk payloads with the same hook destination as individual pushes.
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.
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 Beanstalkd 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 tube statistics, treating a tube that has not been created as empty.
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.
Delete a message from the Beanstalk queue.
Get the queue or return the default.
Get the underlying Pheanstalk instance.
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 306
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.
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.
at line 314
protected array
prepareBatchJobs(array $jobs, mixed $data, string|null $queue)
Prepare bulk payloads with the same hook destination as individual pushes.
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.
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 35
__construct(PheanstalkManagerInterface $pheanstalk, string $default, int $timeToRun, int $blockFor = 0, bool $dispatchAfterCommit = false)
Create a new Beanstalkd queue instance.
at line 48
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
protected TubeStats|null
tubeStats(UnitEnum|string|null $queue)
Get tube statistics, treating a tube that has not been created as empty.
at line 100
int
totalSize()
Get the number of jobs across every queue.
at line 112
int
totalPendingSize()
Get the number of pending jobs across every queue.
at line 120
int
totalDelayedSize()
Get the number of delayed jobs across every queue.
at line 128
int
totalReservedSize()
Get the number of reserved jobs across every queue.
at line 136
Collection
pendingJobs(UnitEnum|string|null $queue = null)
Get the pending jobs for the given queue.
at line 144
Collection
delayedJobs(UnitEnum|string|null $queue = null)
Get the delayed jobs for the given queue.
at line 152
Collection
reservedJobs(UnitEnum|string|null $queue = null)
Get the reserved jobs for the given queue.
at line 160
Collection
allPendingJobs()
Get all pending jobs across every queue.
at line 168
Collection
allDelayedJobs()
Get all delayed jobs across every queue.
at line 176
Collection
allReservedJobs()
Get all reserved jobs across every queue.
at line 184
int|null
creationTimeOfOldestPendingJob(UnitEnum|string|null $queue = null)
Get the creation timestamp of the oldest pending job, excluding delayed jobs.
at line 193
mixed
push(object|string $job, mixed $data = '', UnitEnum|string|null $queue = null)
Push a new job onto the queue.
at line 211
mixed
pushRaw(string $payload, UnitEnum|string|null $queue = null, array $options = [])
Push a raw payload onto the queue.
at line 226
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 256
Job|null
pop(UnitEnum|string|null $queue = null)
Pop the next job off of the queue.
at line 286
void
deleteMessage(UnitEnum|string $queue, int|string $id)
Delete a message from the Beanstalk queue.
at line 296
string
getQueue(UnitEnum|string|null $queue)
Get the queue or return the default.
at line 324
PheanstalkManagerInterface
getPheanstalk()
Get the underlying Pheanstalk instance.