-
Notifications
You must be signed in to change notification settings - Fork 12
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #4 from michalsn/feat/redis
feat: redis
- Loading branch information
Showing
17 changed files
with
1,152 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,147 @@ | ||
| <?php | ||
|
|
||
| namespace Michalsn\CodeIgniterQueue\Handlers; | ||
|
|
||
| use CodeIgniter\Exceptions\CriticalError; | ||
| use CodeIgniter\I18n\Time; | ||
| use Exception; | ||
| use Michalsn\CodeIgniterQueue\Config\Queue as QueueConfig; | ||
| use Michalsn\CodeIgniterQueue\Entities\QueueJob; | ||
| use Michalsn\CodeIgniterQueue\Enums\Status; | ||
| use Michalsn\CodeIgniterQueue\Interfaces\QueueInterface; | ||
| use Michalsn\CodeIgniterQueue\Payload; | ||
| use Predis\Client; | ||
| use Throwable; | ||
|
|
||
| class PredisHandler extends BaseHandler implements QueueInterface | ||
| { | ||
| private readonly Client $predis; | ||
|
|
||
| public function __construct(protected QueueConfig $config) | ||
| { | ||
| try { | ||
| $this->predis = new Client($config->predis, ['prefix' => $config->predis['prefix']]); | ||
| $this->predis->time(); | ||
| } catch (Exception $e) { | ||
| throw new CriticalError('Queue: Predis connection refused (' . $e->getMessage() . ').'); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Add job to the queue. | ||
| */ | ||
| public function push(string $queue, string $job, array $data): bool | ||
| { | ||
| $this->validateJobAndPriority($queue, $job); | ||
|
|
||
| helper('text'); | ||
|
|
||
| $queueJob = new QueueJob([ | ||
| 'id' => random_string('numeric', 16), | ||
| 'queue' => $queue, | ||
| 'payload' => new Payload($job, $data), | ||
| 'priority' => $this->priority, | ||
| 'status' => Status::PENDING->value, | ||
| 'attempts' => 0, | ||
| 'available_at' => Time::now()->timestamp, | ||
| ]); | ||
|
|
||
| $result = $this->predis->zadd("queues:{$queue}:{$this->priority}", [json_encode($queueJob) => Time::now()->timestamp]); | ||
|
|
||
| $this->priority = null; | ||
|
|
||
| return $result > 0; | ||
| } | ||
|
|
||
| /** | ||
| * Get job from the queue. | ||
| */ | ||
| public function pop(string $queue, array $priorities): ?QueueJob | ||
| { | ||
| $tasks = []; | ||
| $now = Time::now()->timestamp; | ||
|
|
||
| foreach ($priorities as $priority) { | ||
| if ($tasks = $this->predis->zrangebyscore("queues:{$queue}:{$priority}", '-inf', $now, ['LIMIT' => [0, 1]])) { | ||
| if ($this->predis->zrem("queues:{$queue}:{$priority}", ...$tasks)) { | ||
| break; | ||
| } | ||
| $tasks = []; | ||
| } | ||
| } | ||
|
|
||
| if (empty($tasks[0])) { | ||
| return null; | ||
| } | ||
|
|
||
| $queueJob = new QueueJob(json_decode((string) $tasks[0], true)); | ||
|
|
||
| // Set the actual status as in DB. | ||
| $queueJob->status = Status::RESERVED->value; | ||
| $queueJob->syncOriginal(); | ||
|
|
||
| $this->predis->hset("queues:{$queue}::reserved", $queueJob->id, json_encode($queueJob)); | ||
|
|
||
| return $queueJob; | ||
| } | ||
|
|
||
| /** | ||
| * Schedule job for later | ||
| */ | ||
| public function later(QueueJob $queueJob, int $seconds): bool | ||
| { | ||
| $queueJob->status = Status::PENDING->value; | ||
| $queueJob->available_at = Time::now()->addSeconds($seconds)->timestamp; | ||
|
|
||
| if ($result = $this->predis->zadd("queues:{$queueJob->queue}:{$queueJob->priority}", [json_encode($queueJob) => $queueJob->available_at->timestamp])) { | ||
| $this->predis->hdel("queues:{$queueJob->queue}::reserved", $queueJob->id); | ||
| } | ||
|
|
||
| return $result > 0; | ||
| } | ||
|
|
||
| /** | ||
| * Move job to failed table or move and delete. | ||
| */ | ||
| public function failed(QueueJob $queueJob, Throwable $err, bool $keepJob): bool | ||
| { | ||
| if ($keepJob) { | ||
| $this->logFailed($queueJob, $err); | ||
| } | ||
|
|
||
| return (bool) $this->predis->hdel("queues:{$queueJob->queue}::reserved", $queueJob->id); | ||
| } | ||
|
|
||
| /** | ||
| * Change job status to DONE or delete it. | ||
| */ | ||
| public function done(QueueJob $queueJob, bool $keepJob): bool | ||
| { | ||
| if ($keepJob) { | ||
| $queueJob->status = Status::DONE->value; | ||
| $this->predis->lpush("queues:{$queueJob->queue}::done", [json_encode($queueJob)]); | ||
| } | ||
|
|
||
| return (bool) $this->predis->hdel("queues:{$queueJob->queue}::reserved", $queueJob->id); | ||
| } | ||
|
|
||
| /** | ||
| * Delete queue jobs | ||
| */ | ||
| public function clear(?string $queue = null): bool | ||
| { | ||
| if ($queue !== null) { | ||
| if ($keys = $this->predis->keys("queues:{$queue}:*")) { | ||
| return $this->predis->del($keys) > 0; | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
|
|
||
| if ($keys = $this->predis->keys('queues:*')) { | ||
| return $this->predis->del($keys) > 0; | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
| } |
Oops, something went wrong.