BeanstalkdAdapter
class BeanstalkdAdapter extends QueueAdapter (View source)
Constants
| EXIT_SUCCESS |
|
| EXIT_ERROR |
|
| EXIT_MEMORY_LIMIT |
|
| private MAX_PRIORITY |
Maximum priority value for Beanstalkd |
| private QUEUE_CACHE_KEY |
Cache key for storing queue names |
| private RESERVE_TIMEOUT |
Seconds the server is allowed to hold a reserve before answering Pheanstalk 8 adds this to the socket receive timeout while it waits, so any value is safe there. Pheanstalk 5 does not, and reads with the plain receive timeout, which defaults to 10 seconds: past that the socket read expires before beanstalkd answers and an idle tube surfaces as a connection error instead of "no job". Stay below it to support both. |
Properties
| protected float | $start_time | Define the start time |
from QueueAdapter |
| protected float | $processing_timeout | Define the processing timeout |
from QueueAdapter |
| protected int | $timeout | Define the work time out |
from QueueAdapter |
| protected string | $queue | Determine the default watch name |
from QueueAdapter |
| protected int | $tries | The number of working attempts |
from QueueAdapter |
| protected int | $sleep | Define the sleep time |
from QueueAdapter |
| static protected bool | $suppressLogging | Whether to suppress logging (useful for testing) |
from QueueAdapter |
Methods
Enable or disable logging suppression
Configure the Beanstalkd queue adapter
Update the processing timeout
Run the queue worker
Get the size of the queue
Flush all tasks from the queue
Store the failed payload for later inspection
Convert priority level to Beanstalkd priority value
Details
static void
suppressLogging(bool $suppress = true)
Enable or disable logging suppression
QueueAdapter
configure(array $config)
Configure the Beanstalkd queue adapter
bool
push(QueueTask $task)
Push a task onto the queue
string
serializeProducer(QueueTask $task)
Create task serialization
QueueTask
unserializeProducer(string $task)
Create task unserialize
void
sleep(int $seconds)
Sleep the process
void
setTimeout(int $timeout)
Set worker timeout
void
updateProcessingTimeout(int|null $timeout = null)
Update the processing timeout
final void
work(int $timeout, int $memory)
Launch the worker
protected bool
supportsAsyncSignals()
Determine if "async" signals are supported.
protected void
listenForSignals()
Enable async signals for the process.
void
run(string|null $queue = null)
Run the queue worker
protected bool
timeoutReached(int $timeout)
Determine if the timeout is reached
void
kill(int $status = 0)
Kill the process.
void
setTries(int $tries)
Set task tries
int
getTries()
Get task tries
void
setSleep(int $sleep)
Set sleep time
string
getQueue(string|null $queue = null)
Get the queue or return the default.
void
setQueue(string $queue)
Set the queue name
int
size(string|null $queue = null)
Get the size of the queue
void
flush(string|null $queue = null)
Flush all tasks from the queue
protected void
logError(Throwable $exception)
Log an error
final protected string
generateId()
Generate the task id
protected void
recordFailedPayload(string $key, mixed $payload)
Store the failed payload for later inspection
Recording is best effort: the cache is not guaranteed to be configured in a worker process, and a throw here would escape the failure handler and kill the worker before the message is settled, making it redeliver.
protected void
logProcessingTask(QueueTask $task)
Log processing task
protected void
logProcessedTask(QueueTask $task)
Log processed task
protected void
logFailedTask(QueueTask|null $task, Throwable $e)
Log failed task
The task is nullable because a failure can occur before the task is resolved, typically when the payload cannot be unserialized.
int
getPriority(int $priority)
Convert priority level to Beanstalkd priority value
Priority mapping:
- 0: Highest priority (urgent)
- 1: Default priority (normal)
- 2: Default priority (normal)
- 3+: Lowest priority (bulk/background)