webfiori / queue

v1.2.0 2026-09-20 19:05 UTC

Requires

• php: >=8.1

Requires (Dev)

• friendsofphp/php-cs-fixer: ^3.0
• phpunit/phpunit: ^10.0

MIT

• Ibrahim BinAlshikh

Published on Packagist.org on 2026-09-20 19:10 UTC

README

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

Table of Contents

• Supported PHP Versions
• Features
• Installation
• Usage
• API
• Testing
• Examples
• Contributing
• License
• Support
• Changelog

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('[e-mail]', 'Welcome!'));
$queue->dispatch(new SendEmailJob('[e-mail]', 'Priority!'), priority: 10);
$queue->dispatch(new SendEmailJob('[e-mail]', '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('[e-mail]', 'Hello'));
QueueFacade::process();

Failed Jobs

// View failed jobs (each is a QueuedJob)
$failed = $queue->getFailed();

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

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

Observing Failures

By default, exceptions thrown by a job's handle() are caught and turned into retries or a failed-job record. Register an opt-in error callback to observe every caught throwable — with full exception context — for both intermediate retries and terminal failures. This makes it easy to bridge queue failures into a centralized logger (e.g. a globally registered WebFiori\Error\Handler).

$queue->setOnError(function (?Job $job, \Throwable $e, int $attempts, bool $willRetry): void {
    // $job is null if the stored payload was not a valid Job instance.
    error_log(sprintf(
        '[queue] %s failed on attempt %d (%s): %s',
        $job !== null ? get_class($job) : 'invalid-payload',
        $attempts,
        $willRetry ? 'will retry' : 'terminal',
        $e->getMessage()
    ));
});

The failed QueuedJob also records richer context — its getFailReason() now includes the exception class in addition to the message. Behavior is unchanged when no callback is registered. The same callback can be set on the facade via QueueFacade::setOnError(...).

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
setOnError(?callable $cb): Queue	Register a callback invoked on every caught throwable: fn(?Job $job, \Throwable $e, int $attempts, bool $willRetry)
getOnError(): ?callable	Get the registered error callback

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

Testing

Run the test suite with:

composer test

Examples

Runnable examples are available in the examples/ directory:

• examples/01-basic-queue.php — dispatching and processing jobs
• examples/02-custom-storage.php — implementing a custom storage backend

Contributing

Contributions are welcome. Please open an issue to discuss significant changes, follow Conventional Commits for commit messages, and ensure composer test passes before opening a pull request.

License

MIT

Support

• Issues: GitHub Issues

Changelog

See CHANGELOG.md for a full history of changes and releases.