WebFiori Queue

A lightweight job queue library for PHP with file-based storage, priority ordering, and retry logic.

Supported PHP Versions

This library requires PHP 8.1 or higher.

Build Status

Features

• Job interface — define units of work with retry configuration
• FileQueueStorage — file-based backend, zero infrastructure needed
• Priority ordering — higher priority jobs processed first
• Delayed dispatch — schedule jobs to run after a delay
• Automatic retry — failed jobs re-queued with configurable backoff
• Failed job tracking — inspect and retry failed jobs
• Static facade (QueueFacade) for quick usage without DI
• Zero dependencies — requires only PHP 8.1+

Installation

composer require webfiori/queue

Usage

Define a Job

use WebFiori\Queue\Job;

class SendEmailJob implements Job {
    public function __construct(
        private string $to,
        private string $subject
    ) {}

    public function handle(): void {
        // Send the email
        mail($this->to, $this->subject, 'Hello!');
    }

    public function getMaxAttempts(): int {
        return 3;
    }

    public function getRetryDelaySeconds(): int {
        return 60; // wait 60s × attempt number before retry
    }
}

Dispatch and Process

use WebFiori\Queue\FileQueueStorage;
use WebFiori\Queue\Queue;

$queue = new Queue(new FileQueueStorage('/path/to/storage'));

// Dispatch jobs
$queue->dispatch(new SendEmailJob('user@example.com', 'Welcome!'));
$queue->dispatch(new SendEmailJob('vip@example.com', 'Priority!'), priority: 10);
$queue->dispatch(new SendEmailJob('later@example.com', 'Delayed'), delaySeconds: 300);

// Process pending jobs (call this from a scheduler or worker)
$processed = $queue->process(limit: 50);

Static Facade

use WebFiori\Queue\QueueFacade;

QueueFacade::dispatch(new SendEmailJob('user@example.com', 'Hello'));
QueueFacade::process();

Failed Jobs

// View failed jobs
$failed = $queue->getFailed();

// Retry a specific failed job
$queue->retry($failed[0]['id']);

// Clear all failed jobs
$queue->flush();

Inspecting Pending Jobs

// List all pending jobs (including delayed ones not yet available)
$pending = $queue->getPending();

foreach ($pending as $queuedJob) {
    echo sprintf(
        "ID: %s | Priority: %d | Attempts: %d | Available: %s\n",
        $queuedJob->getId(),
        $queuedJob->getPriority(),
        $queuedJob->getAttempts(),
        date('Y-m-d H:i:s', $queuedJob->getAvailableAt())
    );
}

// Quick count (works with any storage backend)
$count = $queue->getPendingCount();

Note: getPending() requires a storage backend that implements ListableQueueStorage (e.g. FileQueueStorage). Backends like SQS that don't support listing will throw a LogicException.

API

Job (interface)

Method	Description
handle(): void	Execute the job logic
getMaxAttempts(): int	Maximum retry attempts
getRetryDelaySeconds(): int	Base delay between retries (multiplied by attempt number)

Queue

Method	Description
__construct(QueueStorage $storage)	Create queue with storage backend
dispatch(Job $job, int $priority = 0, int $delaySeconds = 0): string	Add job to queue, returns job ID
process(int $limit = 10): int	Process pending jobs, returns count processed
retry(string $id): void	Retry a failed job
getPendingCount(): int	Number of pending jobs
getPending(): array	All pending jobs (requires ListableQueueStorage)
getFailed(): array	All failed jobs
flush(): void	Remove all failed jobs
getStorage(): QueueStorage	Get the storage backend

QueueStorage (interface)

Method	Description
push(string $id, string $payload, int $priority, int $availableAt): void	Store a job
pop(int $limit): array	Retrieve available jobs
markComplete(string $id): void	Remove completed job
markFailed(string $id, string $reason, int $attempts): void	Move to failed
setAttempts(string $id, int $attempts): void	Update attempt count
retry(string $id): void	Move failed job back to pending
getPendingCount(): int	Count pending jobs
getFailed(): array	Get all failed jobs
flush(): void	Clear failed jobs

ListableQueueStorage (interface, extends QueueStorage)

Method	Description
getPending(): array	Returns all pending jobs (including delayed), sorted by priority desc

Backends that support listing (files, database, Redis) implement ListableQueueStorage. Backends that only support push/pop semantics (SQS, RabbitMQ) implement the base QueueStorage.

QueueFacade

Static wrapper. Same methods as Queue plus getInstance(), setInstance(), reset().

License

MIT