diff --git a/.gitattributes b/.gitattributes index 6b03c45..b99c30d 100644 --- a/.gitattributes +++ b/.gitattributes @@ -1,3 +1,11 @@ +* text=auto eol=lf +*.php text eol=lf +*.md text eol=lf +*.json text eol=lf +*.xml text eol=lf +*.yml text eol=lf +*.yaml text eol=lf + /tests export-ignore /examples export-ignore /CHANGELOG.md export-ignore diff --git a/README.md b/README.md index bcda79f..2c66661 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,28 @@ $queue->retry($failed[0]['id']); $queue->flush(); ``` +### Inspecting Pending Jobs + +```php +// 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) @@ -133,6 +155,7 @@ $queue->flush(); | `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 | @@ -151,6 +174,14 @@ $queue->flush(); | `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()`. diff --git a/WebFiori/Queue/FileQueueStorage.php b/WebFiori/Queue/FileQueueStorage.php index d8b0aec..1ae4bc6 100644 --- a/WebFiori/Queue/FileQueueStorage.php +++ b/WebFiori/Queue/FileQueueStorage.php @@ -17,7 +17,7 @@ * Stores each job as a JSON file. Pending jobs are in the 'pending' subdirectory, * failed jobs in the 'failed' subdirectory. */ -class FileQueueStorage implements QueueStorage { +class FileQueueStorage implements ListableQueueStorage { private string $baseDir; /** @@ -31,7 +31,7 @@ public function __construct(string $baseDir) { $this->ensureDir($this->getFailedDir()); } /** - * {@inheritDoc} + * @see QueueStorage */ public function flush(): void { $files = glob($this->getFailedDir().DIRECTORY_SEPARATOR.'*.json'); @@ -49,7 +49,7 @@ public function getBaseDir(): string { return $this->baseDir; } /** - * {@inheritDoc} + * @see QueueStorage */ public function getFailed(): array { $files = glob($this->getFailedDir().DIRECTORY_SEPARATOR.'*.json'); @@ -74,13 +74,52 @@ public function getFailed(): array { return $failed; } /** - * {@inheritDoc} + * @see QueueStorage */ public function getPendingCount(): int { return count(glob($this->getPendingDir().DIRECTORY_SEPARATOR.'*.json')); } /** - * {@inheritDoc} + * Returns all pending jobs, including those not yet available due to delay. + * + * Jobs are sorted by priority descending, then createdAt ascending. + * This method does not remove jobs from the queue. + * + * @return QueuedJob[] Array of all pending queued jobs. + */ + public function getPending(): array { + $files = glob($this->getPendingDir().DIRECTORY_SEPARATOR.'*.json'); + $jobs = []; + + foreach ($files as $file) { + $data = json_decode(file_get_contents($file), true); + + if ($data === null) { + continue; + } + + $jobs[] = new QueuedJob( + $data['id'], + $data['payload'] ?? '', + $data['priority'] ?? 0, + $data['attempts'] ?? 0, + $data['available_at'] ?? 0, + $data['created_at'] ?? 0 + ); + } + + usort($jobs, function (QueuedJob $a, QueuedJob $b) { + if ($a->getPriority() !== $b->getPriority()) { + return $b->getPriority() - $a->getPriority(); + } + + return $a->getCreatedAt() - $b->getCreatedAt(); + }); + + return $jobs; + } + /** + * @see QueueStorage */ public function markComplete(string $id): void { $file = $this->getPendingDir().DIRECTORY_SEPARATOR.$id.'.json'; @@ -90,7 +129,7 @@ public function markComplete(string $id): void { } } /** - * {@inheritDoc} + * @see QueueStorage */ public function markFailed(QueuedJob $job): void { $pendingFile = $this->getPendingDir().DIRECTORY_SEPARATOR.$job->getId().'.json'; @@ -114,7 +153,7 @@ public function markFailed(QueuedJob $job): void { ); } /** - * {@inheritDoc} + * @see QueueStorage */ public function pop(int $limit = 10): array { $files = glob($this->getPendingDir().DIRECTORY_SEPARATOR.'*.json'); @@ -153,7 +192,7 @@ public function pop(int $limit = 10): array { return array_slice($jobs, 0, $limit); } /** - * {@inheritDoc} + * @see QueueStorage */ public function push(QueuedJob $job): void { $data = [ @@ -171,7 +210,7 @@ public function push(QueuedJob $job): void { ); } /** - * {@inheritDoc} + * @see QueueStorage */ public function retry(string $id): void { $failedFile = $this->getFailedDir().DIRECTORY_SEPARATOR.$id.'.json'; diff --git a/WebFiori/Queue/ListableQueueStorage.php b/WebFiori/Queue/ListableQueueStorage.php new file mode 100644 index 0000000..ac75d31 --- /dev/null +++ b/WebFiori/Queue/ListableQueueStorage.php @@ -0,0 +1,36 @@ +storage->getPendingCount(); } + /** + * Returns all pending jobs, including delayed ones not yet available. + * + * This method requires a storage backend that implements ListableQueueStorage. + * If the backend does not support listing, a LogicException is thrown. + * + * @return QueuedJob[] Array of all pending queued jobs. + * + * @throws \LogicException If the storage backend does not implement ListableQueueStorage. + */ + public function getPending(): array { + if (!($this->storage instanceof ListableQueueStorage)) { + throw new \LogicException( + 'The configured storage backend does not support listing pending jobs. ' + . 'Use a ListableQueueStorage implementation (e.g. FileQueueStorage).' + ); + } + + return $this->storage->getPending(); + } /** * Returns the storage backend. * diff --git a/WebFiori/Queue/QueueFacade.php b/WebFiori/Queue/QueueFacade.php index b331c28..6be2263 100644 --- a/WebFiori/Queue/QueueFacade.php +++ b/WebFiori/Queue/QueueFacade.php @@ -56,6 +56,20 @@ public static function getInstance(): Queue { public static function getPendingCount(): int { return self::getInstance()->getPendingCount(); } + /** + * Returns all pending jobs, including delayed ones not yet available. + * + * Requires the queue's storage backend to implement ListableQueueStorage. + * + * @return QueuedJob[] Array of all pending queued jobs. + * + * @throws \LogicException If the storage backend does not support listing. + * + * @see Queue::getPending() + */ + public static function getPending(): array { + return self::getInstance()->getPending(); + } /** * @see Queue::process() */ diff --git a/examples/01-basic-queue.php b/examples/01-basic-queue.php index 2d32b42..2a0c787 100644 --- a/examples/01-basic-queue.php +++ b/examples/01-basic-queue.php @@ -86,7 +86,25 @@ public function handle(): void { echo "Pending jobs: ".$queue->getPendingCount()."\n\n"; -// --- Step 3: Process pending jobs --- +// --- Step 3: Inspect pending jobs --- +// getPending() returns all pending jobs including delayed ones. +// It does NOT remove them from the queue — it's read-only. +// Jobs are sorted by priority (highest first), then creation time. +echo "--- Pending Jobs ---\n"; +$pending = $queue->getPending(); + +foreach ($pending as $job) { + echo sprintf( + " [%s] priority=%d, available=%s\n", + substr($job->getId(), 0, 8).'...', + $job->getPriority(), + $job->isAvailable() ? 'now' : date('H:i:s', $job->getAvailableAt()) + ); +} + +echo "\n"; + +// --- Step 4: 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. diff --git a/tests/QueueFacadeTest.php b/tests/QueueFacadeTest.php index 6397a87..1ea19f7 100644 --- a/tests/QueueFacadeTest.php +++ b/tests/QueueFacadeTest.php @@ -89,6 +89,29 @@ public function testResetCreatesNewInstance() { $second = QueueFacade::getInstance(); $this->assertNotSame($first, $second); } + /** + * @test + */ + public function testFacadeGetPending() { + QueueFacade::reset(); + $storageDir = sys_get_temp_dir().DIRECTORY_SEPARATOR.'wf_facade_getpending_test'; + $queue = new \WebFiori\Queue\Queue(new \WebFiori\Queue\FileQueueStorage($storageDir)); + QueueFacade::setInstance($queue); + + QueueFacade::dispatch(new \WebFiori\Queue\Tests\SuccessJob()); + QueueFacade::dispatch(new \WebFiori\Queue\Tests\SuccessJob()); + + $pending = QueueFacade::getPending(); + $this->assertCount(2, $pending); + + foreach ($pending as $job) { + $this->assertInstanceOf(\WebFiori\Queue\QueuedJob::class, $job); + } + + // Cleanup + QueueFacade::reset(); + $this->removeDir($storageDir); + } private function removeDir(string $dir): void { if (!is_dir($dir)) { diff --git a/tests/QueueTest.php b/tests/QueueTest.php index 8b69298..b20676e 100644 --- a/tests/QueueTest.php +++ b/tests/QueueTest.php @@ -354,4 +354,86 @@ public function testFlushOnlyRemovesFailed() { $this->queue->flush(); $this->assertCount(0, $this->queue->getFailed()); } + /** + * @test + */ + public function testGetPendingReturnsAllJobs() { + $this->queue->dispatch(new SuccessJob()); + $this->queue->dispatch(new SuccessJob()); + $this->queue->dispatch(new SuccessJob()); + + $pending = $this->queue->getPending(); + $this->assertCount(3, $pending); + + foreach ($pending as $job) { + $this->assertInstanceOf(QueuedJob::class, $job); + } + } + /** + * @test + */ + public function testGetPendingIncludesDelayedJobs() { + $this->queue->dispatch(new SuccessJob(), 0, 0); + $this->queue->dispatch(new SuccessJob(), 0, 3600); // delayed 1 hour + + $pending = $this->queue->getPending(); + $this->assertCount(2, $pending); + } + /** + * @test + */ + public function testGetPendingReturnsEmptyWhenNoJobs() { + $pending = $this->queue->getPending(); + $this->assertCount(0, $pending); + $this->assertSame([], $pending); + } + /** + * @test + */ + public function testGetPendingDoesNotRemoveJobs() { + $this->queue->dispatch(new SuccessJob()); + $this->queue->dispatch(new SuccessJob()); + + $this->queue->getPending(); + $this->assertEquals(2, $this->queue->getPendingCount()); + + // Call again — still 2 + $this->queue->getPending(); + $this->assertEquals(2, $this->queue->getPendingCount()); + } + /** + * @test + */ + public function testGetPendingSortedByPriority() { + $this->queue->dispatch(new LowPriorityJob(), 1); + $this->queue->dispatch(new HighPriorityJob(), 10); + $this->queue->dispatch(new CountingJob(), 5); + + $pending = $this->queue->getPending(); + $this->assertCount(3, $pending); + $this->assertEquals(10, $pending[0]->getPriority()); + $this->assertEquals(5, $pending[1]->getPriority()); + $this->assertEquals(1, $pending[2]->getPriority()); + } + /** + * @test + */ + public function testGetPendingThrowsOnNonListableStorage() { + $mockStorage = new class implements \WebFiori\Queue\QueueStorage { + public function flush(): void {} + public function getFailed(): array { return []; } + public function getPendingCount(): int { return 0; } + public function markComplete(string $id): void {} + public function markFailed(QueuedJob $job): void {} + public function pop(int $limit = 10): array { return []; } + public function push(QueuedJob $job): void {} + public function retry(string $id): void {} + }; + + $queue = new Queue($mockStorage); + + $this->expectException(\LogicException::class); + $this->expectExceptionMessage('does not support listing pending jobs'); + $queue->getPending(); + } }