Phalcon queue

NOTE

All classes are prefixed with Phalcon

Queue\AdapterFactory¶

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

Phalcon\Factory\AbstractConfigFactory
- Phalcon\Factory\AbstractFactory
  - Phalcon\Queue\AdapterFactory

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

Method Summary¶

public __construct( array $services = [] ) AdapterFactory constructor. public ConnectionFactoryInterface newInstance(string $name,array $options = []) Creates a new ConnectionFactory for the named adapter. protected string getExceptionClass() protected array getServices() Returns the available adapters.

Methods¶

Public · 2

`__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.

Protected · 2

`getExceptionClass()`¶

protected function getExceptionClass(): string;

`getServices()`¶

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

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

Phalcon\Queue\Adapter\AbstractConsumer - implements Phalcon\Contracts\Queue\Consumer

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

Method Summary¶

public void acknowledge( MessageInterface $message ) Acknowledges the message; the transport may then discard it. public QueueInterface getQueue() Returns the queue this consumer reads from. public MessageInterface|null receive( int $timeout = 0 ) Receives a message, blocking up to timeout milliseconds (0 = block public MessageInterface|null receiveNoWait() Receives a message without blocking, or null when none is ready. public void reject(MessageInterface $message,bool $requeue = false) Rejects the message. When requeue is true the transport redelivers it. public void setPollInterval( int $pollInterval ) Sets the poll interval (in milliseconds) used by receive().

Properties¶

protected int $pollInterval = 200 Milliseconds slept between poll attempts.

protected QueueInterface $queue The queue this consumer reads from.

Methods¶

Public · 6

`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¶

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

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

Method Summary¶

public QueueInterface createQueue( string $queueName ) Creates a queue destination by name. public QueueInterface createTemporaryQueue() Creates a uniquely named temporary queue. public TopicInterface createTopic( string $topicName ) Creates a topic destination by name.

Methods¶

Public · 3

`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¶

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

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

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¶

public int|null getDeliveryDelay() public int|null getPriority() public int|null getTimeToLive() public void send(DestinationInterface $destination,MessageInterface $message) public ProducerInterface setDeliveryDelay( mixed $deliveryDelay = null ) public ProducerInterface setPriority( mixed $priority = null ) public ProducerInterface setTimeToLive( mixed $timeToLive = null )

Methods¶

Public · 7

`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¶

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

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\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 int|bool put(string $data,int $priority,int $delay,int $ttr) Puts a job on the queue using the currently used tube. Returns the new public bool|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|bool 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 bool|int write( string $data ) Writes data to the socket, connecting first when needed.

Properties¶

protected resource $connection 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 $watchedTubes = [] Tubes currently on the watch list, keyed by tube name. A fresh connection watches "default".

Methods¶

Public · 16

`__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¶

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

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

Method Summary¶

public __construct( array $options = [] ) public ContextInterface createContext()

Properties¶

protected array $options = []

Methods¶

Public · 2

`__construct()`¶

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

`createContext()`¶

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
- Phalcon\Queue\Adapter\Beanstalk\BeanstalkConsumer - implements Phalcon\Contracts\Queue\VisibilityAware

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

Method Summary¶

public __construct(BeanstalkConnection $connection,QueueInterface $queue) public void acknowledge( MessageInterface $message ) public MessageInterface|null receive( int $timeout = 0 ) public MessageInterface|null receiveNoWait() public void reject(MessageInterface $message,bool $requeue = false) public bool touch( MessageInterface $message ) Extends the time-to-run window of a reserved job (VisibilityAware).

Constants¶

int DEFAULT_PRIORITY = 100 Default Beanstalkd priority used when releasing or burying.

Properties¶

protected BeanstalkConnection $connection

Methods¶

Public · 6

`__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¶

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
- Phalcon\Queue\Adapter\Beanstalk\BeanstalkContext - implements Phalcon\Contracts\Queue\Inspectable

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 Milliseconds slept between poll passes by a subscription consumer.

protected int $port = 11300

protected int $ttr = 86400 Default time-to-run (seconds) applied to every put.

Methods¶

Public · 10

`__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¶

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
- Phalcon\Queue\Adapter\Beanstalk\BeanstalkMessage

Uses Phalcon\Queue\Adapter\AbstractMessage

Method Summary¶

public string|null getJobId() public void setJobId( string $jobId )

Properties¶

protected string | null $jobId = null The reserved Beanstalkd job id, or null before it is reserved.

Methods¶

Public · 2

`getJobId()`¶

public function getJobId(): string|null;

`setJobId()`¶

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
- 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¶

public __construct( BeanstalkContext $context ) public int|null getDeliveryDelay() public int|null getPriority() public void send(DestinationInterface $destination,MessageInterface $message) public ProducerInterface setDeliveryDelay( mixed $deliveryDelay = null ) public ProducerInterface setPriority( mixed $priority = null )

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

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¶

Class Source on GitHub

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

Phalcon\Queue\Adapter\AbstractSubscriptionConsumer
- Phalcon\Queue\Adapter\Beanstalk\BeanstalkSubscriptionConsumer

Uses Phalcon\Queue\Adapter\AbstractSubscriptionConsumer

Method Summary¶

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

Properties¶

protected BeanstalkContext $context Retained for transports that may later need it for a native multi-queue receive; the shared poll loop does not use it.

Methods¶

Public · 1

`__construct()`¶

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

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

Uses Phalcon\Contracts\Queue\Queue

Method Summary¶

public __construct( string $queueName ) GenericQueue constructor. public string getQueueName() Returns the queue name.

Properties¶

protected string $queueName = ""

Methods¶

Public · 2

`__construct()`¶

public function __construct( string $queueName );

GenericQueue constructor.

`getQueueName()`¶

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

Uses Phalcon\Contracts\Queue\Topic

Method Summary¶

public __construct( string $topicName ) GenericTopic constructor. public string getTopicName() Returns the topic name.

Properties¶

protected string $topicName = ""

Methods¶

Public · 2

`__construct()`¶

public function __construct( string $topicName );

GenericTopic constructor.

`getTopicName()`¶

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

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

Method Summary¶

public __construct( array $options = [] ) MemoryConnectionFactory constructor. public ContextInterface createContext() Creates a new in-process context.

Properties¶

protected array $options = []

Methods¶

Public · 2

`__construct()`¶

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

MemoryConnectionFactory constructor.

`createContext()`¶

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
- Phalcon\Queue\Adapter\Memory\MemoryConsumer

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

Method Summary¶

public __construct(MemoryContext $context,QueueInterface $queue) MemoryConsumer constructor. public void acknowledge( MessageInterface $message ) No-op: a received message has already been removed from the queue. public MessageInterface|null receiveNoWait() Removes and returns the next message, or null when the queue is empty. public void reject(MessageInterface $message,bool $requeue = false) Rejects the message. When requeue is true it is put back on the queue.

Properties¶

protected MemoryContext $context

Methods¶

Public · 4

`__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¶

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
- 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¶

public void close() Closes the context and drops every stored message. public ConsumerInterface createConsumer( DestinationInterface $destination ) Creates a consumer for the given queue destination. public MessageInterface createMessage(string $body = "",array $properties = [],array $headers = []) Creates a message. public ProducerInterface createProducer() Creates a producer. public SubscriptionConsumerInterface createSubscriptionConsumer() Creates a subscription consumer. public MessageInterface|null popMessage( string $queueName ) Removes the front message from a queue, or null when it is empty. public void purgeQueue( QueueInterface $queue ) Removes all messages from the given queue. public void pushMessage(string $queueName,MessageInterface $message) Appends a message to the back of a queue.

Properties¶

protected array $queues = [] Named queues: queue name => list of messages (FIFO).

Methods¶

Public · 8

`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¶

Class Source on GitHub

In-process message. All behavior comes from MessageTrait.

Phalcon\Queue\Adapter\AbstractMessage
- 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
- Phalcon\Queue\Adapter\Memory\MemoryProducer

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

Method Summary¶

public __construct( MemoryContext $context ) public void send(DestinationInterface $destination,MessageInterface $message)

Properties¶

protected MemoryContext $context

Methods¶

Public · 2

`__construct()`¶

public function __construct( MemoryContext $context );

`send()`¶

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
- Phalcon\Queue\Adapter\Memory\MemorySubscriptionConsumer

Uses Phalcon\Queue\Adapter\AbstractSubscriptionConsumer

Method Summary¶

public __construct( MemoryContext $context )

Properties¶

protected MemoryContext $context Retained for transports that may later need it for a native multi-queue receive; the shared poll loop does not use it.

Methods¶

Public · 1

`__construct()`¶

public function __construct( MemoryContext $context );

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

Method Summary¶

public array|null decode( string $payload ) Decodes a serialized payload into a normalized {body, properties, public string encode( MessageInterface $message ) Serializes a message into its wire envelope.

Methods¶

Public · 2

`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¶

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¶

public void assertQueue(DestinationInterface $destination,string $action) Throws InvalidDestinationException unless the destination is a queue.

Methods¶

Public · 1

`assertQueue()`¶

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

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¶

public __construct( array $options = [] ) public ContextInterface createContext()

Properties¶

protected array $options = []

Methods¶

Public · 2

`__construct()`¶

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

`createContext()`¶

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
- Phalcon\Queue\Adapter\Redis\RedisConsumer

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

Method Summary¶

public __construct(RedisContext $context,QueueInterface $queue) public void acknowledge( MessageInterface $message ) No-op: a received message has already been removed from the queue. public MessageInterface|null receive( int $timeout = 0 ) public MessageInterface|null receiveNoWait() public void reject(MessageInterface $message,bool $requeue = false)

Properties¶

protected RedisContext $context

Methods¶

Public · 5

`__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¶

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 (<key>: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
- 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

Method Summary¶

public __construct(mixed $redis,string $prefix = "phalcon_queue:",int $pollInterval = 200) public MessageInterface|null blockingPop(string $queueName,int $timeout) Blocking pop from the back of a queue list. Promotes any due delayed public void close() public ConsumerInterface createConsumer( DestinationInterface $destination ) public MessageInterface createMessage(string $body = "",array $properties = [],array $headers = []) public ProducerInterface createProducer() public SubscriptionConsumerInterface createSubscriptionConsumer() public MessageInterface|null popMessage( string $queueName ) Non-blocking pop from the back of a queue list, or null when empty. public void purgeQueue( QueueInterface $queue ) public void pushMessage(string $queueName,MessageInterface $message,int $delay = 0) Sends a message to a queue. With a positive delay (milliseconds) the

Properties¶

protected int $pollInterval = 200 Milliseconds slept between poll passes by a subscription consumer.

protected string $prefix = "phalcon_queue:" Key prefix applied to every queue (and its delayed companion set).

protected \Redis $redis The connected ext-redis client.

Methods¶

Public · 10

`__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¶

Class Source on GitHub

Redis-backed message. All behavior comes from MessageTrait.

Phalcon\Queue\Adapter\AbstractMessage
- 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
- 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¶

public __construct( RedisContext $context ) public int|null getDeliveryDelay() public void send(DestinationInterface $destination,MessageInterface $message) public ProducerInterface setDeliveryDelay( mixed $deliveryDelay = null )

Properties¶

protected RedisContext $context

protected int | null $deliveryDelay = null Delivery delay in milliseconds, or null when not set.

Methods¶

Public · 4

`__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¶

Class Source on GitHub

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

Phalcon\Queue\Adapter\AbstractSubscriptionConsumer
- Phalcon\Queue\Adapter\Redis\RedisSubscriptionConsumer

Uses Phalcon\Queue\Adapter\AbstractSubscriptionConsumer

Method Summary¶

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

Properties¶

protected RedisContext $context Retained for transports that may later need it for a native multi-queue receive; the shared poll loop does not use it.

Methods¶

Public · 1

`__construct()`¶

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

Queue\Adapter\Stream\StreamConnectionFactory¶

Class Source on GitHub

Builds a StreamContext.

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

Phalcon\Queue\Adapter\Stream\StreamConnectionFactory - implements Phalcon\Contracts\Queue\ConnectionFactory

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

Method Summary¶

public __construct( array $options = [] ) public ContextInterface createContext()

Properties¶

protected array $options = []

Methods¶

Public · 2

`__construct()`¶

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

`createContext()`¶

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
- Phalcon\Queue\Adapter\Stream\StreamConsumer

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

Method Summary¶

public __construct(StreamContext $context,QueueInterface $queue,int $pollInterval = 200) public void acknowledge( MessageInterface $message ) No-op: a received message has already been removed from the queue file. public MessageInterface|null receiveNoWait() public void reject(MessageInterface $message,bool $requeue = false)

Properties¶

protected StreamContext $context

Methods¶

Public · 4

`__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¶

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
- 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¶

public __construct(string $storageDir,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 MessageInterface|null popMessage( string $queueName ) Removes the front message from a queue file, or null when it is empty. public void purgeQueue( QueueInterface $queue ) public void pushMessage(string $queueName,MessageInterface $message) Appends a message to the back of a queue file.

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

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¶

Class Source on GitHub

Filesystem-backed message. All behavior comes from MessageTrait.

Phalcon\Queue\Adapter\AbstractMessage
- 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
- Phalcon\Queue\Adapter\Stream\StreamProducer

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

Method Summary¶

public __construct( StreamContext $context ) public void send(DestinationInterface $destination,MessageInterface $message)

Properties¶

protected StreamContext $context

Methods¶

Public · 2

`__construct()`¶

public function __construct( StreamContext $context );

`send()`¶

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
- Phalcon\Queue\Adapter\Stream\StreamSubscriptionConsumer

Uses Phalcon\Queue\Adapter\AbstractSubscriptionConsumer

Method Summary¶

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

Properties¶

protected StreamContext $context Retained for transports that may later need it for a native multi-queue receive; the shared poll loop does not use it.

Methods¶

Public · 1

`__construct()`¶

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

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: \ [--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
  - Phalcon\Cli\Task
    - Phalcon\Queue\Cli\ConsumerTask

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

Method Summary¶

public int mainAction()

Methods¶

Public · 1

`mainAction()`¶

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¶

public __construct(QueueInterface $queue,ProcessorInterface $processor,ConsumerInterface $consumer) public ConsumerInterface getConsumer() public ProcessorInterface getProcessor() public QueueInterface getQueue()

Properties¶

protected ConsumerInterface $consumer

protected ProcessorInterface $processor

protected QueueInterface $queue

Methods¶

Public · 4

`__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¶

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
- Phalcon\Queue\Consumer\QueueConsumer - implements Phalcon\Events\EventsAwareInterface

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¶

public __construct( ContextInterface $context ) public QueueConsumer bind(QueueInterface $queue,ProcessorInterface $processor) Binds a processor to a queue. Returns self for chaining. public void consume( int $timeout = 0 ) Runs the consumption loop, blocking up to timeout milliseconds (0 = public bool consumeOnce() Polls every bound queue once, processing up to one message from each. public void end() Fires the queue:afterEnd event. Called once the loop exits. public bool isStopRequested() Whether a stop has been requested (by a signal, stop(), or an public void setPollInterval( int $pollInterval ) Sets the poll interval (in milliseconds). public bool start() Resets the stop flag and fires queue:beforeStart. Returns false when a public void stop() Requests the consumption loop to stop after the current message.

Properties¶

protected array $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()`¶

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¶

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¶

public __construct(QueueConsumer $consumer,WorkerOptions $options = null) public void handleSignal( int $signal ) Signal handler: requests a graceful stop. public int run() Runs the worker until a lifetime bound trips or a stop is requested.

Properties¶

protected QueueConsumer $consumer

protected WorkerOptions $options

Methods¶

Public · 3

`__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¶

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¶

public __construct(int $maxMessages = 0,int $maxSeconds = 0,int $maxMemory = 0,int $jitter = 0) public int getJitter() public int getMaxMemory() public int getMaxMessages() public int getMaxSeconds()

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

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¶

Class Source on GitHub

Thrown when the transport does not support a delivery delay.

BaseException
- Phalcon\Queue\Exceptions\Exception
  - Phalcon\Queue\Exceptions\DeliveryDelayNotSupportedException

Method Summary¶

public __construct()

Methods¶

Public · 1

`__construct()`¶

public function __construct();

Queue\Exceptions\Exception¶

Class Source on GitHub

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

BaseException
- Phalcon\Queue\Exceptions\Exception - implements Phalcon\Queue\Exceptions\QueueThrowable

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.

BaseException
- Phalcon\Queue\Exceptions\Exception
  - Phalcon\Queue\Exceptions\InvalidDestinationException

Method Summary¶

public __construct( string $action )

Methods¶

Public · 1

`__construct()`¶

public function __construct( string $action );

Queue\Exceptions\InvalidMessageException¶

Class Source on GitHub

Thrown when a message is not valid for the operation.

BaseException
- Phalcon\Queue\Exceptions\Exception
  - Phalcon\Queue\Exceptions\InvalidMessageException

Method Summary¶

public __construct()

Methods¶

Public · 1

`__construct()`¶

public function __construct();

Queue\Exceptions\PriorityNotSupportedException¶

Class Source on GitHub

Thrown when the transport does not support message priority.

BaseException
- Phalcon\Queue\Exceptions\Exception
  - Phalcon\Queue\Exceptions\PriorityNotSupportedException

Method Summary¶

public __construct()

Methods¶

Public · 1

`__construct()`¶

public function __construct();

Queue\Exceptions\PurgeQueueNotSupportedException¶

Class Source on GitHub

Thrown when the transport does not support purging a queue.

BaseException
- Phalcon\Queue\Exceptions\Exception
  - Phalcon\Queue\Exceptions\PurgeQueueNotSupportedException

Method Summary¶

public __construct()

Methods¶

Public · 1

`__construct()`¶

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

Queue\Exceptions\SubscriptionConsumerNotSupportedException¶

Class Source on GitHub

Thrown when the transport does not support subscription consumers.

BaseException
- Phalcon\Queue\Exceptions\Exception
  - Phalcon\Queue\Exceptions\SubscriptionConsumerNotSupportedException

Method Summary¶

public __construct()

Methods¶

Public · 1

`__construct()`¶

public function __construct();

Queue\Exceptions\TemporaryQueueNotSupportedException¶

Class Source on GitHub

Thrown when the transport does not support temporary queues.

BaseException
- Phalcon\Queue\Exceptions\Exception
  - Phalcon\Queue\Exceptions\TemporaryQueueNotSupportedException

Method Summary¶

public __construct()

Methods¶

Public · 1

`__construct()`¶

public function __construct();

Queue\Exceptions\TimeToLiveNotSupportedException¶

Class Source on GitHub

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

BaseException
- Phalcon\Queue\Exceptions\Exception
  - Phalcon\Queue\Exceptions\TimeToLiveNotSupportedException

Method Summary¶

public __construct()

Methods¶

Public · 1

`__construct()`¶

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
- Phalcon\Queue\QueueFactory

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

Method Summary¶

public __construct( AdapterFactory $factory = null ) QueueFactory constructor. A default AdapterFactory is created when none public ContextInterface load( mixed $config ) Builds a Context from a config array/object. public ContextInterface newInstance(string $name,array $options = []) Builds a Context for the named adapter. protected string getExceptionClass()

Properties¶

protected AdapterFactory $adapterFactory

Methods¶

Public · 3

`__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.

Protected · 1

`getExceptionClass()`¶

protected function getExceptionClass(): string;

Phalcon queue

Queue\AdapterFactory¶

Method Summary¶

Methods¶

__construct()¶

newInstance()¶

getExceptionClass()¶

getServices()¶

Queue\Adapter\AbstractConsumer¶

Method Summary¶

Properties¶

Methods¶

acknowledge()¶

getQueue()¶

receive()¶

receiveNoWait()¶

reject()¶

setPollInterval()¶

Queue\Adapter\AbstractContext¶

Method Summary¶

Methods¶

createQueue()¶

createTemporaryQueue()¶

createTopic()¶

Queue\Adapter\AbstractMessage¶

Queue\Adapter\AbstractProducer¶

Method Summary¶

Methods¶

getDeliveryDelay()¶

getPriority()¶

getTimeToLive()¶

send()¶

setDeliveryDelay()¶

setPriority()¶

setTimeToLive()¶

Queue\Adapter\AbstractSubscriptionConsumer¶

Queue\Adapter\Beanstalk\BeanstalkConnection¶

Method Summary¶

Properties¶

Methods¶

__construct()¶

buryJob()¶

connect()¶

deleteJob()¶

disconnect()¶

ignoreTube()¶

put()¶

read()¶

readStatus()¶

releaseJob()¶

reserve()¶

statsTube()¶

touchJob()¶

useTube()¶

watchTube()¶

write()¶

Queue\Adapter\Beanstalk\BeanstalkConnectionFactory¶

Method Summary¶

Properties¶

Methods¶

__construct()¶

createContext()¶

Queue\Adapter\Beanstalk\BeanstalkConsumer¶

Method Summary¶

Constants¶

Properties¶

Methods¶

__construct()¶

acknowledge()¶

receive()¶

receiveNoWait()¶

reject()¶

touch()¶

Queue\Adapter\Beanstalk\BeanstalkContext¶

Method Summary¶

Properties¶

Methods¶

__construct()¶

close()¶

createConsumer()¶

`__construct()`¶

`newInstance()`¶

`getExceptionClass()`¶

`getServices()`¶

`acknowledge()`¶

`getQueue()`¶

`receive()`¶

`receiveNoWait()`¶

`reject()`¶

`setPollInterval()`¶

`createQueue()`¶

`createTemporaryQueue()`¶

`createTopic()`¶

`getDeliveryDelay()`¶

`getPriority()`¶

`getTimeToLive()`¶

`send()`¶

`setDeliveryDelay()`¶

`setPriority()`¶

`setTimeToLive()`¶

`__construct()`¶

`buryJob()`¶

`connect()`¶

`deleteJob()`¶

`disconnect()`¶

`ignoreTube()`¶

`put()`¶

`read()`¶

`readStatus()`¶

`releaseJob()`¶

`reserve()`¶

`statsTube()`¶

`touchJob()`¶

`useTube()`¶

`watchTube()`¶

`write()`¶

`__construct()`¶

`createContext()`¶

`__construct()`¶

`acknowledge()`¶

`receive()`¶

`receiveNoWait()`¶

`reject()`¶

`touch()`¶

`__construct()`¶

`close()`¶

`createConsumer()`¶

`createMessage()`¶

`createProducer()`¶

`createSubscriptionConsumer()`¶

`getStats()`¶

`getTtr()`¶

`purgeQueue()`¶

`putMessage()`¶

`getJobId()`¶

`setJobId()`¶

`__construct()`¶

`getDeliveryDelay()`¶

`getPriority()`¶

`send()`¶

`setDeliveryDelay()`¶

`setPriority()`¶

`__construct()`¶

`__construct()`¶

`getQueueName()`¶

`__construct()`¶

`getTopicName()`¶

`__construct()`¶

`createContext()`¶

`__construct()`¶

`acknowledge()`¶

`receiveNoWait()`¶

`reject()`¶

`close()`¶

`createConsumer()`¶

`createMessage()`¶

`createProducer()`¶

`createSubscriptionConsumer()`¶

`popMessage()`¶

`purgeQueue()`¶