---
title: "Phalcon Queue"
version: "5.19"
---
> Documentation Index
> Fetch the complete documentation index at: https://docs.phalcon.io/llms.txt
> Use this file to discover all available pages before exploring further.
# Phalcon Queue
:::info[NOTE]
All classes are prefixed with `Phalcon`
:::
## Queue\AdapterFactory
Class
Source on GitHub
Maps an adapter name to its ConnectionFactory. Mirrors
Phalcon\Storage\AdapterFactory.
- [`Phalcon\Factory\AbstractConfigFactory`](/5.19/api/phalcon_factory/#factoryabstractconfigfactory)
- [`Phalcon\Factory\AbstractFactory`](/5.19/api/phalcon_factory/#factoryabstractfactory)
- **`Phalcon\Queue\AdapterFactory`**
__Uses__ `Phalcon\Config\ConfigInterface` · `Phalcon\Contracts\Queue\ConnectionFactory` · `Phalcon\Contracts\Queue\QueueTypes` · `Phalcon\Factory\AbstractFactory` · `Phalcon\Queue\Adapter\Beanstalk\BeanstalkConnectionFactory` · `Phalcon\Queue\Adapter\Memory\MemoryConnectionFactory` · `Phalcon\Queue\Adapter\Redis\RedisConnectionFactory` · `Phalcon\Queue\Adapter\Stream\StreamConnectionFactory` · `Phalcon\Queue\Exceptions\Exception`
### Method Summary
### Methods
Public · 2
__construct()
```php
public function __construct( array $services = [] );
```
AdapterFactory constructor.
newInstance()
```php
public function newInstance(
string $name,
array $options = []
): ConnectionFactoryInterface;
```
Creates a new ConnectionFactory for the named adapter.
Protected · 2
getExceptionClass()
```php
protected function getExceptionClass(): string;
```
Returns the exception class for the factory
getServices()
```php
protected function getServices(): array;
```
Returns the available adapters.
## Queue\Adapter\AbstractConsumer
Abstract
Source on GitHub
Shared consumer base. Implements the blocking `receive()` as a polling loop
on top of the abstract `receiveNoWait()`; concrete consumers provide the
transport-specific `receiveNoWait`, `acknowledge`, `reject` and `getQueue`.
Transports with a native blocking receive (Redis BRPOP, Beanstalk reserve)
override `receive()` instead of polling.
- **`Phalcon\Queue\Adapter\AbstractConsumer`** - implements [`Phalcon\Contracts\Queue\Consumer`](/5.19/api/phalcon_contracts/#contractsqueueconsumer)
- [`Phalcon\Queue\Adapter\Beanstalk\BeanstalkConsumer`](#queueadapterbeanstalkbeanstalkconsumer)
- [`Phalcon\Queue\Adapter\Memory\MemoryConsumer`](#queueadaptermemorymemoryconsumer)
- [`Phalcon\Queue\Adapter\Redis\RedisConsumer`](#queueadapterredisredisconsumer)
- [`Phalcon\Queue\Adapter\Stream\StreamConsumer`](#queueadapterstreamstreamconsumer)
__Uses__ `Phalcon\Contracts\Queue\Consumer` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Queue`
### Method Summary
### Properties
protected
int
$pollInterval = 200
Milliseconds slept between poll attempts.
protected
QueueInterface
$queue
The queue this consumer reads from.
### Methods
Public · 6
acknowledge()
```php
abstract public function acknowledge( MessageInterface $message ): void;
```
Acknowledges the message; the transport may then discard it.
getQueue()
```php
public function getQueue(): QueueInterface;
```
Returns the queue this consumer reads from.
receive()
```php
public function receive( int $timeout = 0 ): MessageInterface|null;
```
Receives a message, blocking up to timeout milliseconds (0 = block
until one is available), by polling `receiveNoWait()` every
`pollInterval` milliseconds. Returns null when none arrives in time.
receiveNoWait()
```php
abstract public function receiveNoWait(): MessageInterface|null;
```
Receives a message without blocking, or null when none is ready.
reject()
```php
abstract public function reject(
MessageInterface $message,
bool $requeue = false
): void;
```
Rejects the message. When requeue is true the transport redelivers it.
setPollInterval()
```php
public function setPollInterval( int $pollInterval ): void;
```
Sets the poll interval (in milliseconds) used by `receive()`.
## Queue\Adapter\AbstractContext
Abstract
Source on GitHub
Shared transport-session base. Every transport builds the same destination
value objects (GenericQueue / GenericTopic) and the same uniquely named
temporary queue, so those factories live here once. Concrete contexts
implement the transport-specific factories (consumer, producer, message,
subscription consumer) and the storage operations.
- **`Phalcon\Queue\Adapter\AbstractContext`** - implements [`Phalcon\Contracts\Queue\Context`](/5.19/api/phalcon_contracts/#contractsqueuecontext)
- [`Phalcon\Queue\Adapter\Beanstalk\BeanstalkContext`](#queueadapterbeanstalkbeanstalkcontext)
- [`Phalcon\Queue\Adapter\Memory\MemoryContext`](#queueadaptermemorymemorycontext)
- [`Phalcon\Queue\Adapter\Redis\RedisContext`](#queueadapterredisrediscontext)
- [`Phalcon\Queue\Adapter\Stream\StreamContext`](#queueadapterstreamstreamcontext)
__Uses__ `Phalcon\Contracts\Queue\Context` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Contracts\Queue\Topic`
### Method Summary
### Methods
Public · 3
createQueue()
```php
public function createQueue( string $queueName ): QueueInterface;
```
Creates a queue destination by name.
createTemporaryQueue()
```php
public function createTemporaryQueue(): QueueInterface;
```
Creates a uniquely named temporary queue.
createTopic()
```php
public function createTopic( string $topicName ): TopicInterface;
```
Creates a topic destination by name.
## Queue\Adapter\AbstractMessage
Abstract
Source on GitHub
Shared base for the concrete adapter messages.
@todo Remove in v7. Kept only for backwards compatibility; compose
Phalcon\Queue\Adapter\Traits\MessageTrait directly instead of extending this.
- **`Phalcon\Queue\Adapter\AbstractMessage`** - implements [`Phalcon\Contracts\Queue\Message`](/5.19/api/phalcon_contracts/#contractsqueuemessage)
- [`Phalcon\Queue\Adapter\Beanstalk\BeanstalkMessage`](#queueadapterbeanstalkbeanstalkmessage)
- [`Phalcon\Queue\Adapter\Memory\MemoryMessage`](#queueadaptermemorymemorymessage)
- [`Phalcon\Queue\Adapter\Redis\RedisMessage`](#queueadapterredisredismessage)
- [`Phalcon\Queue\Adapter\Stream\StreamMessage`](#queueadapterstreamstreammessage)
__Uses__ `Phalcon\Contracts\Queue\Message` · `Phalcon\Queue\Adapter\Traits\MessageTrait`
## Queue\Adapter\AbstractProducer
Abstract
Source on GitHub
Shared producer base. Defaults every optional capability (delivery delay,
priority, time to live) to "unsupported": the getter returns null and the
setter throws the matching exception for any non-null value. A concrete
producer overrides only the capabilities its transport actually supports,
and implements `send()`.
- **`Phalcon\Queue\Adapter\AbstractProducer`** - implements [`Phalcon\Contracts\Queue\Producer`](/5.19/api/phalcon_contracts/#contractsqueueproducer)
- [`Phalcon\Queue\Adapter\Beanstalk\BeanstalkProducer`](#queueadapterbeanstalkbeanstalkproducer)
- [`Phalcon\Queue\Adapter\Memory\MemoryProducer`](#queueadaptermemorymemoryproducer)
- [`Phalcon\Queue\Adapter\Redis\RedisProducer`](#queueadapterredisredisproducer)
- [`Phalcon\Queue\Adapter\Stream\StreamProducer`](#queueadapterstreamstreamproducer)
__Uses__ `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Producer` · `Phalcon\Queue\Exceptions\DeliveryDelayNotSupportedException` · `Phalcon\Queue\Exceptions\PriorityNotSupportedException` · `Phalcon\Queue\Exceptions\TimeToLiveNotSupportedException`
### Method Summary
### Methods
Public · 7
getDeliveryDelay()
```php
public function getDeliveryDelay(): int|null;
```
getPriority()
```php
public function getPriority(): int|null;
```
getTimeToLive()
```php
public function getTimeToLive(): int|null;
```
send()
```php
abstract public function send(
DestinationInterface $destination,
MessageInterface $message
): void;
```
setDeliveryDelay()
```php
public function setDeliveryDelay( mixed $deliveryDelay = null ): ProducerInterface;
```
setPriority()
```php
public function setPriority( mixed $priority = null ): ProducerInterface;
```
setTimeToLive()
```php
public function setTimeToLive( mixed $timeToLive = null ): ProducerInterface;
```
## Queue\Adapter\AbstractSubscriptionConsumer
Abstract
Source on GitHub
Shared subscription-consumer base.
@todo Remove in v7. Kept only for backwards compatibility; compose
Phalcon\Queue\Adapter\Traits\SubscriptionConsumerTrait directly instead of
extending this.
- **`Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`** - implements [`Phalcon\Contracts\Queue\SubscriptionConsumer`](/5.19/api/phalcon_contracts/#contractsqueuesubscriptionconsumer)
- [`Phalcon\Queue\Adapter\Beanstalk\BeanstalkSubscriptionConsumer`](#queueadapterbeanstalkbeanstalksubscriptionconsumer)
- [`Phalcon\Queue\Adapter\Memory\MemorySubscriptionConsumer`](#queueadaptermemorymemorysubscriptionconsumer)
- [`Phalcon\Queue\Adapter\Redis\RedisSubscriptionConsumer`](#queueadapterredisredissubscriptionconsumer)
- [`Phalcon\Queue\Adapter\Stream\StreamSubscriptionConsumer`](#queueadapterstreamstreamsubscriptionconsumer)
__Uses__ `Phalcon\Contracts\Queue\SubscriptionConsumer` · `Phalcon\Queue\Adapter\Traits\SubscriptionConsumerTrait`
## Queue\Adapter\Beanstalk\BeanstalkConnection
Class
Source on GitHub
Dependency-free socket client for the Beanstalkd work queue, implementing
the subset of the 1.2 protocol the adapter needs (use/watch/ignore, put,
reserve-with-timeout, delete/release/bury/touch). Recovered and trimmed
from the original Phalcon\Queue\Beanstalk transport.
- **`Phalcon\Queue\Adapter\Beanstalk\BeanstalkConnection`**
__Uses__ `Phalcon\Contracts\Queue\QueueTypes` · `Phalcon\Queue\Exceptions\Exception` · `Phalcon\Traits\Php\FileTrait`
### Method Summary
public
__construct(string $host = "127.0.0.1",int $port = 11300,bool $persistent = false)
public
bool
buryJob(string $id,int $priority)
Puts a reserved job into the "buried" state.
public
resource
connect()
Opens the socket connection to the Beanstalkd server.
public
bool
deleteJob( string $id )
Removes a job from the server entirely.
public
bool
disconnect()
Closes the connection to the server.
public
bool
ignoreTube( string $tube )
Removes the named tube from the watch list for the connection.
public
false|int
put(string $data,int $priority,int $delay,int $ttr)
Puts a job on the queue using the currently used tube. Returns the new
public
false|string
read( int $length = 0 )
Reads a packet from the socket. Verifies the connection is available
public
array
readStatus()
Reads the latest status line and splits it into tokens.
public
bool
releaseJob(string $id,int $priority,int $delay)
Puts a reserved job back into the ready queue.
public
array|null
reserve( mixed $timeout = null )
Reserves a ready job from a watched tube. A null timeout blocks until a
public
array|false
statsTube( string $tube )
Returns the Beanstalkd statistics for a tube as an associative array, or
public
bool
touchJob( string $id )
Extends the time-to-run of a reserved job.
public
bool
useTube( string $tube )
Changes the tube new jobs are put on. By default this is "default".
public
bool
watchTube( string $tube )
Adds the named tube to the watch list for the connection.
public
false|int
write( string $data )
Writes data to the socket, connecting first when needed.
### Properties
protected
resource|null
$connection = null
Connection resource.
protected
string
$host = "127.0.0.1"
protected
bool
$persistent = false
protected
int
$port = 11300
protected
string
$usedTube = "default"
Tube currently selected with use. A fresh connection uses "default".
protected
array<string, bool>
$watchedTubes = []
Tubes currently on the watch list, keyed by tube name. A fresh connection watches "default".
### Methods
Public · 16
__construct()
```php
public function __construct(
string $host = "127.0.0.1",
int $port = 11300,
bool $persistent = false
);
```
buryJob()
```php
public function buryJob(
string $id,
int $priority
): bool;
```
Puts a reserved job into the "buried" state.
connect()
```php
public function connect(): resource;
```
Opens the socket connection to the Beanstalkd server.
deleteJob()
```php
public function deleteJob( string $id ): bool;
```
Removes a job from the server entirely.
disconnect()
```php
public function disconnect(): bool;
```
Closes the connection to the server.
ignoreTube()
```php
public function ignoreTube( string $tube ): bool;
```
Removes the named tube from the watch list for the connection.
put()
```php
public function put(
string $data,
int $priority,
int $delay,
int $ttr
): false|int;
```
Puts a job on the queue using the currently used tube. Returns the new
job id, or false when the server did not accept it.
read()
```php
public function read( int $length = 0 ): false|string;
```
Reads a packet from the socket. Verifies the connection is available
first.
readStatus()
```php
public function readStatus(): array;
```
Reads the latest status line and splits it into tokens.
releaseJob()
```php
public function releaseJob(
string $id,
int $priority,
int $delay
): bool;
```
Puts a reserved job back into the ready queue.
reserve()
```php
public function reserve( mixed $timeout = null ): array|null;
```
Reserves a ready job from a watched tube. A null timeout blocks until a
job is available; otherwise it blocks up to timeout seconds. Returns
[id, body] or null when none is reserved.
statsTube()
```php
public function statsTube( string $tube ): array|false;
```
Returns the Beanstalkd statistics for a tube as an associative array, or
false when the tube does not exist.
touchJob()
```php
public function touchJob( string $id ): bool;
```
Extends the time-to-run of a reserved job.
useTube()
```php
public function useTube( string $tube ): bool;
```
Changes the tube new jobs are put on. By default this is "default".
watchTube()
```php
public function watchTube( string $tube ): bool;
```
Adds the named tube to the watch list for the connection.
write()
```php
public function write( string $data ): false|int;
```
Writes data to the socket, connecting first when needed.
## Queue\Adapter\Beanstalk\BeanstalkConnectionFactory
Class
Source on GitHub
Builds a BeanstalkContext.
Options:
- host: server host (default 127.0.0.1).
- port: server port (default 11300).
- persistent: use a persistent socket (default false).
- ttr: default time-to-run in seconds for every job (default 86400).
- pollInterval: milliseconds between subscription poll passes (default 200).
- **`Phalcon\Queue\Adapter\Beanstalk\BeanstalkConnectionFactory`** - implements [`Phalcon\Contracts\Queue\ConnectionFactory`](/5.19/api/phalcon_contracts/#contractsqueueconnectionfactory)
__Uses__ `Phalcon\Contracts\Queue\ConnectionFactory` · `Phalcon\Contracts\Queue\Context` · `Phalcon\Contracts\Queue\QueueTypes`
### Method Summary
### Properties
protected
array
$options = []
### Methods
Public · 2
__construct()
```php
public function __construct( array $options = [] );
```
createContext()
```php
public function createContext(): ContextInterface;
```
## Queue\Adapter\Beanstalk\BeanstalkConsumer
Class
Source on GitHub
Receives messages from a single Beanstalkd tube over its own connection.
`receive()` is overridden to use the native blocking reserve. Implements
VisibilityAware: a reserved job has a time-to-run window that `touch()`
extends; acknowledging deletes the job, rejecting releases it (requeue) or
buries it.
- [`Phalcon\Queue\Adapter\AbstractConsumer`](#queueadapterabstractconsumer)
- **`Phalcon\Queue\Adapter\Beanstalk\BeanstalkConsumer`** - implements [`Phalcon\Contracts\Queue\VisibilityAware`](/5.19/api/phalcon_contracts/#contractsqueuevisibilityaware)
__Uses__ `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Contracts\Queue\VisibilityAware` · `Phalcon\Queue\Adapter\AbstractConsumer` · `Phalcon\Queue\Adapter\MessageEnvelope`
### Method Summary
### Constants
int
DEFAULT_PRIORITY = 100
Default Beanstalkd priority used when releasing or burying.
### Properties
protected
BeanstalkConnection
$connection
### Methods
Public · 6
__construct()
```php
public function __construct(
BeanstalkConnection $connection,
QueueInterface $queue
);
```
acknowledge()
```php
public function acknowledge( MessageInterface $message ): void;
```
receive()
```php
public function receive( int $timeout = 0 ): MessageInterface|null;
```
receiveNoWait()
```php
public function receiveNoWait(): MessageInterface|null;
```
reject()
```php
public function reject(
MessageInterface $message,
bool $requeue = false
): void;
```
touch()
```php
public function touch( MessageInterface $message ): bool;
```
Extends the time-to-run window of a reserved job (VisibilityAware).
## Queue\Adapter\Beanstalk\BeanstalkContext
Class
Source on GitHub
Beanstalkd transport session. A queue maps to a Beanstalkd tube. Producers
share the context connection (`use` + `put`); each consumer owns its own
connection, because Beanstalkd only lets the reserving connection delete,
release, bury or touch a job. The destination factories come from
AbstractContext.
- [`Phalcon\Queue\Adapter\AbstractContext`](#queueadapterabstractcontext)
- **`Phalcon\Queue\Adapter\Beanstalk\BeanstalkContext`** - implements [`Phalcon\Contracts\Queue\Inspectable`](/5.19/api/phalcon_contracts/#contractsqueueinspectable)
__Uses__ `Phalcon\Contracts\Queue\Consumer` · `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Inspectable` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Producer` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Contracts\Queue\SubscriptionConsumer` · `Phalcon\Queue\Adapter\AbstractContext` · `Phalcon\Queue\Adapter\QueueDestinationGuard`
### Method Summary
public
__construct(string $host,int $port,bool $persistent = false,int $ttr = 86400,int $pollInterval = 200)
public
void
close()
public
ConsumerInterface
createConsumer( DestinationInterface $destination )
public
MessageInterface
createMessage(string $body = "",array $properties = [],array $headers = [])
public
ProducerInterface
createProducer()
public
SubscriptionConsumerInterface
createSubscriptionConsumer()
public
array
getStats( QueueInterface $queue )
Returns the Beanstalkd stats-tube fields for the queue's tube as an
public
int
getTtr()
Default time-to-run (seconds) for new jobs. Used by BeanstalkProducer.
public
void
purgeQueue( QueueInterface $queue )
public
void
putMessage(string $tube,string $payload,int $priority,int $delay,int $ttr)
Puts a serialized payload on a tube via the shared connection.
### Properties
protected
BeanstalkConnection|null
$connection = null
Shared connection used by producers and purges.
protected
string
$host = "127.0.0.1"
protected
bool
$persistent = false
protected
int
$pollInterval = 200
protected
int
$port = 11300
protected
int
$ttr = 86400
### Methods
Public · 10
__construct()
```php
public function __construct(
string $host,
int $port,
bool $persistent = false,
int $ttr = 86400,
int $pollInterval = 200
);
```
close()
```php
public function close(): void;
```
createConsumer()
```php
public function createConsumer( DestinationInterface $destination ): ConsumerInterface;
```
createMessage()
```php
public function createMessage(
string $body = "",
array $properties = [],
array $headers = []
): MessageInterface;
```
createProducer()
```php
public function createProducer(): ProducerInterface;
```
createSubscriptionConsumer()
```php
public function createSubscriptionConsumer(): SubscriptionConsumerInterface;
```
getStats()
```php
public function getStats( QueueInterface $queue ): array;
```
Returns the Beanstalkd `stats-tube` fields for the queue's tube as an
associative array, with numeric values cast to int (the `name` field is
kept as a string). When the tube exists the result is the full Beanstalkd
stats-tube field set (current-jobs-*, total-jobs, the `cmd-*` counters and
tube-configuration fields).
The `current-jobs-*` backlog keys are always present: an unknown tube
(no jobs, not used or watched) has zero backlog, so those keys are
returned at zero. This keeps the backlog shape independent of transient
watcher state. Runs on a fresh short-lived connection (like purgeQueue)
so the read never shares the producer's socket.
getTtr()
```php
public function getTtr(): int;
```
Default time-to-run (seconds) for new jobs. Used by BeanstalkProducer.
purgeQueue()
```php
public function purgeQueue( QueueInterface $queue ): void;
```
putMessage()
```php
public function putMessage(
string $tube,
string $payload,
int $priority,
int $delay,
int $ttr
): void;
```
Puts a serialized payload on a tube via the shared connection.
Internal transport API used by BeanstalkProducer.
## Queue\Adapter\Beanstalk\BeanstalkMessage
Class
Source on GitHub
Beanstalkd-backed message. Carries the reserved job id so the consumer can
delete, release, bury or touch it; all other behavior comes from
MessageTrait.
- [`Phalcon\Queue\Adapter\AbstractMessage`](#queueadapterabstractmessage)
- **`Phalcon\Queue\Adapter\Beanstalk\BeanstalkMessage`**
__Uses__ `Phalcon\Queue\Adapter\AbstractMessage`
### Method Summary
### Properties
protected
string|null
$jobId = null
The reserved Beanstalkd job id, or null before it is reserved.
### Methods
Public · 2
getJobId()
```php
public function getJobId(): string|null;
```
setJobId()
```php
public function setJobId( string $jobId ): void;
```
## Queue\Adapter\Beanstalk\BeanstalkProducer
Class
Source on GitHub
Sends messages to a Beanstalkd tube. Delivery delay (rounded down to whole
seconds) and message priority are supported natively; Beanstalkd has no
message expiry, so time to live is not (the default from AbstractProducer
rejects it).
- [`Phalcon\Queue\Adapter\AbstractProducer`](#queueadapterabstractproducer)
- **`Phalcon\Queue\Adapter\Beanstalk\BeanstalkProducer`**
__Uses__ `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Producer` · `Phalcon\Queue\Adapter\AbstractProducer` · `Phalcon\Queue\Adapter\MessageEnvelope` · `Phalcon\Queue\Adapter\QueueDestinationGuard`
### Method Summary
### Constants
int
DEFAULT_PRIORITY = 100
Default Beanstalkd priority (0 = most urgent).
### Properties
protected
BeanstalkContext
$context
protected
int|null
$deliveryDelay = null
Delivery delay in milliseconds, or null when not set.
protected
int|null
$priority = null
Job priority, or null when not set.
### Methods
Public · 6
__construct()
```php
public function __construct( BeanstalkContext $context );
```
getDeliveryDelay()
```php
public function getDeliveryDelay(): int|null;
```
getPriority()
```php
public function getPriority(): int|null;
```
send()
```php
public function send(
DestinationInterface $destination,
MessageInterface $message
): void;
```
setDeliveryDelay()
```php
public function setDeliveryDelay( mixed $deliveryDelay = null ): ProducerInterface;
```
setPriority()
```php
public function setPriority( mixed $priority = null ): ProducerInterface;
```
## Queue\Adapter\Beanstalk\BeanstalkSubscriptionConsumer
Class
Source on GitHub
Consumes from several Beanstalkd tubes at once. The round-robin poll loop
lives in SubscriptionConsumerTrait.
- [`Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`](#queueadapterabstractsubscriptionconsumer)
- **`Phalcon\Queue\Adapter\Beanstalk\BeanstalkSubscriptionConsumer`**
__Uses__ `Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`
### Method Summary
### Properties
protected
BeanstalkContext
$context
### Methods
Public · 1
__construct()
```php
public function __construct(
BeanstalkContext $context,
int $pollInterval = 200
);
```
The context is retained for transports that may later need it for a
native multi-queue receive; the shared poll loop does not use it.
## Queue\Adapter\GenericQueue
Class
Source on GitHub
A named queue destination shared by every transport. A queue name is the
only knowledge a destination carries, so the adapters need no transport
specific subclass.
- **`Phalcon\Queue\Adapter\GenericQueue`** - implements [`Phalcon\Contracts\Queue\Queue`](/5.19/api/phalcon_contracts/#contractsqueuequeue)
__Uses__ `Phalcon\Contracts\Queue\Queue`
### Method Summary
### Properties
protected
string
$queueName = ""
### Methods
Public · 2
__construct()
```php
public function __construct( string $queueName );
```
GenericQueue constructor.
getQueueName()
```php
public function getQueueName(): string;
```
Returns the queue name.
## Queue\Adapter\GenericTopic
Class
Source on GitHub
A named topic destination shared by every transport. A topic name is the
only knowledge a destination carries, so the adapters need no transport
specific subclass.
- **`Phalcon\Queue\Adapter\GenericTopic`** - implements [`Phalcon\Contracts\Queue\Topic`](/5.19/api/phalcon_contracts/#contractsqueuetopic)
__Uses__ `Phalcon\Contracts\Queue\Topic`
### Method Summary
### Properties
protected
string
$topicName = ""
### Methods
Public · 2
__construct()
```php
public function __construct( string $topicName );
```
GenericTopic constructor.
getTopicName()
```php
public function getTopicName(): string;
```
Returns the topic name.
## Queue\Adapter\Memory\MemoryConnectionFactory
Class
Source on GitHub
Builds a MemoryContext. The Memory transport takes no options.
- **`Phalcon\Queue\Adapter\Memory\MemoryConnectionFactory`** - implements [`Phalcon\Contracts\Queue\ConnectionFactory`](/5.19/api/phalcon_contracts/#contractsqueueconnectionfactory)
__Uses__ `Phalcon\Contracts\Queue\ConnectionFactory` · `Phalcon\Contracts\Queue\Context` · `Phalcon\Contracts\Queue\QueueTypes`
### Method Summary
### Properties
protected
array
$options = []
### Methods
Public · 2
__construct()
```php
public function __construct( array $options = [] );
```
MemoryConnectionFactory constructor.
createContext()
```php
public function createContext(): ContextInterface;
```
Creates a new in-process context.
## Queue\Adapter\Memory\MemoryConsumer
Class
Source on GitHub
Receives messages from a single in-process queue. `receive()` is the
polling loop inherited from AbstractConsumer.
- [`Phalcon\Queue\Adapter\AbstractConsumer`](#queueadapterabstractconsumer)
- **`Phalcon\Queue\Adapter\Memory\MemoryConsumer`**
__Uses__ `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Queue\Adapter\AbstractConsumer`
### Method Summary
### Properties
protected
MemoryContext
$context
### Methods
Public · 4
__construct()
```php
public function __construct(
MemoryContext $context,
QueueInterface $queue
);
```
MemoryConsumer constructor.
acknowledge()
```php
public function acknowledge( MessageInterface $message ): void;
```
No-op: a received message has already been removed from the queue.
receiveNoWait()
```php
public function receiveNoWait(): MessageInterface|null;
```
Removes and returns the next message, or null when the queue is empty.
reject()
```php
public function reject(
MessageInterface $message,
bool $requeue = false
): void;
```
Rejects the message. When requeue is true it is put back on the queue.
## Queue\Adapter\Memory\MemoryContext
Class
Source on GitHub
In-process transport session. Owns the named FIFO queues that this context's
producers and consumers share. The destination factories (createQueue /
createTopic / createTemporaryQueue) come from AbstractContext.
- [`Phalcon\Queue\Adapter\AbstractContext`](#queueadapterabstractcontext)
- **`Phalcon\Queue\Adapter\Memory\MemoryContext`**
__Uses__ `Phalcon\Contracts\Queue\Consumer` · `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Producer` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Contracts\Queue\SubscriptionConsumer` · `Phalcon\Queue\Adapter\AbstractContext` · `Phalcon\Queue\Adapter\QueueDestinationGuard`
### Method Summary
### Properties
protected
array<string, list<MessageInterface>>
$queues = []
Named queues: queue name => list of messages (FIFO).
### Methods
Public · 8
close()
```php
public function close(): void;
```
Closes the context and drops every stored message.
createConsumer()
```php
public function createConsumer( DestinationInterface $destination ): ConsumerInterface;
```
Creates a consumer for the given queue destination.
createMessage()
```php
public function createMessage(
string $body = "",
array $properties = [],
array $headers = []
): MessageInterface;
```
Creates a message.
createProducer()
```php
public function createProducer(): ProducerInterface;
```
Creates a producer.
createSubscriptionConsumer()
```php
public function createSubscriptionConsumer(): SubscriptionConsumerInterface;
```
Creates a subscription consumer.
popMessage()
```php
public function popMessage( string $queueName ): MessageInterface|null;
```
Removes the front message from a queue, or null when it is empty.
Internal transport API used by MemoryConsumer.
purgeQueue()
```php
public function purgeQueue( QueueInterface $queue ): void;
```
Removes all messages from the given queue.
pushMessage()
```php
public function pushMessage(
string $queueName,
MessageInterface $message
): void;
```
Appends a message to the back of a queue.
Internal transport API used by MemoryProducer.
## Queue\Adapter\Memory\MemoryMessage
Class
Source on GitHub
In-process message. All behavior comes from MessageTrait.
- [`Phalcon\Queue\Adapter\AbstractMessage`](#queueadapterabstractmessage)
- **`Phalcon\Queue\Adapter\Memory\MemoryMessage`**
__Uses__ `Phalcon\Queue\Adapter\AbstractMessage`
## Queue\Adapter\Memory\MemoryProducer
Class
Source on GitHub
Sends messages into an in-process queue. The Memory transport delivers
immediately and in-process, so delivery delay, priority and time to live are
not supported (the defaults from AbstractProducer reject them).
- [`Phalcon\Queue\Adapter\AbstractProducer`](#queueadapterabstractproducer)
- **`Phalcon\Queue\Adapter\Memory\MemoryProducer`**
__Uses__ `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Queue\Adapter\AbstractProducer` · `Phalcon\Queue\Adapter\QueueDestinationGuard`
### Method Summary
### Properties
protected
MemoryContext
$context
### Methods
Public · 2
__construct()
```php
public function __construct( MemoryContext $context );
```
send()
```php
public function send(
DestinationInterface $destination,
MessageInterface $message
): void;
```
## Queue\Adapter\Memory\MemorySubscriptionConsumer
Class
Source on GitHub
Consumes from several in-process queues at once. The round-robin poll loop
lives in SubscriptionConsumerTrait.
- [`Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`](#queueadapterabstractsubscriptionconsumer)
- **`Phalcon\Queue\Adapter\Memory\MemorySubscriptionConsumer`**
__Uses__ `Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`
### Method Summary
### Properties
protected
MemoryContext
$context
### Methods
Public · 1
__construct()
```php
public function __construct( MemoryContext $context );
```
The context is retained for transports that may later need it for a
native multi-queue receive; the shared poll loop does not use it.
## Queue\Adapter\MessageEnvelope
Class
Source on GitHub
Encodes and decodes the \{body, properties, headers\} envelope shared by every
transport that persists a message as a serialized string (Stream, Redis,
Beanstalk). Centralizes the wire shape, the object-injection-safe
`allowed_classes => false` guard, and the missing-key defaults, so each
adapter only supplies its own concrete message factory around `decode()`.
- **`Phalcon\Queue\Adapter\MessageEnvelope`**
__Uses__ `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\QueueTypes`
### Method Summary
### Methods
Public · 2
decode()
```php
public static function decode( string $payload ): array|null;
```
Decodes a serialized payload into a normalized \{body, properties,
headers\} array, or null when the payload is not a valid envelope.
encode()
```php
public static function encode( MessageInterface $message ): string;
```
Serializes a message into its wire envelope.
## Queue\Adapter\QueueDestinationGuard
Class
Source on GitHub
Shared "destination must be a queue" guard. Producers (on send) and contexts
(on createConsumer) both reject any non-queue destination with the same typed
exception; this keeps that single rule in one place. The `action` verb
("send to", "consume from") tailors the message to the caller.
- **`Phalcon\Queue\Adapter\QueueDestinationGuard`**
__Uses__ `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Queue\Exceptions\InvalidDestinationException`
### Method Summary
### Methods
Public · 1
assertQueue()
```php
public static function assertQueue(
DestinationInterface $destination,
string $action
): void;
```
Throws InvalidDestinationException unless the destination is a queue.
## Queue\Adapter\Redis\RedisConnectionFactory
Class
Source on GitHub
Connects to a Redis server (ext-redis) and builds a RedisContext. The
connection (connect/pconnect, auth, database select) is delegated to
Phalcon\Storage\Adapter\Redis so the queue reuses the framework's hardened
connection handling instead of re-implementing it.
Options:
- host: server host (default 127.0.0.1).
- port: server port (default 6379).
- timeout: connection timeout in seconds (default 0).
- persistent: use a persistent connection (default false).
- persistentId: identifier for the persistent connection.
- auth: password, or [user, password] for ACL auth.
- index: database index to SELECT (default 0).
- prefix: key prefix for every queue (default "phalcon_queue:").
- pollInterval: milliseconds between subscription poll passes (default 200).
- **`Phalcon\Queue\Adapter\Redis\RedisConnectionFactory`** - implements [`Phalcon\Contracts\Queue\ConnectionFactory`](/5.19/api/phalcon_contracts/#contractsqueueconnectionfactory)
__Uses__ `Phalcon\Contracts\Queue\ConnectionFactory` · `Phalcon\Contracts\Queue\Context` · `Phalcon\Contracts\Queue\QueueTypes` · `Phalcon\Queue\Exceptions\Exception` · `Phalcon\Storage\Adapter\Redis` · `Phalcon\Storage\Exception` · `Phalcon\Storage\SerializerFactory`
### Method Summary
### Properties
protected
array
$options = []
### Methods
Public · 2
__construct()
```php
public function __construct( array $options = [] );
```
createContext()
```php
public function createContext(): ContextInterface;
```
## Queue\Adapter\Redis\RedisConsumer
Class
Source on GitHub
Receives messages from a single Redis queue. `receive()` is overridden to
use the native blocking BRPOP (in one-second chunks, so due delayed
messages keep getting promoted) instead of the inherited polling loop.
- [`Phalcon\Queue\Adapter\AbstractConsumer`](#queueadapterabstractconsumer)
- **`Phalcon\Queue\Adapter\Redis\RedisConsumer`**
__Uses__ `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Queue\Adapter\AbstractConsumer`
### Method Summary
### Properties
protected
RedisContext
$context
### Methods
Public · 5
__construct()
```php
public function __construct(
RedisContext $context,
QueueInterface $queue
);
```
acknowledge()
```php
public function acknowledge( MessageInterface $message ): void;
```
No-op: a received message has already been removed from the queue.
receive()
```php
public function receive( int $timeout = 0 ): MessageInterface|null;
```
receiveNoWait()
```php
public function receiveNoWait(): MessageInterface|null;
```
reject()
```php
public function reject(
MessageInterface $message,
bool $requeue = false
): void;
```
## Queue\Adapter\Redis\RedisContext
Class
Source on GitHub
Redis transport session (ext-redis). Each queue is a Redis list; messages
are LPUSHed on send and RPOP/BRPOPed on receive, giving FIFO delivery.
Delayed messages live in a companion sorted set (`:delayed`) scored by
their due time in milliseconds, and are promoted into the list once due. The
destination factories come from AbstractContext.
- [`Phalcon\Queue\Adapter\AbstractContext`](#queueadapterabstractcontext)
- **`Phalcon\Queue\Adapter\Redis\RedisContext`**
__Uses__ `Phalcon\Contracts\Queue\Consumer` · `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Producer` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Contracts\Queue\SubscriptionConsumer` · `Phalcon\Queue\Adapter\AbstractContext` · `Phalcon\Queue\Adapter\MessageEnvelope` · `Phalcon\Queue\Adapter\QueueDestinationGuard` · `Redis`
### Method Summary
### Properties
protected
int
$pollInterval = 200
protected
string
$prefix = "phalcon_queue:"
protected
RedisService
$redis
### Methods
Public · 10
__construct()
```php
public function __construct(
RedisService $redis,
string $prefix = "phalcon_queue:",
int $pollInterval = 200
);
```
blockingPop()
```php
public function blockingPop(
string $queueName,
int $timeout
): MessageInterface|null;
```
Blocking pop from the back of a queue list. Promotes any due delayed
messages first, then blocks up to timeout seconds. Internal transport
API used by RedisConsumer.
close()
```php
public function close(): void;
```
createConsumer()
```php
public function createConsumer( DestinationInterface $destination ): ConsumerInterface;
```
createMessage()
```php
public function createMessage(
string $body = "",
array $properties = [],
array $headers = []
): MessageInterface;
```
createProducer()
```php
public function createProducer(): ProducerInterface;
```
createSubscriptionConsumer()
```php
public function createSubscriptionConsumer(): SubscriptionConsumerInterface;
```
popMessage()
```php
public function popMessage( string $queueName ): MessageInterface|null;
```
Non-blocking pop from the back of a queue list, or null when empty.
Promotes any due delayed messages first. Internal transport API used
by RedisConsumer.
purgeQueue()
```php
public function purgeQueue( QueueInterface $queue ): void;
```
pushMessage()
```php
public function pushMessage(
string $queueName,
MessageInterface $message,
int $delay = 0
): void;
```
Sends a message to a queue. With a positive delay (milliseconds) the
message is parked in the delayed set; otherwise it is pushed onto the
front of the list. Internal transport API used by RedisProducer.
## Queue\Adapter\Redis\RedisMessage
Class
Source on GitHub
Redis-backed message. All behavior comes from MessageTrait.
- [`Phalcon\Queue\Adapter\AbstractMessage`](#queueadapterabstractmessage)
- **`Phalcon\Queue\Adapter\Redis\RedisMessage`**
__Uses__ `Phalcon\Queue\Adapter\AbstractMessage`
## Queue\Adapter\Redis\RedisProducer
Class
Source on GitHub
Sends messages to a Redis queue. Delivery delay is supported (via the
delayed sorted set); priority and time to live are not (the defaults from
AbstractProducer reject them).
- [`Phalcon\Queue\Adapter\AbstractProducer`](#queueadapterabstractproducer)
- **`Phalcon\Queue\Adapter\Redis\RedisProducer`**
__Uses__ `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Producer` · `Phalcon\Queue\Adapter\AbstractProducer` · `Phalcon\Queue\Adapter\QueueDestinationGuard`
### Method Summary
### Properties
protected
RedisContext
$context
protected
int|null
$deliveryDelay = null
Delivery delay in milliseconds, or null when not set.
### Methods
Public · 4
__construct()
```php
public function __construct( RedisContext $context );
```
getDeliveryDelay()
```php
public function getDeliveryDelay(): int|null;
```
send()
```php
public function send(
DestinationInterface $destination,
MessageInterface $message
): void;
```
setDeliveryDelay()
```php
public function setDeliveryDelay( mixed $deliveryDelay = null ): ProducerInterface;
```
## Queue\Adapter\Redis\RedisSubscriptionConsumer
Class
Source on GitHub
Consumes from several Redis queues at once. The round-robin poll loop lives
in SubscriptionConsumerTrait.
- [`Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`](#queueadapterabstractsubscriptionconsumer)
- **`Phalcon\Queue\Adapter\Redis\RedisSubscriptionConsumer`**
__Uses__ `Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`
### Method Summary
### Properties
protected
RedisContext
$context
### Methods
Public · 1
__construct()
```php
public function __construct(
RedisContext $context,
int $pollInterval = 200
);
```
The context is retained for transports that may later need it for a
native multi-queue receive; the shared poll loop does not use it.
## Queue\Adapter\Stream\StreamConnectionFactory
Class
Source on GitHub
Builds a StreamContext.
Options:
- storageDir: directory holding the queue files (default: a private
"phalcon_queue" subdirectory of the system temp dir).
- pollInterval: milliseconds between consumer poll attempts (default 200).
- **`Phalcon\Queue\Adapter\Stream\StreamConnectionFactory`** - implements [`Phalcon\Contracts\Queue\ConnectionFactory`](/5.19/api/phalcon_contracts/#contractsqueueconnectionfactory)
__Uses__ `Phalcon\Contracts\Queue\ConnectionFactory` · `Phalcon\Contracts\Queue\Context` · `Phalcon\Contracts\Queue\QueueTypes`
### Method Summary
### Properties
protected
array
$options = []
### Methods
Public · 2
__construct()
```php
public function __construct( array $options = [] );
```
createContext()
```php
public function createContext(): ContextInterface;
```
## Queue\Adapter\Stream\StreamConsumer
Class
Source on GitHub
Receives messages from a single filesystem queue. `receive()` is the
polling loop inherited from AbstractConsumer.
- [`Phalcon\Queue\Adapter\AbstractConsumer`](#queueadapterabstractconsumer)
- **`Phalcon\Queue\Adapter\Stream\StreamConsumer`**
__Uses__ `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Queue\Adapter\AbstractConsumer`
### Method Summary
### Properties
protected
StreamContext
$context
### Methods
Public · 4
__construct()
```php
public function __construct(
StreamContext $context,
QueueInterface $queue,
int $pollInterval = 200
);
```
acknowledge()
```php
public function acknowledge( MessageInterface $message ): void;
```
No-op: a received message has already been removed from the queue file.
receiveNoWait()
```php
public function receiveNoWait(): MessageInterface|null;
```
reject()
```php
public function reject(
MessageInterface $message,
bool $requeue = false
): void;
```
## Queue\Adapter\Stream\StreamContext
Class
Source on GitHub
Filesystem transport session. Each queue is one append-only file under the
configured directory; cross-process safety comes from flock. One message
per line, stored as base64(serialize([...])) so bodies with newlines are
safe. The destination factories come from AbstractContext.
- [`Phalcon\Queue\Adapter\AbstractContext`](#queueadapterabstractcontext)
- **`Phalcon\Queue\Adapter\Stream\StreamContext`**
__Uses__ `Phalcon\Contracts\Queue\Consumer` · `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Producer` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Contracts\Queue\SubscriptionConsumer` · `Phalcon\Queue\Adapter\AbstractContext` · `Phalcon\Queue\Adapter\MessageEnvelope` · `Phalcon\Queue\Adapter\QueueDestinationGuard` · `Phalcon\Traits\Php\FileTrait`
### Method Summary
### Properties
protected
int
$pollInterval = 200
Milliseconds slept between poll attempts by consumers.
protected
string
$storageDir = ""
Directory (with trailing separator) that holds the queue files.
### Methods
Public · 9
__construct()
```php
public function __construct(
string $storageDir,
int $pollInterval = 200
);
```
close()
```php
public function close(): void;
```
createConsumer()
```php
public function createConsumer( DestinationInterface $destination ): ConsumerInterface;
```
createMessage()
```php
public function createMessage(
string $body = "",
array $properties = [],
array $headers = []
): MessageInterface;
```
createProducer()
```php
public function createProducer(): ProducerInterface;
```
createSubscriptionConsumer()
```php
public function createSubscriptionConsumer(): SubscriptionConsumerInterface;
```
popMessage()
```php
public function popMessage( string $queueName ): MessageInterface|null;
```
Removes the front message from a queue file, or null when it is empty.
Internal transport API used by StreamConsumer.
purgeQueue()
```php
public function purgeQueue( QueueInterface $queue ): void;
```
pushMessage()
```php
public function pushMessage(
string $queueName,
MessageInterface $message
): void;
```
Appends a message to the back of a queue file.
Internal transport API used by StreamProducer.
## Queue\Adapter\Stream\StreamMessage
Class
Source on GitHub
Filesystem-backed message. All behavior comes from MessageTrait.
- [`Phalcon\Queue\Adapter\AbstractMessage`](#queueadapterabstractmessage)
- **`Phalcon\Queue\Adapter\Stream\StreamMessage`**
__Uses__ `Phalcon\Queue\Adapter\AbstractMessage`
## Queue\Adapter\Stream\StreamProducer
Class
Source on GitHub
Appends messages to a filesystem queue. The Stream transport delivers in
insertion order with no scheduling, so delivery delay, priority and time to
live are not supported (the defaults from AbstractProducer reject them).
- [`Phalcon\Queue\Adapter\AbstractProducer`](#queueadapterabstractproducer)
- **`Phalcon\Queue\Adapter\Stream\StreamProducer`**
__Uses__ `Phalcon\Contracts\Queue\Destination` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Queue\Adapter\AbstractProducer` · `Phalcon\Queue\Adapter\QueueDestinationGuard`
### Method Summary
### Properties
protected
StreamContext
$context
### Methods
Public · 2
__construct()
```php
public function __construct( StreamContext $context );
```
send()
```php
public function send(
DestinationInterface $destination,
MessageInterface $message
): void;
```
## Queue\Adapter\Stream\StreamSubscriptionConsumer
Class
Source on GitHub
Consumes from several filesystem queues at once. The round-robin poll loop
lives in SubscriptionConsumerTrait.
- [`Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`](#queueadapterabstractsubscriptionconsumer)
- **`Phalcon\Queue\Adapter\Stream\StreamSubscriptionConsumer`**
__Uses__ `Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`
### Method Summary
### Properties
protected
StreamContext
$context
### Methods
Public · 1
__construct()
```php
public function __construct(
StreamContext $context,
int $pollInterval = 200
);
```
The context is retained for transports that may later need it for a
native multi-queue receive; the shared poll loop does not use it.
## Queue\Adapter\Traits\MessageTrait
Trait
Source on GitHub
Shared implementation of every Message getter/setter, plus the
correlation-id / message-id / timestamp / reply-to header conveniences.
Concrete adapter messages use this trait.
The convenience accessors are stored as transport headers under fixed keys
for binary compatibility with the wider interop ecosystem.
- **`Phalcon\Queue\Adapter\Traits\MessageTrait`**
__Uses__ `Phalcon\Contracts\Queue\QueueTypes`
__Used by__ [`Phalcon\Queue\Adapter\AbstractMessage`](#queueadapterabstractmessage)
### Method Summary
### Properties
protected
string
$body = ""
protected
array
$headers = []
protected
array
$properties = []
protected
bool
$redelivered = false
### Methods
Public · 21
__construct()
```php
public function __construct(
string $body = "",
array $properties = [],
array $headers = []
);
```
Message constructor.
getBody()
```php
public function getBody(): string;
```
Returns the message body.
getCorrelationId()
```php
public function getCorrelationId(): string|null;
```
Returns the correlation id used to correlate request/reply messages.
```php
public function getHeader(
string $name,
mixed $defaultValue = null
): mixed;
```
Returns a single header value, or the default when it is not set.
```php
public function getHeaders(): array;
```
Returns all transport headers.
getMessageId()
```php
public function getMessageId(): string|null;
```
Returns the message id.
getProperties()
```php
public function getProperties(): array;
```
Returns all application properties.
getProperty()
```php
public function getProperty(
string $name,
mixed $defaultValue = null
): mixed;
```
Returns a single property value, or the default when it is not set.
getReplyTo()
```php
public function getReplyTo(): string|null;
```
Returns the reply-to destination name.
getTimestamp()
```php
public function getTimestamp(): int|null;
```
Returns the timestamp (in milliseconds) or null when it is not set.
isRedelivered()
```php
public function isRedelivered(): bool;
```
Whether the message has been redelivered.
setBody()
```php
public function setBody( string $body ): void;
```
Sets the message body.
setCorrelationId()
```php
public function setCorrelationId( string $correlationId ): void;
```
Sets the correlation id.
```php
public function setHeader(
string $name,
mixed $value
): void;
```
Sets a single transport header.
```php
public function setHeaders( array $headers ): void;
```
Replaces all transport headers.
setMessageId()
```php
public function setMessageId( string $messageId ): void;
```
Sets the message id.
setProperties()
```php
public function setProperties( array $properties ): void;
```
Replaces all application properties.
setProperty()
```php
public function setProperty(
string $name,
mixed $value
): void;
```
Sets a single application property.
setRedelivered()
```php
public function setRedelivered( bool $redelivered ): void;
```
Marks the message as redelivered.
setReplyTo()
```php
public function setReplyTo( string $replyTo ): void;
```
Sets the reply-to destination name.
setTimestamp()
```php
public function setTimestamp( int $timestamp ): void;
```
Sets the timestamp (in milliseconds).
## Queue\Adapter\Traits\SubscriptionConsumerTrait
Trait
Source on GitHub
Shared subscription-consumer implementation. Implements the round-robin poll
loop that dispatches each subscribed consumer's messages to its callback; a
callback returning false stops consumption. The loop relies only on the
consumer's `receiveNoWait()`, so it is transport-agnostic. Concrete adapters
keep just the constructor that captures their context and poll interval.
- **`Phalcon\Queue\Adapter\Traits\SubscriptionConsumerTrait`**
__Uses__ `Phalcon\Contracts\Queue\QueueTypes`
__Used by__ [`Phalcon\Queue\Adapter\AbstractSubscriptionConsumer`](#queueadapterabstractsubscriptionconsumer)
### Method Summary
### Properties
protected
int
$pollInterval = 200
Milliseconds slept between poll passes.
protected
array
$subscriptions = []
Subscriptions keyed by queue name: [consumer, callback].
### Methods
Public · 4
consume()
```php
public function consume( int $timeout = 0 ): void;
```
Polls every subscription, dispatching each message to its callback,
blocking up to timeout milliseconds (0 = block until a callback
returns false).
subscribe()
```php
public function subscribe(
\Phalcon\Contracts\Queue\Consumer $consumer,
callable $callback
): void;
```
Subscribes a consumer; the callback receives each delivered message.
unsubscribe()
```php
public function unsubscribe( \Phalcon\Contracts\Queue\Consumer $consumer ): void;
```
Removes a previously subscribed consumer.
unsubscribeAll()
```php
public function unsubscribeAll(): void;
```
Removes every subscribed consumer.
## Queue\Cli\ConsumerTask
Class
Source on GitHub
Optional CLI runner for a queue worker - the only class coupled to
Phalcon\Cli. A thin adapter: it resolves the context from the `queueFactory`
service, binds one queue to one processor (both given as command arguments),
and runs a Worker whose lifetime bounds come from CLI options. Users not on
Phalcon\Cli use Worker directly.
Usage:
<task> <queueName> <processorServiceId> \
[--max-messages=N] [--max-time=SECONDS] \
[--max-memory=MB] [--jitter=SECONDS]
Register it in your own Phalcon\Cli\Console; it is not auto-wired into
FactoryDefault.
- `\stdClass`
- [`Phalcon\Di\Injectable`](/5.19/api/phalcon_di/#diinjectable)
- [`Phalcon\Cli\Task`](/5.19/api/phalcon_cli/#clitask)
- **`Phalcon\Queue\Cli\ConsumerTask`**
__Uses__ `Phalcon\Cli\Dispatcher` · `Phalcon\Cli\Task` · `Phalcon\Config\Config` · `Phalcon\Config\ConfigInterface` · `Phalcon\Contracts\Queue\Processor` · `Phalcon\Di\DiInterface` · `Phalcon\Queue\Consumer\QueueConsumer` · `Phalcon\Queue\Consumer\Worker` · `Phalcon\Queue\Consumer\WorkerOptions` · `Phalcon\Queue\QueueFactory`
### Method Summary
### Methods
Public · 1
mainAction()
```php
public function mainAction(): int;
```
## Queue\Consumer\BoundProcessor
Class
Source on GitHub
Binds a processor to a queue, together with the consumer that reads it.
- **`Phalcon\Queue\Consumer\BoundProcessor`**
__Uses__ `Phalcon\Contracts\Queue\Consumer` · `Phalcon\Contracts\Queue\Processor` · `Phalcon\Contracts\Queue\Queue`
### Method Summary
### Properties
protected
ConsumerInterface
$consumer
protected
ProcessorInterface
$processor
protected
QueueInterface
$queue
### Methods
Public · 4
__construct()
```php
public function __construct(
QueueInterface $queue,
ProcessorInterface $processor,
ConsumerInterface $consumer
);
```
getConsumer()
```php
public function getConsumer(): ConsumerInterface;
```
getProcessor()
```php
public function getProcessor(): ProcessorInterface;
```
getQueue()
```php
public function getQueue(): QueueInterface;
```
## Queue\Consumer\Events
Class
Source on GitHub
Lifecycle event names fired by the queue consumer through
Phalcon\Events\Manager. One public constant per event.
- **`Phalcon\Queue\Consumer\Events`**
### Constants
string
AFTER_END = "queue:afterEnd"
string
AFTER_PROCESS = "queue:afterProcess"
string
AFTER_RECEIVE = "queue:afterReceive"
string
BEFORE_PROCESS = "queue:beforeProcess"
string
BEFORE_RECEIVE = "queue:beforeReceive"
string
BEFORE_START = "queue:beforeStart"
string
PROCESSOR_EXCEPTION = "queue:processorException"
## Queue\Consumer\QueueConsumer
Class
Source on GitHub
Lean consumption runner. Binds processors to queues, polls each bound queue
round-robin, and dispatches messages to their processors while firing the
lifecycle events on `Phalcon\Queue\Consumer\Events` through the events
manager. The long-running operational shell (lifetime, signals) lives in
`Phalcon\Queue\Consumer\Worker`, which drives `consumeOnce()` and shares the
stop signal through `stop()` / `isStopRequested()`.
- [`Phalcon\Events\AbstractEventsAware`](/5.19/api/phalcon_events/#eventsabstracteventsaware)
- **`Phalcon\Queue\Consumer\QueueConsumer`** - implements [`Phalcon\Events\EventsAwareInterface`](/5.19/api/phalcon_events/#eventseventsawareinterface)
__Uses__ `Phalcon\Contracts\Queue\Consumer` · `Phalcon\Contracts\Queue\Context` · `Phalcon\Contracts\Queue\Message` · `Phalcon\Contracts\Queue\Processor` · `Phalcon\Contracts\Queue\Queue` · `Phalcon\Events\AbstractEventsAware` · `Phalcon\Events\EventsAwareInterface` · `Throwable`
### Method Summary
### Properties
protected
array<string, BoundProcessor>
$bindings = []
Bound processors keyed by queue name.
protected
ContextInterface
$context
protected
int
$pollInterval = 200
Milliseconds slept between poll passes when nothing was received.
protected
bool
$shouldStop = false
### Methods
Public · 9
__construct()
```php
public function __construct( ContextInterface $context );
```
bind()
```php
public function bind(
QueueInterface $queue,
ProcessorInterface $processor
): QueueConsumer;
```
Binds a processor to a queue. Returns self for chaining.
consume()
```php
public function consume( int $timeout = 0 ): void;
```
Runs the consumption loop, blocking up to timeout milliseconds (0 =
block until stopped). The simple loop; production setups use Worker.
consumeOnce()
```php
public function consumeOnce(): int;
```
Polls every bound queue once, dispatching any messages found. Returns
the number of messages processed in this pass, so callers (the Worker)
can apply a message-count limit across several bound queues.
end()
```php
public function end(): void;
```
Fires the `queue:afterEnd` event. Called once the loop exits.
isStopRequested()
```php
public function isStopRequested(): bool;
```
Whether a stop has been requested (by a signal, `stop()`, or an
`afterReceive` listener returning false).
setPollInterval()
```php
public function setPollInterval( int $pollInterval ): void;
```
Sets the poll interval (in milliseconds).
start()
```php
public function start(): bool;
```
Resets the stop flag and fires `queue:beforeStart`. Returns false when a
listener cancels the start.
stop()
```php
public function stop(): void;
```
Requests the consumption loop to stop after the current message.
## Queue\Consumer\Worker
Class
Source on GitHub
Long-running operational shell around a QueueConsumer. Owns the outer loop,
the bounded lifetime (max messages / seconds / memory, plus jitter) and -
when ext-pcntl is available - graceful shutdown on SIGTERM/SIGINT/SIGQUIT.
The current message always finishes before the loop stops (drain, not
guillotine), because the stop flag is only checked between iterations.
- **`Phalcon\Queue\Consumer\Worker`**
__Uses__ `Phalcon\Traits\Php\InfoTrait`
### Method Summary
### Properties
protected
QueueConsumer
$consumer
protected
WorkerOptions
$options
### Methods
Public · 3
__construct()
```php
public function __construct(
QueueConsumer $consumer,
WorkerOptions|null $options = null
);
```
handleSignal()
```php
public function handleSignal( int $signal ): void;
```
Signal handler: requests a graceful stop.
run()
```php
public function run(): int;
```
Runs the worker until a lifetime bound trips or a stop is requested.
Returns the number of messages processed.
## Queue\Consumer\WorkerOptions
Class
Source on GitHub
Immutable lifetime bounds for a Worker. A value of 0 means "no limit".
The worker stops on whichever bound trips first.
- **`Phalcon\Queue\Consumer\WorkerOptions`**
### Method Summary
### Properties
protected
int
$jitter = 0
Seconds added to maxSeconds (randomised per worker) so a pool does not restart in lockstep.
protected
int
$maxMemory = 0
Memory ceiling in megabytes.
protected
int
$maxMessages = 0
Maximum number of messages to process.
protected
int
$maxSeconds = 0
Maximum run time in seconds.
### Methods
Public · 5
__construct()
```php
public function __construct(
int $maxMessages = 0,
int $maxSeconds = 0,
int $maxMemory = 0,
int $jitter = 0
);
```
getJitter()
```php
public function getJitter(): int;
```
getMaxMemory()
```php
public function getMaxMemory(): int;
```
getMaxMessages()
```php
public function getMaxMessages(): int;
```
getMaxSeconds()
```php
public function getMaxSeconds(): int;
```
## Queue\Exceptions\DeliveryDelayNotSupportedException
Class
Source on GitHub
Thrown when the transport does not support a delivery delay.
- `\Exception`
- [`Phalcon\Queue\Exceptions\Exception`](#queueexceptionsexception)
- **`Phalcon\Queue\Exceptions\DeliveryDelayNotSupportedException`**
### Method Summary
### Methods
Public · 1
__construct()
```php
public function __construct();
```
## Queue\Exceptions\Exception
Class
Source on GitHub
Generic exception for the Queue component, and the base for every typed
queue exception.
- `\Exception`
- **`Phalcon\Queue\Exceptions\Exception`** - implements [`Phalcon\Queue\Exceptions\QueueThrowable`](#queueexceptionsqueuethrowable)
- [`Phalcon\Queue\Exceptions\DeliveryDelayNotSupportedException`](#queueexceptionsdeliverydelaynotsupportedexception)
- [`Phalcon\Queue\Exceptions\InvalidDestinationException`](#queueexceptionsinvaliddestinationexception)
- [`Phalcon\Queue\Exceptions\InvalidMessageException`](#queueexceptionsinvalidmessageexception)
- [`Phalcon\Queue\Exceptions\PriorityNotSupportedException`](#queueexceptionsprioritynotsupportedexception)
- [`Phalcon\Queue\Exceptions\PurgeQueueNotSupportedException`](#queueexceptionspurgequeuenotsupportedexception)
- [`Phalcon\Queue\Exceptions\SubscriptionConsumerNotSupportedException`](#queueexceptionssubscriptionconsumernotsupportedexception)
- [`Phalcon\Queue\Exceptions\TemporaryQueueNotSupportedException`](#queueexceptionstemporaryqueuenotsupportedexception)
- [`Phalcon\Queue\Exceptions\TimeToLiveNotSupportedException`](#queueexceptionstimetolivenotsupportedexception)
__Uses__ `Exception`
## Queue\Exceptions\InvalidDestinationException
Class
Source on GitHub
Thrown when a destination is not valid for the operation, for example a
Topic passed where a Queue is required. The action verb ("send to",
"consume from") tailors the message to the failing operation.
- `\Exception`
- [`Phalcon\Queue\Exceptions\Exception`](#queueexceptionsexception)
- **`Phalcon\Queue\Exceptions\InvalidDestinationException`**
### Method Summary
### Methods
Public · 1
__construct()
```php
public function __construct( string $action );
```
## Queue\Exceptions\InvalidMessageException
Class
Source on GitHub
Thrown when a message is not valid for the operation.
- `\Exception`
- [`Phalcon\Queue\Exceptions\Exception`](#queueexceptionsexception)
- **`Phalcon\Queue\Exceptions\InvalidMessageException`**
### Method Summary
### Methods
Public · 1
__construct()
```php
public function __construct();
```
## Queue\Exceptions\PriorityNotSupportedException
Class
Source on GitHub
Thrown when the transport does not support message priority.
- `\Exception`
- [`Phalcon\Queue\Exceptions\Exception`](#queueexceptionsexception)
- **`Phalcon\Queue\Exceptions\PriorityNotSupportedException`**
### Method Summary
### Methods
Public · 1
__construct()
```php
public function __construct();
```
## Queue\Exceptions\PurgeQueueNotSupportedException
Class
Source on GitHub
Thrown when the transport does not support purging a queue.
- `\Exception`
- [`Phalcon\Queue\Exceptions\Exception`](#queueexceptionsexception)
- **`Phalcon\Queue\Exceptions\PurgeQueueNotSupportedException`**
### Method Summary
### Methods
Public · 1
__construct()
```php
public function __construct();
```
## Queue\Exceptions\QueueThrowable
Interface
Source on GitHub
Base throwable contract for the Queue component. Every queue exception
implements it, so callers can catch all queue errors with a single type.
- `\Throwable`
- **`Phalcon\Queue\Exceptions\QueueThrowable`**
__Uses__ `Throwable`
## Queue\Exceptions\SubscriptionConsumerNotSupportedException
Class
Source on GitHub
Thrown when the transport does not support subscription consumers.
- `\Exception`
- [`Phalcon\Queue\Exceptions\Exception`](#queueexceptionsexception)
- **`Phalcon\Queue\Exceptions\SubscriptionConsumerNotSupportedException`**
### Method Summary
### Methods
Public · 1
__construct()
```php
public function __construct();
```
## Queue\Exceptions\TemporaryQueueNotSupportedException
Class
Source on GitHub
Thrown when the transport does not support temporary queues.
- `\Exception`
- [`Phalcon\Queue\Exceptions\Exception`](#queueexceptionsexception)
- **`Phalcon\Queue\Exceptions\TemporaryQueueNotSupportedException`**
### Method Summary
### Methods
Public · 1
__construct()
```php
public function __construct();
```
## Queue\Exceptions\TimeToLiveNotSupportedException
Class
Source on GitHub
Thrown when the transport does not support a message time to live.
- `\Exception`
- [`Phalcon\Queue\Exceptions\Exception`](#queueexceptionsexception)
- **`Phalcon\Queue\Exceptions\TimeToLiveNotSupportedException`**
### Method Summary
### Methods
Public · 1
__construct()
```php
public function __construct();
```
## Queue\QueueFactory
Class
Source on GitHub
Builds a queue Context from the standard Phalcon config shape. Mirrors
Phalcon\Cache\CacheFactory.
- [`Phalcon\Factory\AbstractConfigFactory`](/5.19/api/phalcon_factory/#factoryabstractconfigfactory)
- **`Phalcon\Queue\QueueFactory`**
__Uses__ `Phalcon\Config\ConfigInterface` · `Phalcon\Contracts\Queue\Context` · `Phalcon\Contracts\Queue\QueueTypes` · `Phalcon\Factory\AbstractConfigFactory` · `Phalcon\Queue\Exceptions\Exception`
### Method Summary
### Properties
protected
AdapterFactory
$adapterFactory
### Methods
Public · 3
__construct()
```php
public function __construct( AdapterFactory|null $factory = null );
```
A default AdapterFactory is created when none is supplied, so the
factory is usable straight from the DI container.
load()
```php
public function load( mixed $config ): ContextInterface;
```
Builds a Context from a config array/object.
newInstance()
```php
public function newInstance(
string $name,
array $options = []
): ContextInterface;
```
Builds a Context for the named adapter.
Protected · 1
getExceptionClass()
```php
protected function getExceptionClass(): string;
```
Returns the exception class for the factory
Source: https://docs.phalcon.io/5.19/api/phalcon_queue/index.mdx