diff --git a/WebFiori/Queue/FileQueueStorage.php b/WebFiori/Queue/FileQueueStorage.php index 255acca..d8b0aec 100644 --- a/WebFiori/Queue/FileQueueStorage.php +++ b/WebFiori/Queue/FileQueueStorage.php @@ -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 + ); } } @@ -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 ); @@ -117,20 +127,27 @@ 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); @@ -138,17 +155,17 @@ public function pop(int $limit = 10): array { /** * {@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 ); @@ -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); diff --git a/WebFiori/Queue/Queue.php b/WebFiori/Queue/Queue.php index 5006132..04d9b2b 100644 --- a/WebFiori/Queue/Queue.php +++ b/WebFiori/Queue/Queue.php @@ -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; @@ -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. @@ -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; } @@ -51,7 +58,7 @@ public function flush(): void { /** * Returns all failed jobs. * - * @return array + * @return QueuedJob[] */ public function getFailed(): array { return $this->storage->getFailed(); @@ -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. @@ -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; } @@ -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); } } } @@ -151,7 +166,6 @@ private function decrypt(string $data): string { return $plaintext !== false ? $plaintext : $data; } - /** * Encrypts data if QUEUE_KEY environment variable is set. * diff --git a/WebFiori/Queue/QueueStorage.php b/WebFiori/Queue/QueueStorage.php index 8137a11..57b3652 100644 --- a/WebFiori/Queue/QueueStorage.php +++ b/WebFiori/Queue/QueueStorage.php @@ -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; /** @@ -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; } diff --git a/WebFiori/Queue/QueuedJob.php b/WebFiori/Queue/QueuedJob.php new file mode 100644 index 0000000..3404766 --- /dev/null +++ b/WebFiori/Queue/QueuedJob.php @@ -0,0 +1,178 @@ +id = $id; + $this->payload = $payload; + $this->priority = $priority; + $this->attempts = $attempts; + $this->availableAt = $availableAt > 0 ? $availableAt : time(); + $this->createdAt = $createdAt > 0 ? $createdAt : time(); + $this->failReason = $failReason; + } + /** + * Returns the number of times this job has been attempted. + * + * @return int + */ + public function getAttempts(): int { + return $this->attempts; + } + /** + * Returns the Unix timestamp when this job becomes available for processing. + * + * @return int + */ + public function getAvailableAt(): int { + return $this->availableAt; + } + /** + * Returns the Unix timestamp when this job was first queued. + * + * @return int + */ + public function getCreatedAt(): int { + return $this->createdAt; + } + /** + * Returns the failure reason, or null if the job has not failed. + * + * @return string|null + */ + public function getFailReason(): ?string { + return $this->failReason; + } + /** + * Returns the unique identifier of this job. + * + * @return string + */ + public function getId(): string { + return $this->id; + } + /** + * Returns the serialized job payload. + * + * This is an opaque string — the storage should not attempt to + * interpret, parse, or modify it. + * + * @return string + */ + public function getPayload(): string { + return $this->payload; + } + /** + * Returns the job priority. + * + * Higher values indicate higher priority. Jobs with higher priority + * are returned first by QueueStorage::pop(). + * + * @return int + */ + public function getPriority(): int { + return $this->priority; + } + /** + * Checks if this job is currently available for processing. + * + * @return bool True if availableAt <= current time. + */ + public function isAvailable(): bool { + return $this->availableAt <= time(); + } + /** + * Sets the number of attempts. + * + * @param int $attempts The new attempt count. + */ + public function setAttempts(int $attempts): void { + $this->attempts = $attempts; + } + /** + * Sets when the job becomes available. + * + * @param int $timestamp Unix timestamp. + */ + public function setAvailableAt(int $timestamp): void { + $this->availableAt = $timestamp; + } + /** + * Sets the failure reason. + * + * @param string|null $reason The failure message. + */ + public function setFailReason(?string $reason): void { + $this->failReason = $reason; + } +} diff --git a/examples/01-basic-queue.php b/examples/01-basic-queue.php index fbef385..2d32b42 100644 --- a/examples/01-basic-queue.php +++ b/examples/01-basic-queue.php @@ -2,6 +2,12 @@ /** * Example: Dispatching and processing jobs. + * + * This example demonstrates the basic queue workflow: + * 1. Create a job class that implements the Job interface + * 2. Create a queue with a storage backend + * 3. Dispatch jobs to the queue + * 4. Process pending jobs (typically done by a scheduler or worker) */ require_once __DIR__.'/../vendor/autoload.php'; @@ -9,45 +15,88 @@ use WebFiori\Queue\Job; use WebFiori\Queue\Queue; +/** + * A job that simulates sending an email. + * + * Every job must implement the Job interface which requires three methods: + * - handle(): The actual work the job performs + * - getMaxAttempts(): How many times to retry if handle() throws an exception + * - getRetryDelaySeconds(): Base delay between retries (multiplied by attempt number) + */ class SendEmailJob implements Job { private string $subject; private string $to; + /** + * Constructor receives the data needed to perform the job. + * This data is serialized when the job is stored in the queue, + * and deserialized when the job is processed. + */ public function __construct(string $to, string $subject) { $this->to = $to; $this->subject = $subject; } + /** + * If handle() fails, the queue will retry up to this many times. + * After all attempts are exhausted, the job moves to the failed queue. + */ public function getMaxAttempts(): int { return 3; } + /** + * Delay between retries in seconds. + * The actual delay is: getRetryDelaySeconds() * attemptNumber + * So with 60 seconds: 1st retry after 60s, 2nd after 120s, 3rd after 180s. + */ public function getRetryDelaySeconds(): int { return 60; } + /** + * This method contains the actual work. + * It is called by Queue::process() when the job is picked up. + * If this method throws an exception, the job will be retried + * up to getMaxAttempts() times. + */ public function handle(): void { echo "Sending email to {$this->to}: {$this->subject}\n"; } } -// Create queue with file storage +// --- Step 1: Create a queue with file-based storage --- +// FileQueueStorage stores each job as a JSON file in the given directory. +// Two subdirectories are created automatically: 'pending/' and 'failed/' $queue = new Queue(new FileQueueStorage(__DIR__.'/queue-storage')); -// Dispatch jobs +// --- Step 2: Dispatch jobs --- +// dispatch() serializes the job object and stores it in the queue. +// It returns a unique job ID (32-character hex string). +// The job is NOT executed here — it's just stored for later processing. $queue->dispatch(new SendEmailJob('user@example.com', 'Welcome!')); -$queue->dispatch(new SendEmailJob('admin@example.com', 'New signup'), 10); // high priority -$queue->dispatch(new SendEmailJob('support@example.com', 'Ticket update')); + +// Jobs can have priority. Higher priority = processed first. +// Default priority is 0. +$queue->dispatch(new SendEmailJob('admin@example.com', 'New signup'), 10); + +// Jobs can be delayed. This job won't be available for processing +// until 300 seconds (5 minutes) from now. +$queue->dispatch(new SendEmailJob('support@example.com', 'Ticket update'), 0, 0); echo "Pending jobs: ".$queue->getPendingCount()."\n\n"; -// Process all pending jobs +// --- Step 3: Process pending jobs --- +// process() picks up available jobs (sorted by priority), deserializes them, +// and calls handle() on each one. +// The $limit parameter controls how many jobs to process in one call. +// In production, this is typically called by a scheduler task every minute. echo "Processing...\n"; $processed = $queue->process(); echo "Processed: $processed jobs\n"; echo "Remaining: ".$queue->getPendingCount()."\n"; -// Cleanup +// --- Cleanup (for this example only) --- array_map('unlink', glob(__DIR__.'/queue-storage/pending/*.json')); array_map('unlink', glob(__DIR__.'/queue-storage/failed/*.json')); rmdir(__DIR__.'/queue-storage/pending'); diff --git a/examples/02-custom-storage.php b/examples/02-custom-storage.php new file mode 100644 index 0000000..c99326f --- /dev/null +++ b/examples/02-custom-storage.php @@ -0,0 +1,199 @@ +failed = []; + } + + /** + * Returns all failed jobs for inspection. + * + * @return array Each element has: id, payload, reason, attempts + */ + public function getFailed(): array { + return array_values($this->failed); + } + + /** + * Returns the number of jobs waiting to be processed. + * + * @return int + */ + public function getPendingCount(): int { + return count($this->pending); + } + + /** + * Remove a completed job from the pending queue. + * Called by Queue::process() after handle() succeeds. + * + * @param string $id The job identifier. + */ + public function markComplete(string $id): void { + unset($this->pending[$id]); + } + + /** + * Move a job from pending to failed. + * Called by Queue::process() when all retry attempts are exhausted. + * + * @param string $id The job identifier. + * @param string $reason The exception message from the last failed attempt. + * @param int $attempts Total number of attempts made. + */ + public function markFailed(string $id, string $reason, int $attempts): void { + $data = $this->pending[$id] ?? []; + unset($this->pending[$id]); + $data['reason'] = $reason; + $data['attempts'] = $attempts; + $this->failed[$id] = $data; + } + + /** + * Retrieve the next available jobs from the queue. + * + * Must return jobs that: + * 1. Have available_at <= current time (not delayed) + * 2. Are sorted by priority (highest first) + * 3. Are limited to $limit count + * + * @param int $limit Maximum number of jobs to return. + * @return array Each element is an associative array with keys: + * id, payload, priority, attempts, available_at + */ + public function pop(int $limit = 10): array { + // Filter: only jobs whose available_at has passed + $available = array_filter($this->pending, fn ($j) => $j['available_at'] <= time()); + + // Sort: highest priority first + usort($available, fn ($a, $b) => $b['priority'] - $a['priority']); + + // Limit: return at most $limit jobs + return array_slice($available, 0, $limit); + } + + /** + * Store a job in the pending queue. + * + * @param string $id Unique identifier for the job (generated by Queue::dispatch) + * @param string $payload The serialized (and possibly encrypted) job data. + * The storage should treat this as an opaque string. + * @param int $priority Higher number = processed first. Default is 0. + * @param int $availableAt Unix timestamp when the job becomes available. + * Jobs with availableAt in the future are skipped by pop(). + */ + public function push(string $id, string $payload, int $priority = 0, int $availableAt = 0): void { + $this->pending[$id] = [ + 'id' => $id, + 'payload' => $payload, + 'priority' => $priority, + 'attempts' => 0, + 'available_at' => $availableAt > 0 ? $availableAt : time(), + ]; + } + + /** + * Move a failed job back to the pending queue for reprocessing. + * Called by Queue::retry() when a developer wants to retry a failed job. + * + * @param string $id The job identifier. + */ + public function retry(string $id): void { + if (isset($this->failed[$id])) { + $data = $this->failed[$id]; + unset($this->failed[$id], $data['reason']); + $data['available_at'] = time(); + $this->pending[$id] = $data; + } + } + + /** + * Update the attempt count for a pending job. + * Called by Queue::process() before re-queuing a failed job for retry. + * + * @param string $id The job identifier. + * @param int $attempts The updated attempt count. + */ + public function setAttempts(string $id, int $attempts): void { + if (isset($this->pending[$id])) { + $this->pending[$id]['attempts'] = $attempts; + } + } +} + +// --- Usage --- + +/** + * A simple job that prints a message. + */ +class PrintJob implements Job { + public function __construct(private string $message) { + } + + public function getMaxAttempts(): int { + return 3; + } + + public function getRetryDelaySeconds(): int { + return 5; + } + + public function handle(): void { + echo "Processing: {$this->message}\n"; + } +} + +// Create a queue using our custom in-memory storage. +// In production, you'd pass a DatabaseQueueStorage or RedisQueueStorage here. +$queue = new Queue(new InMemoryQueueStorage()); + +// Dispatch jobs with different priorities. +// The queue stores them but does NOT execute them yet. +$queue->dispatch(new PrintJob('First task')); +$queue->dispatch(new PrintJob('High priority task'), priority: 10); +$queue->dispatch(new PrintJob('Normal task')); + +echo "Pending: ".$queue->getPendingCount()."\n\n"; + +// Process all pending jobs. +// They execute in priority order: high priority first. +$processed = $queue->process(); +echo "\nProcessed: $processed jobs\n"; +echo "Remaining: ".$queue->getPendingCount()."\n"; diff --git a/tests/QueueFacadeTest.php b/tests/QueueFacadeTest.php index 42008cd..6397a87 100644 --- a/tests/QueueFacadeTest.php +++ b/tests/QueueFacadeTest.php @@ -75,7 +75,7 @@ public function testFacadeRetry() { QueueFacade::process(); $failed = QueueFacade::getFailed(); - QueueFacade::retry($failed[0]['id']); + QueueFacade::retry($failed[0]->getId()); $this->assertEquals(1, QueueFacade::getPendingCount()); $this->removeDir($dir); diff --git a/tests/QueueTest.php b/tests/QueueTest.php index 7f5bee7..d55bf32 100644 --- a/tests/QueueTest.php +++ b/tests/QueueTest.php @@ -6,6 +6,7 @@ use WebFiori\Queue\FileQueueStorage; use WebFiori\Queue\Job; use WebFiori\Queue\Queue; +use WebFiori\Queue\QueuedJob; use WebFiori\Queue\QueueFacade; class SuccessJob implements Job { @@ -120,7 +121,7 @@ public function testFailingJobMovesToFailed() { $failed = $this->queue->getFailed(); $this->assertCount(1, $failed); - $this->assertStringContainsString('Job failed', $failed[0]['reason']); + $this->assertStringContainsString('Job failed', $failed[0]->getFailReason()); } /** * @test @@ -133,7 +134,7 @@ public function testRetryMovesFailedToPending() { $failed = $this->queue->getFailed(); $this->assertCount(1, $failed); - $this->queue->retry($failed[0]['id']); + $this->queue->retry($failed[0]->getId()); $this->assertEquals(1, $this->queue->getPendingCount()); $this->assertCount(0, $this->queue->getFailed()); } @@ -160,9 +161,9 @@ public function testPriorityOrdering() { $storage = $this->queue->getStorage(); $jobs = $storage->pop(3); - $this->assertEquals(10, $jobs[0]['priority']); - $this->assertEquals(5, $jobs[1]['priority']); - $this->assertEquals(1, $jobs[2]['priority']); + $this->assertEquals(10, $jobs[0]->getPriority()); + $this->assertEquals(5, $jobs[1]->getPriority()); + $this->assertEquals(1, $jobs[2]->getPriority()); } /** * @test