Skip to content

Dev #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
usernane merged 3 commits into from
May 29, 2026
Merged

Dev #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
56a0cd6
docs: add custom storage backend example
May 28, 2026
25d7cfa
docs: add detailed inline comments to examples
May 28, 2026
410634c
refactor: introduce QueuedJob value object for storage layer
May 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 51 additions & 44 deletions WebFiori/Queue/FileQueueStorage.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,15 @@ public function getFailed(): array {
$data = json_decode(file_get_contents($file), true);

if ($data !== null) {
$failed[] = $data;
$failed[] = new QueuedJob(
$data['id'],
$data['payload'] ?? '',
$data['priority'] ?? 0,
$data['attempts'] ?? 0,
0,
$data['created_at'] ?? 0,
$data['reason'] ?? null
);
}
}

Expand All @@ -84,21 +92,23 @@ public function markComplete(string $id): void {
/**
* {@inheritDoc}
*/
public function markFailed(string $id, string $reason, int $attempts): void {
$pendingFile = $this->getPendingDir().DIRECTORY_SEPARATOR.$id.'.json';
$data = [];
public function markFailed(QueuedJob $job): void {
$pendingFile = $this->getPendingDir().DIRECTORY_SEPARATOR.$job->getId().'.json';

if (file_exists($pendingFile)) {
$data = json_decode(file_get_contents($pendingFile), true) ?? [];
unlink($pendingFile);
}

$data['reason'] = $reason;
$data['attempts'] = $attempts;
$data['failed_at'] = time();

$data = [
'id' => $job->getId(),
'payload' => $job->getPayload(),
'priority' => $job->getPriority(),
'attempts' => $job->getAttempts(),
'reason' => $job->getFailReason(),
'failed_at' => time(),
];
file_put_contents(
$this->getFailedDir().DIRECTORY_SEPARATOR.$id.'.json',
$this->getFailedDir().DIRECTORY_SEPARATOR.$job->getId().'.json',
json_encode($data),
LOCK_EX
);
Expand All @@ -117,38 +127,45 @@ public function pop(int $limit = 10): array {
continue;
}

if ($data['available_at'] > time()) {
$job = new QueuedJob(
$data['id'],
$data['payload'],
$data['priority'] ?? 0,
$data['attempts'] ?? 0,
$data['available_at'] ?? 0,
$data['created_at'] ?? 0
);

if (!$job->isAvailable()) {
continue;
}

$jobs[] = $data;
$jobs[] = $job;
}

// Sort by priority descending, then by created_at ascending
usort($jobs, function ($a, $b) {
if ($a['priority'] !== $b['priority']) {
return $b['priority'] - $a['priority'];
usort($jobs, function (QueuedJob $a, QueuedJob $b) {
if ($a->getPriority() !== $b->getPriority()) {
return $b->getPriority() - $a->getPriority();
}

return $a['created_at'] - $b['created_at'];
return $a->getCreatedAt() - $b->getCreatedAt();
});

return array_slice($jobs, 0, $limit);
}
/**
* {@inheritDoc}
*/
public function push(string $id, string $payload, int $priority = 0, int $availableAt = 0): void {
public function push(QueuedJob $job): void {
$data = [
'id' => $id,
'payload' => $payload,
'priority' => $priority,
'attempts' => 0,
'available_at' => $availableAt > 0 ? $availableAt : time(),
'created_at' => time(),
'id' => $job->getId(),
'payload' => $job->getPayload(),
'priority' => $job->getPriority(),
'attempts' => $job->getAttempts(),
'available_at' => $job->getAvailableAt(),
'created_at' => $job->getCreatedAt(),
];
file_put_contents(
$this->getPendingDir().DIRECTORY_SEPARATOR.$id.'.json',
$this->getPendingDir().DIRECTORY_SEPARATOR.$job->getId().'.json',
json_encode($data),
LOCK_EX
);
Expand All @@ -166,27 +183,17 @@ public function retry(string $id): void {
$data = json_decode(file_get_contents($failedFile), true);
unlink($failedFile);

unset($data['reason'], $data['failed_at']);
$data['available_at'] = time();

file_put_contents(
$this->getPendingDir().DIRECTORY_SEPARATOR.$id.'.json',
json_encode($data),
LOCK_EX
$job = new QueuedJob(
$data['id'],
$data['payload'],
$data['priority'] ?? 0,
$data['attempts'] ?? 0,
time(),
$data['created_at'] ?? time()
);
$this->push($job);
}
/**
* {@inheritDoc}
*/
public function setAttempts(string $id, int $attempts): void {
$file = $this->getPendingDir().DIRECTORY_SEPARATOR.$id.'.json';

if (file_exists($file)) {
$data = json_decode(file_get_contents($file), true);
$data['attempts'] = $attempts;
file_put_contents($file, json_encode($data), LOCK_EX);
}
}
private function ensureDir(string $dir): void {
if (!is_dir($dir)) {
mkdir($dir, 0755, true);
Expand Down
36 changes: 25 additions & 11 deletions WebFiori/Queue/Queue.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,9 @@

/**
* Core queue class that dispatches and processes jobs.
*
* Handles serialization, encryption, and retry logic. The storage layer
* only deals with QueuedJob value objects containing opaque payloads.
*/
class Queue {
private QueueStorage $storage;
Expand All @@ -28,6 +31,8 @@ public function __construct(QueueStorage $storage) {
/**
* Dispatch a job to the queue.
*
* The job is serialized, optionally encrypted, and stored via the storage backend.
*
* @param Job $job The job to dispatch.
* @param int $priority Job priority (higher = processed first).
* @param int $delaySeconds Seconds to wait before the job becomes available.
Expand All @@ -38,7 +43,9 @@ public function dispatch(Job $job, int $priority = 0, int $delaySeconds = 0): st
$id = $this->generateId();
$payload = $this->encrypt(serialize($job));
$availableAt = $delaySeconds > 0 ? time() + $delaySeconds : 0;
$this->storage->push($id, $payload, $priority, $availableAt);

$queuedJob = new QueuedJob($id, $payload, $priority, 0, $availableAt);
$this->storage->push($queuedJob);

return $id;
}
Expand All @@ -51,7 +58,7 @@ public function flush(): void {
/**
* Returns all failed jobs.
*
* @return array
* @return QueuedJob[]
*/
public function getFailed(): array {
return $this->storage->getFailed();
Expand All @@ -75,6 +82,9 @@ public function getStorage(): QueueStorage {
/**
* Process pending jobs from the queue.
*
* Retrieves available jobs from storage, decrypts and deserializes them,
* then calls handle(). Failed jobs are retried or moved to the failed queue.
*
* @param int $limit Maximum number of jobs to process in this run.
*
* @return int Number of jobs successfully processed.
Expand All @@ -83,15 +93,17 @@ public function process(int $limit = 10): int {
$pending = $this->storage->pop($limit);
$processed = 0;

foreach ($pending as $item) {
$id = $item['id'];
$attempts = ($item['attempts'] ?? 0) + 1;
foreach ($pending as $queuedJob) {
$id = $queuedJob->getId();
$attempts = $queuedJob->getAttempts() + 1;

try {
$job = unserialize($this->decrypt($item['payload']));
$job = unserialize($this->decrypt($queuedJob->getPayload()));

if (!($job instanceof Job)) {
$this->storage->markFailed($id, 'Payload is not a valid Job instance.', $attempts);
$queuedJob->setAttempts($attempts);
$queuedJob->setFailReason('Payload is not a valid Job instance.');
$this->storage->markFailed($queuedJob);

continue;
}
Expand All @@ -101,13 +113,16 @@ public function process(int $limit = 10): int {
$processed++;
} catch (\Throwable $e) {
if ($attempts >= $job->getMaxAttempts()) {
$this->storage->markFailed($id, $e->getMessage(), $attempts);
$queuedJob->setAttempts($attempts);
$queuedJob->setFailReason($e->getMessage());
$this->storage->markFailed($queuedJob);
} else {
// Re-queue with updated attempt count and delay
$this->storage->markComplete($id);
$delay = $job->getRetryDelaySeconds() * $attempts;
$this->storage->push($id, $item['payload'], $item['priority'] ?? 0, time() + $delay);
$this->storage->setAttempts($id, $attempts);
$queuedJob->setAttempts($attempts);
$queuedJob->setAvailableAt(time() + $delay);
$this->storage->push($queuedJob);
}
}
}
Expand Down Expand Up @@ -151,7 +166,6 @@ private function decrypt(string $data): string {

return $plaintext !== false ? $plaintext : $data;
}

/**
* Encrypts data if QUEUE_KEY environment variable is set.
*
Expand Down
49 changes: 25 additions & 24 deletions WebFiori/Queue/QueueStorage.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,20 @@

/**
* Interface for queue storage backends.
*
* Implementations persist QueuedJob objects and retrieve them for processing.
* The storage layer never interprets the job payload — it treats it as an
* opaque string. Serialization and encryption are handled by the Queue class.
*/
interface QueueStorage {
/**
* Remove all failed jobs.
* Remove all failed jobs permanently.
*/
public function flush(): void;
/**
* Returns all failed jobs.
*
* @return array Array of associative arrays with keys: id, payload, reason, attempts.
* @return QueuedJob[] Array of failed queued jobs.
*/
public function getFailed(): array;
/**
Expand All @@ -32,47 +36,44 @@ public function getFailed(): array;
*/
public function getPendingCount(): int;
/**
* Mark a job as completed and remove it from the queue.
* Remove a completed job from the pending queue.
*
* Called after a job's handle() method succeeds.
*
* @param string $id The job identifier.
*/
public function markComplete(string $id): void;
/**
* Mark a job as failed.
* Move a job from pending to the failed queue.
*
* @param string $id The job identifier.
* @param string $reason The failure reason.
* @param int $attempts Number of attempts made.
* Called when all retry attempts are exhausted.
*
* @param QueuedJob $job The failed job with failReason and attempts set.
*/
public function markFailed(string $id, string $reason, int $attempts): void;
public function markFailed(QueuedJob $job): void;
/**
* Pop the next available jobs from the queue.
* Retrieve the next available jobs from the pending queue.
*
* Must return jobs where:
* 1. availableAt <= current time (not delayed)
* 2. Sorted by priority descending (highest first)
* 3. Limited to $limit count
*
* @param int $limit Maximum number of jobs to retrieve.
*
* @return array Array of associative arrays with keys: id, payload, attempts, priority.
* @return QueuedJob[] Array of available queued jobs.
*/
public function pop(int $limit = 10): array;
/**
* Push a job payload onto the queue.
* Store a queued job in the pending queue.
*
* @param string $id Unique job identifier.
* @param string $payload Serialized job data.
* @param int $priority Job priority (higher = processed first).
* @param int $availableAt Unix timestamp when the job becomes available.
* @param QueuedJob $job The job entry to store.
*/
public function push(string $id, string $payload, int $priority = 0, int $availableAt = 0): void;
public function push(QueuedJob $job): void;
/**
* Retry a failed job by moving it back to the pending queue.
* Move a failed job back to the pending queue for reprocessing.
*
* @param string $id The job identifier.
*/
public function retry(string $id): void;
/**
* Update the attempt count for a pending job.
*
* @param string $id The job identifier.
* @param int $attempts The new attempt count.
*/
public function setAttempts(string $id, int $attempts): void;
}
Loading
Loading