PhenixPHP

Appearance

Queue ​

Table of Contents ​

Introduction ​

Phenix queues allow you to process work asynchronously using Tasks.

A queueable task is a class that extends Phenix\Tasks\QueuableTask (from src/Tasks), and is pushed/processed through the queue system in src/Queue.

Supported drivers:

  • parallel
  • database
  • redis

Creating Tasks ​

Generate a Queuable Task ​

Generate a queueable task:

sh
php phenix make:task SendWelcomeEmail --queue

Force overwrite:

sh
php phenix make:task SendWelcomeEmail --queue --force
php phenix make:task SendWelcomeEmail --queue -f

Task Class ​

PHP
<?php

declare(strict_types=1);

namespace App\Tasks;

use Amp\Cancellation;
use Amp\Sync\Channel;
use Phenix\Tasks\QueuableTask;
use Phenix\Tasks\Result;

class SendWelcomeEmail extends QueuableTask
{
    protected ?int $maxTries = 5;

    public function __construct()
    {
        $this->setTimeout(120);
    }

    protected function handle(Channel $channel, Cancellation $cancellation): Result
    {
        // ...

        return Result::success(message: 'Welcome email sent');
    }
}

Dispatching Tasks ​

Basic Dispatch ​

PHP
use App\Tasks\SendWelcomeEmail;

SendWelcomeEmail::dispatch();

Dispatch to a Specific Queue and Connection ​

Use enqueue() when you want to configure queue/connection before dispatching:

PHP
SendWelcomeEmail::enqueue()
    ->onQueue('emails')
    ->onConnection('default')
    ->dispatch();

Conditional Dispatch ​

PHP
SendWelcomeEmail::dispatchIf(true);

$pending = SendWelcomeEmail::enqueueIf(fn (): bool => true);
$pending?->dispatch();

Dispatch with Queue Facade ​

PHP
use App\Tasks\SendWelcomeEmail;
use Phenix\Facades\Queue;

Queue::push(new SendWelcomeEmail());
Queue::pushOn('emails', new SendWelcomeEmail());

You can also consume manually:

PHP
$task = Queue::pop('emails');
$tasks = Queue::popChunk(10, 'emails');

Running Queue Workers ​

Start the Worker ​

sh
php phenix queue:work

You can pass the connection as an argument:

sh
php phenix queue:work mysql

Process a Single Iteration ​

sh
php phenix queue:work --once
php phenix queue:work -o

Process a Specific Queue ​

sh
php phenix queue:work --queue=emails

Chunk Processing ​

Enable chunk mode:

sh
php phenix queue:work --chunks

In chunk mode, prefer one queue name per worker process.

Sleep Time ​

Seconds to sleep when no tasks are available:

sh
php phenix queue:work --sleep=1
php phenix queue:work -s 1

Retries, Attempts and Timeouts ​

Execution flow:

  • Worker executes task->output().
  • On success, the task is marked completed.
  • On failure, the task is retried until max tries; then it is marked failed.

Max tries precedence:

  • QueuableTask::$maxTries in your task class (if defined).
  • Otherwise worker option maxTries (default: 3).

Timeout is per task through setTimeout() / getTimeout() and enforced with TimeoutCancellation.

Queue Drivers ​

Parallel Driver ​

parallel uses in-memory queueing with async processing. I automatically starts processing after call push().

Database Driver ​

database stores tasks in a SQL table and reserves tasks atomically via transactions.

Generate the queue table migration:

sh
php phenix queue:table
php phenix queue:table --force
php phenix queue:table -f

Then run:

sh
php phenix migrate

Generated migration filename:

  • database/migrations/20250101205638_create_tasks_table.php

Redis Driver ​

redis stores each payload under its task ID and uses queue-specific ready, reserved, and delayed structures. Lua scripts atomically perform push, pop/reserve, acknowledgement, retry and failure transitions.

If a worker dies after reservation, the task ID and payload remain available in Redis until the reservation can be recovered. Redis persistence and replication must be configured separately according to the application's durability needs. The next pop from that queue moves an expired task back to its original ready queue. Delivery is therefore at least once; task handlers must remain idempotent.

Configure the lease duration with REDIS_QUEUE_RESERVATION_TIMEOUT. It must be longer than the expected execution time of a task. A task still running after its lease expires may be reserved by another worker.

This key layout is incompatible with the previous Redis queue implementation. Drain old Redis queues before upgrading; existing queues:{name} payload lists are not migrated automatically.

Calling clear() removes pending and delayed tasks from the selected queue. It does not delete active reservations; those remain available for acknowledgement or recovery after their lease expires.

CLI Commands ​

sh
php phenix make:task SendWelcomeEmail --queue

php phenix queue:work
php phenix queue:work mysql --queue=default --sleep=3
php phenix queue:work --once
php phenix queue:work --chunks

php phenix queue:table
php phenix queue:table --force
php phenix queue:table -f

Configuration Reference ​

Queue configuration is defined in config/queue.php:

PHP
<?php

declare(strict_types=1);

return [
    'default' => env('QUEUE_DRIVER', static fn (): string => 'database'),

    'drivers' => [
        'parallel' => [
            'timeout' => env('PARALLEL_QUEUE_TIMEOUT', static fn (): int => 2),
            'chunk_processing' => env('PARALLEL_QUEUE_CHUNK_PROCESSING', static fn (): bool => true),
            'chunk_size' => env('PARALLEL_QUEUE_CHUNK_SIZE', static fn (): int => 10),
            'max_retries' => env('PARALLEL_QUEUE_MAX_RETRIES', static fn (): int => 3),
            'retry_delay' => env('PARALLEL_QUEUE_RETRY_DELAY', static fn (): int => 2),
            'interval' => env('PARALLEL_QUEUE_INTERVAL', static fn (): float => 2.0),
        ],

        'database' => [
            'connection' => env('DB_QUEUE_CONNECTION', static fn (): string => 'mysql'),
            'table' => env('DB_QUEUE_TABLE', static fn (): string => 'tasks'),
            'queue' => env('DB_QUEUE', static fn (): string => 'default'),
        ],

        'redis' => [
            'connection' => env('REDIS_QUEUE_CONNECTION', static fn (): string => 'default'),
            'queue' => env('REDIS_QUEUE', static fn (): string => 'default'),
            'reservation_timeout' => env('REDIS_QUEUE_RESERVATION_TIMEOUT', static fn (): int => 60),
        ],
    ],
];

Queue Facade Reference ​

Phenix\Facades\Queue methods:

  • Queue operations:
    • push(QueuableTask $task)
    • pushOn(string $queue, QueuableTask $task)
    • pop(string|null $queueName = null)
    • popChunk(int $limit, string|null $queueName = null)
    • size()
    • clear()
    • getConnectionName()
    • setConnectionName(string $name)
    • driver(QueueDriver|null $driverName = null)
  • Logging and fakes:
    • log()
    • fake()
    • fakeWhen(string $taskClass, Closure $callback)
    • fakeTimes(string $taskClass, int $times)
    • fakeOnce(string $taskClass)
    • fakeOnly(string $taskClass)
    • fakeExcept(string $taskClass)
    • getQueueLog()
    • resetQueueLog()
    • resetFaking()
  • Assertions helper:
    • expect(string $taskClass): Phenix\Testing\TestQueue

Testing Queue Tasks ​

Queue Logging ​

PHP
use App\Tasks\SendWelcomeEmail;
use Phenix\Facades\Queue;

Queue::log();
Queue::push(new SendWelcomeEmail());

$log = Queue::getQueueLog();

Queue::resetQueueLog();

Faking Queue Pushes ​

PHP
use App\Tasks\SendWelcomeEmail;
use Phenix\Data\Collection;
use Phenix\Facades\Queue;

Queue::fake();
Queue::fakeOnly(SendWelcomeEmail::class);
Queue::fakeExcept(SendWelcomeEmail::class);
Queue::fakeOnce(SendWelcomeEmail::class);
Queue::fakeTimes(SendWelcomeEmail::class, 2);

Queue::fakeWhen(SendWelcomeEmail::class, function (Collection $log): bool {
    return $log->count() < 3;
});

Queue::resetFaking();

Queue Assertions ​

PHP
Queue::expect(SendWelcomeEmail::class)->toBePushed();
Queue::expect(SendWelcomeEmail::class)->toBePushedTimes(1);
Queue::expect(SendWelcomeEmail::class)->toNotBePushed();
Queue::expect(SendWelcomeEmail::class)->toPushNothing();

Predicate assertion:

PHP
Queue::expect(SendWelcomeEmail::class)->toBePushed(function ($task): bool {
    return $task !== null && $task->getQueueName() === 'emails';
});

Production Behavior ​

Queue log/fake helpers are intentionally ignored in production environment.