Phalcon Queue

Queue\AdapterFactory

Maps an adapter name to its ConnectionFactory. Mirrors Phalcon\Storage\AdapterFactory.

Uses Phalcon\Contracts\Queue\ConnectionFactory · Phalcon\Factory\AbstractFactory

Method Summary

Methods

__construct()

public function __construct( array $services = [] );

AdapterFactory constructor.

newInstance()

public function newInstance(
    string $name,
    array $options = []
): ConnectionFactoryInterface;

Creates a new ConnectionFactory for the named adapter.

getExceptionClass()

protected function getExceptionClass(): string;

getServices()

protected function getServices(): array;

Returns the available adapters.

Queue\Adapter\AbstractConsumer

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 and reject.

Transports with a native blocking receive (Redis BRPOP, Beanstalk reserve) override receive() instead of polling.

Uses Phalcon\Contracts\Queue\Consumer · Phalcon\Contracts\Queue\Message · Phalcon\Contracts\Queue\Queue

Method Summary

Properties

Methods

acknowledge()

abstract public function acknowledge( MessageInterface $message ): void;

Acknowledges the message; the transport may then discard it.

getQueue()

public function getQueue(): QueueInterface;

Returns the queue this consumer reads from.

receive()

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()

abstract public function receiveNoWait(): MessageInterface|null;

Receives a message without blocking, or null when none is ready.

reject()

abstract public function reject(
    MessageInterface $message,
    bool $requeue = false
): void;

Rejects the message. When requeue is true the transport redelivers it.

setPollInterval()

public function setPollInterval( int $pollInterval ): void;

Sets the poll interval (in milliseconds) used by receive().

Queue\Adapter\AbstractContext

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.

Uses Phalcon\Contracts\Queue\Context · Phalcon\Contracts\Queue\Queue · Phalcon\Contracts\Queue\Topic

Method Summary

Methods

createQueue()

public function createQueue( string $queueName ): QueueInterface;

Creates a queue destination by name.

createTemporaryQueue()

public function createTemporaryQueue(): QueueInterface;

Creates a uniquely named temporary queue.

createTopic()

public function createTopic( string $topicName ): TopicInterface;

Creates a topic destination by name.

Queue\Adapter\AbstractMessage

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.

Uses Phalcon\Contracts\Queue\Message · Phalcon\Queue\Adapter\Traits\MessageTrait

Queue\Adapter\AbstractProducer

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().

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

getDeliveryDelay()

public function getDeliveryDelay(): int|null;

getPriority()

public function getPriority(): int|null;

getTimeToLive()

public function getTimeToLive(): int|null;

send()

abstract public function send(
    DestinationInterface $destination,
    MessageInterface $message
): void;

setDeliveryDelay()

public function setDeliveryDelay( mixed $deliveryDelay = null ): ProducerInterface;

setPriority()

public function setPriority( mixed $priority = null ): ProducerInterface;

setTimeToLive()

public function setTimeToLive( mixed $timeToLive = null ): ProducerInterface;

Queue\Adapter\AbstractSubscriptionConsumer

Shared subscription-consumer base.

@todo Remove in v7. Kept only for backwards compatibility; compose Phalcon\Queue\Adapter\Traits\SubscriptionConsumerTrait directly instead of extending this.

Uses Phalcon\Contracts\Queue\SubscriptionConsumer · Phalcon\Queue\Adapter\Traits\SubscriptionConsumerTrait

Queue\Adapter\Beanstalk\BeanstalkConnection

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.

Uses Phalcon\Queue\Exceptions\Exception · Phalcon\Traits\Php\FileTrait

Method Summary

Properties

Methods

__construct()

public function __construct(
    string $host = "127.0.0.1",
    int $port = 11300,
    bool $persistent = false
);

buryJob()

public function buryJob(
    string $id,
    int $priority
): bool;

Puts a reserved job into the “buried” state.

connect()

public function connect(): resource;

Opens the socket connection to the Beanstalkd server.

deleteJob()

public function deleteJob( string $id ): bool;

Removes a job from the server entirely.

disconnect()

public function disconnect(): bool;

Closes the connection to the server.

ignoreTube()

public function ignoreTube( string $tube ): bool;

Removes the named tube from the watch list for the connection.

put()

public function put(
    string $data,
    int $priority,
    int $delay,
    int $ttr
): int|bool;

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()

public function read( int $length = 0 ): bool|string;

Reads a packet from the socket. Verifies the connection is available first.

readStatus()

public function readStatus(): array;

Reads the latest status line and splits it into tokens.

releaseJob()

public function releaseJob(
    string $id,
    int $priority,
    int $delay
): bool;

Puts a reserved job back into the ready queue.

reserve()

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()

public function statsTube( string $tube ): array|bool;

Returns the Beanstalkd statistics for a tube as an associative array, or false when the tube does not exist.

touchJob()

public function touchJob( string $id ): bool;

Extends the time-to-run of a reserved job.

useTube()

public function useTube( string $tube ): bool;

Changes the tube new jobs are put on. By default this is “default”.

watchTube()

public function watchTube( string $tube ): bool;

Adds the named tube to the watch list for the connection.

write()

public function write( string $data ): bool|int;

Writes data to the socket, connecting first when needed.

Queue\Adapter\Beanstalk\BeanstalkConnectionFactory

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).

Uses Phalcon\Contracts\Queue\ConnectionFactory · Phalcon\Contracts\Queue\Context

Method Summary

Properties

Methods

__construct()

public function __construct( array $options = [] );

createContext()

public function createContext(): ContextInterface;

Queue\Adapter\Beanstalk\BeanstalkConsumer

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.

Uses Phalcon\Contracts\Queue\Message · Phalcon\Contracts\Queue\Queue · Phalcon\Contracts\Queue\VisibilityAware · Phalcon\Queue\Adapter\AbstractConsumer · Phalcon\Queue\Adapter\MessageEnvelope

Method Summary

Constants

Properties

Methods

__construct()

public function __construct(
    BeanstalkConnection $connection,
    QueueInterface $queue
);

acknowledge()

public function acknowledge( MessageInterface $message ): void;

receive()

public function receive( int $timeout = 0 ): MessageInterface|null;

receiveNoWait()

public function receiveNoWait(): MessageInterface|null;

reject()

public function reject(
    MessageInterface $message,
    bool $requeue = false
): void;

touch()

public function touch( MessageInterface $message ): bool;

Extends the time-to-run window of a reserved job (VisibilityAware).

Queue\Adapter\Beanstalk\BeanstalkContext

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.

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

Properties

Methods

__construct()

public function __construct(
    string $host,
    int $port,
    bool $persistent = false,
    int $ttr = 86400,
    int $pollInterval = 200
);

close()

public function close(): void;

createConsumer()

public function createConsumer( DestinationInterface $destination ): ConsumerInterface;

createMessage()

public function createMessage(
    string $body = "",
    array $properties = [],
    array $headers = []
): MessageInterface;

createProducer()

public function createProducer(): ProducerInterface;

createSubscriptionConsumer()

public function createSubscriptionConsumer(): SubscriptionConsumerInterface;

getStats()

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()

public function getTtr(): int;

Default time-to-run (seconds) for new jobs. Used by BeanstalkProducer.

purgeQueue()

public function purgeQueue( QueueInterface $queue ): void;

putMessage()

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

Beanstalkd-backed message. Carries the reserved job id so the consumer can delete, release, bury or touch it; all other behavior comes from MessageTrait.

Uses Phalcon\Queue\Adapter\AbstractMessage

Method Summary

Properties

Methods

getJobId()

public function getJobId(): string|null;

setJobId()

public function setJobId( string $jobId ): void;

Queue\Adapter\Beanstalk\BeanstalkProducer

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).

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

Properties

Methods

__construct()

public function __construct( BeanstalkContext $context );

getDeliveryDelay()

public function getDeliveryDelay(): int|null;

getPriority()

public function getPriority(): int|null;

send()

public function send(
    DestinationInterface $destination,
    MessageInterface $message
): void;

setDeliveryDelay()

public function setDeliveryDelay( mixed $deliveryDelay = null ): ProducerInterface;

setPriority()

public function setPriority( mixed $priority = null ): ProducerInterface;

Queue\Adapter\Beanstalk\BeanstalkSubscriptionConsumer

Consumes from several Beanstalkd tubes at once. The round-robin poll loop lives in SubscriptionConsumerTrait.

Uses Phalcon\Queue\Adapter\AbstractSubscriptionConsumer

Method Summary

Properties

Methods

__construct()

public function __construct(
    BeanstalkContext $context,
    int $pollInterval = 200
);

Queue\Adapter\GenericQueue

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.

Uses Phalcon\Contracts\Queue\Queue

Method Summary

Properties

Methods

__construct()

public function __construct( string $queueName );

GenericQueue constructor.

getQueueName()

public function getQueueName(): string;

Returns the queue name.

Queue\Adapter\GenericTopic

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.

Uses Phalcon\Contracts\Queue\Topic

Method Summary

Properties

Methods

__construct()

public function __construct( string $topicName );

GenericTopic constructor.

getTopicName()

public function getTopicName(): string;

Returns the topic name.

Queue\Adapter\Memory\MemoryConnectionFactory

Builds a MemoryContext. The Memory transport takes no options.

Uses Phalcon\Contracts\Queue\ConnectionFactory · Phalcon\Contracts\Queue\Context

Method Summary

Properties

Methods

__construct()

public function __construct( array $options = [] );

MemoryConnectionFactory constructor.

createContext()

public function createContext(): ContextInterface;

Creates a new in-process context.

Queue\Adapter\Memory\MemoryConsumer

Receives messages from a single in-process queue. receive() is the polling loop inherited from AbstractConsumer.

Uses Phalcon\Contracts\Queue\Message · Phalcon\Contracts\Queue\Queue · Phalcon\Queue\Adapter\AbstractConsumer

Method Summary

Properties

Methods

__construct()

public function __construct(
    MemoryContext $context,
    QueueInterface $queue
);

MemoryConsumer constructor.

acknowledge()

public function acknowledge( MessageInterface $message ): void;

No-op: a received message has already been removed from the queue.

receiveNoWait()

public function receiveNoWait(): MessageInterface|null;

Removes and returns the next message, or null when the queue is empty.

reject()

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

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.

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

Methods

close()

public function close(): void;

Closes the context and drops every stored message.

createConsumer()

public function createConsumer( DestinationInterface $destination ): ConsumerInterface;

Creates a consumer for the given queue destination.

createMessage()

public function createMessage(
    string $body = "",
    array $properties = [],
    array $headers = []
): MessageInterface;

Creates a message.

createProducer()

public function createProducer(): ProducerInterface;

Creates a producer.

createSubscriptionConsumer()

public function createSubscriptionConsumer(): SubscriptionConsumerInterface;

Creates a subscription consumer.

popMessage()

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()

public function purgeQueue( QueueInterface $queue ): void;

Removes all messages from the given queue.

pushMessage()

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

In-process message. All behavior comes from MessageTrait.

Uses Phalcon\Queue\Adapter\AbstractMessage

Queue\Adapter\Memory\MemoryProducer

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).

Uses Phalcon\Contracts\Queue\Destination · Phalcon\Contracts\Queue\Message · Phalcon\Queue\Adapter\AbstractProducer · Phalcon\Queue\Adapter\QueueDestinationGuard

Method Summary

Properties

Methods

__construct()

public function __construct( MemoryContext $context );

send()

public function send(
    DestinationInterface $destination,
    MessageInterface $message
): void;

Queue\Adapter\Memory\MemorySubscriptionConsumer

Consumes from several in-process queues at once. The round-robin poll loop lives in SubscriptionConsumerTrait.

Uses Phalcon\Queue\Adapter\AbstractSubscriptionConsumer

Method Summary

Properties

Methods

__construct()

public function __construct( MemoryContext $context );

Queue\Adapter\MessageEnvelope

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().

Uses Phalcon\Contracts\Queue\Message

Method Summary

Methods

decode()

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()

public static function encode( MessageInterface $message ): string;

Serializes a message into its wire envelope.

Queue\Adapter\QueueDestinationGuard

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.

Uses Phalcon\Contracts\Queue\Destination · Phalcon\Contracts\Queue\Queue · Phalcon\Queue\Exceptions\InvalidDestinationException

Method Summary

Methods

assertQueue()

public static function assertQueue(
    DestinationInterface $destination,
    string $action
): void;

Throws InvalidDestinationException unless the destination is a queue.

Queue\Adapter\Redis\RedisConnectionFactory

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).

Uses Phalcon\Contracts\Queue\ConnectionFactory · Phalcon\Contracts\Queue\Context · Phalcon\Queue\Exceptions\Exception · Phalcon\Storage\Adapter\Redis · Phalcon\Storage\Exception · Phalcon\Storage\SerializerFactory

Method Summary

Properties

Methods

__construct()

public function __construct( array $options = [] );

createContext()

public function createContext(): ContextInterface;

Queue\Adapter\Redis\RedisConsumer

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.

Uses Phalcon\Contracts\Queue\Message · Phalcon\Contracts\Queue\Queue · Phalcon\Queue\Adapter\AbstractConsumer

Method Summary

Properties

Methods

__construct()

public function __construct(
    RedisContext $context,
    QueueInterface $queue
);

acknowledge()

public function acknowledge( MessageInterface $message ): void;

No-op: a received message has already been removed from the queue.

receive()

public function receive( int $timeout = 0 ): MessageInterface|null;

receiveNoWait()

public function receiveNoWait(): MessageInterface|null;

reject()

public function reject(
    MessageInterface $message,
    bool $requeue = false
): void;

Queue\Adapter\Redis\RedisContext

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 (<key>:delayed) scored by their due time in milliseconds, and are promoted into the list once due. The destination factories come from AbstractContext.

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

Method Summary

Properties

Methods

__construct()

public function __construct(
    mixed $redis,
    string $prefix = "phalcon_queue:",
    int $pollInterval = 200
);

blockingPop()

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()

public function close(): void;

createConsumer()

public function createConsumer( DestinationInterface $destination ): ConsumerInterface;

createMessage()

public function createMessage(
    string $body = "",
    array $properties = [],
    array $headers = []
): MessageInterface;

createProducer()

public function createProducer(): ProducerInterface;

createSubscriptionConsumer()

public function createSubscriptionConsumer(): SubscriptionConsumerInterface;

popMessage()

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()

public function purgeQueue( QueueInterface $queue ): void;

pushMessage()

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

Redis-backed message. All behavior comes from MessageTrait.

Uses Phalcon\Queue\Adapter\AbstractMessage

Queue\Adapter\Redis\RedisProducer

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).

Uses Phalcon\Contracts\Queue\Destination · Phalcon\Contracts\Queue\Message · Phalcon\Contracts\Queue\Producer · Phalcon\Queue\Adapter\AbstractProducer · Phalcon\Queue\Adapter\QueueDestinationGuard

Method Summary

Properties

Methods

__construct()

public function __construct( RedisContext $context );

getDeliveryDelay()

public function getDeliveryDelay(): int|null;

send()

public function send(
    DestinationInterface $destination,
    MessageInterface $message
): void;

setDeliveryDelay()

public function setDeliveryDelay( mixed $deliveryDelay = null ): ProducerInterface;

Queue\Adapter\Redis\RedisSubscriptionConsumer

Consumes from several Redis queues at once. The round-robin poll loop lives in SubscriptionConsumerTrait.

Uses Phalcon\Queue\Adapter\AbstractSubscriptionConsumer

Method Summary

Properties

Methods

__construct()

public function __construct(
    RedisContext $context,
    int $pollInterval = 200
);

Queue\Adapter\Stream\StreamConnectionFactory

Builds a StreamContext.

Options:

• storageDir: directory holding the queue files (default: system temp).
• pollInterval: milliseconds between consumer poll attempts (default 200).

Uses Phalcon\Contracts\Queue\ConnectionFactory · Phalcon\Contracts\Queue\Context

Method Summary

Properties

Methods

__construct()

public function __construct( array $options = [] );

createContext()

public function createContext(): ContextInterface;

Queue\Adapter\Stream\StreamConsumer

Receives messages from a single filesystem queue. receive() is the polling loop inherited from AbstractConsumer.

Uses Phalcon\Contracts\Queue\Message · Phalcon\Contracts\Queue\Queue · Phalcon\Queue\Adapter\AbstractConsumer

Method Summary

Properties

Methods

__construct()

public function __construct(
    StreamContext $context,
    QueueInterface $queue,
    int $pollInterval = 200
);

acknowledge()

public function acknowledge( MessageInterface $message ): void;

No-op: a received message has already been removed from the queue file.

receiveNoWait()

public function receiveNoWait(): MessageInterface|null;

reject()

public function reject(
    MessageInterface $message,
    bool $requeue = false
): void;

Queue\Adapter\Stream\StreamContext

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.

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

Methods

__construct()

public function __construct(
    string $storageDir,
    int $pollInterval = 200
);

close()

public function close(): void;

createConsumer()

public function createConsumer( DestinationInterface $destination ): ConsumerInterface;

createMessage()

public function createMessage(
    string $body = "",
    array $properties = [],
    array $headers = []
): MessageInterface;

createProducer()

public function createProducer(): ProducerInterface;

createSubscriptionConsumer()

public function createSubscriptionConsumer(): SubscriptionConsumerInterface;

popMessage()

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()

public function purgeQueue( QueueInterface $queue ): void;

pushMessage()

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

Filesystem-backed message. All behavior comes from MessageTrait.

Uses Phalcon\Queue\Adapter\AbstractMessage

Queue\Adapter\Stream\StreamProducer

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).

Uses Phalcon\Contracts\Queue\Destination · Phalcon\Contracts\Queue\Message · Phalcon\Queue\Adapter\AbstractProducer · Phalcon\Queue\Adapter\QueueDestinationGuard

Method Summary

Properties

Methods

__construct()

public function __construct( StreamContext $context );

send()

public function send(
    DestinationInterface $destination,
    MessageInterface $message
): void;

Queue\Adapter\Stream\StreamSubscriptionConsumer

Consumes from several filesystem queues at once. The round-robin poll loop lives in SubscriptionConsumerTrait.

Uses Phalcon\Queue\Adapter\AbstractSubscriptionConsumer

Method Summary

Properties

Methods

__construct()

public function __construct(
    StreamContext $context,
    int $pollInterval = 200
);

Queue\Cli\ConsumerTask

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.

Uses Phalcon\Cli\Task · Phalcon\Di\DiInterface · Phalcon\Queue\Consumer\QueueConsumer · Phalcon\Queue\Consumer\Worker · Phalcon\Queue\Consumer\WorkerOptions

Method Summary

Methods

mainAction()

public function mainAction(): int;

Queue\Consumer\BoundProcessor

Binds a processor to a queue, together with the consumer that reads it.

Uses Phalcon\Contracts\Queue\Consumer · Phalcon\Contracts\Queue\Processor · Phalcon\Contracts\Queue\Queue

Method Summary

Properties

Methods

__construct()

public function __construct(
    QueueInterface $queue,
    ProcessorInterface $processor,
    ConsumerInterface $consumer
);

getConsumer()

public function getConsumer(): ConsumerInterface;

getProcessor()

public function getProcessor(): ProcessorInterface;

getQueue()

public function getQueue(): QueueInterface;

Queue\Consumer\Events

Lifecycle event names fired by the queue consumer through Phalcon\Events\Manager. One public constant per event.

Constants

Queue\Consumer\QueueConsumer

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().

Uses Phalcon\Contracts\Queue\Context · Phalcon\Contracts\Queue\Message · Phalcon\Contracts\Queue\Processor · Phalcon\Contracts\Queue\Queue · Phalcon\Events\AbstractEventsAware · Phalcon\Events\EventsAwareInterface

Method Summary

Properties

Methods

__construct()

public function __construct( ContextInterface $context );

bind()

public function bind(
    QueueInterface $queue,
    ProcessorInterface $processor
): QueueConsumer;

Binds a processor to a queue. Returns self for chaining.

consume()

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()

public function consumeOnce(): bool;

Polls every bound queue once, processing up to one message from each. Returns true if any message was handled. Sleeps the poll interval when nothing was received so callers can loop tightly.

end()

public function end(): void;

Fires the queue:afterEnd event. Called once the loop exits.

isStopRequested()

public function isStopRequested(): bool;

Whether a stop has been requested (by a signal, stop(), or an afterReceive listener returning false).

setPollInterval()

public function setPollInterval( int $pollInterval ): void;

Sets the poll interval (in milliseconds).

start()

public function start(): bool;

Resets the stop flag and fires queue:beforeStart. Returns false when a listener cancels the start.

stop()

public function stop(): void;

Requests the consumption loop to stop after the current message.

Queue\Consumer\Worker

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.

Uses Phalcon\Traits\Php\InfoTrait

Method Summary

Properties

Methods

__construct()

public function __construct(
    QueueConsumer $consumer,
    WorkerOptions $options = null
);

handleSignal()

public function handleSignal( int $signal ): void;

Signal handler: requests a graceful stop.

run()

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

Immutable lifetime bounds for a Worker. A value of 0 means “no limit”. The worker stops on whichever bound trips first.

Method Summary

Properties

Methods

__construct()

public function __construct(
    int $maxMessages = 0,
    int $maxSeconds = 0,
    int $maxMemory = 0,
    int $jitter = 0
);

getJitter()

public function getJitter(): int;

getMaxMemory()

public function getMaxMemory(): int;

getMaxMessages()

public function getMaxMessages(): int;

getMaxSeconds()

public function getMaxSeconds(): int;

Queue\Exceptions\DeliveryDelayNotSupportedException

Thrown when the transport does not support a delivery delay.

Method Summary

Methods

__construct()

public function __construct();

Queue\Exceptions\Exception

Generic exception for the Queue component, and the base for every typed queue exception.

Uses Exception

Queue\Exceptions\InvalidDestinationException

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.

Method Summary

Methods

__construct()

public function __construct( string $action );

Queue\Exceptions\InvalidMessageException

Thrown when a message is not valid for the operation.

Method Summary

Methods

__construct()

public function __construct();

Queue\Exceptions\PriorityNotSupportedException

Thrown when the transport does not support message priority.

Method Summary

Methods

__construct()

public function __construct();

Queue\Exceptions\PurgeQueueNotSupportedException

Thrown when the transport does not support purging a queue.

Method Summary

Methods

__construct()

public function __construct();

Queue\Exceptions\QueueThrowable

Base throwable contract for the Queue component. Every queue exception implements it, so callers can catch all queue errors with a single type.

Queue\Exceptions\SubscriptionConsumerNotSupportedException

Thrown when the transport does not support subscription consumers.

Method Summary

Methods

__construct()

public function __construct();

Queue\Exceptions\TemporaryQueueNotSupportedException

Thrown when the transport does not support temporary queues.

Method Summary

Methods

__construct()

public function __construct();

Queue\Exceptions\TimeToLiveNotSupportedException

Thrown when the transport does not support a message time to live.

Method Summary

Methods

__construct()

public function __construct();

Queue\QueueFactory

Builds a queue Context from the standard Phalcon config shape. Mirrors Phalcon\Cache\CacheFactory.

Uses Phalcon\Contracts\Queue\Context · Phalcon\Factory\AbstractConfigFactory

Method Summary

Properties

Methods

__construct()

public function __construct( AdapterFactory $factory = null );

QueueFactory constructor. A default AdapterFactory is created when none is supplied, so the factory is usable straight from the DI container.

load()

public function load( mixed $config ): ContextInterface;

Builds a Context from a config array/object.

newInstance()

public function newInstance(
    string $name,
    array $options = []
): ContextInterface;

Builds a Context for the named adapter.

getExceptionClass()

protected function getExceptionClass(): string;