job queue and workers and loggers!

1 files changed, 86 insertions, 0 deletions

diff --git a/Digitigrade/JobQueue.php b/Digitigrade/JobQueue.php
new file mode 100644
index 0000000..9f95584
--- /dev/null
+++ b/Digitigrade/JobQueue.php
@@ -0,0 +1,86 @@
+<?php
+namespace Digitigrade;
+
+use Pheanstalk\Contract\JobIdInterface;
+use Pheanstalk\Pheanstalk;
+use Pheanstalk\Values\TubeName;
+
+class JobQueue extends Singleton {
+    private Pheanstalk $conn;
+
+    private const PRIO_URGENT = 10;
+    private const PRIO_DELAYED = 20;
+    private const PRIO_RETRY = 30;
+    private const PRIO_NORMAL = 50;
+    private const PRIO_LAZY = 100;
+
+    private const TIMEOUT = 60; // seconds
+    private const RETRY_DELAY = 5; // seconds, to the power of the number of tries
+
+    public function __construct() {
+        $config = GlobalConfig::getInstance();
+        $this->conn = Pheanstalk::create($config->getQueueHost(), $config->getQueuePort());
+        $this->conn->useTube(new TubeName($config->getQueueChannel()));
+    }
+
+    public function subscribe() {
+        $this->conn->watch(new TubeName(GlobalConfig::getInstance()->getQueueChannel()));
+    }
+
+    private function submit(Job $job, int $priority, int $delay = 0): JobIdInterface {
+        return $this->conn->put(json_encode($job), $priority, $delay, self::TIMEOUT);
+    }
+
+    public function submitNormal(Job $job) {
+        $this->submit($job, self::PRIO_NORMAL);
+    }
+
+    public function submitUrgent(Job $job) {
+        $this->submit($job, self::PRIO_URGENT);
+    }
+
+    public function submitLazy(Job $job) {
+        $this->submit($job, self::PRIO_LAZY);
+    }
+
+    public function submitDelayed(Job $job, int $delaySecs) {
+        $this->submit($job, self::PRIO_NORMAL, $delaySecs);
+    }
+
+    public function submitDelayedUntil(Job $job, \DateTimeInterface $until) {
+        $delay = $until->getTimestamp() - (new \DateTimeImmutable())->getTimestamp();
+        if ($delay < 0)
+            throw new \InvalidArgumentException('"until" date is in the past!');
+        $this->submitDelayed($job, $delay);
+    }
+
+    public function runNext() {
+        $beanJob = $this->conn->reserve();
+        $job = Job::fromJson($beanJob->getData());
+        $job->remainingTries--;
+        $job->doneTries++;
+        $log = Logger::getInstance();
+        $log->info('Running job ' . $beanJob->getId());
+        try {
+            $job->run();
+            $this->conn->delete($beanJob);
+            $log->info('Job ' . $beanJob->getId() . ' completed successfully');
+        } catch (\Exception $e) {
+            if ($job->remainingTries <= 0) {
+                $this->conn->delete($beanJob);
+                $log->error(
+                    'Ran out of retries trying to run job ' . $beanJob->getId() . ': ' . $beanJob->getData()
+                    . "\n" . $e->getMessage()
+                );
+            } else {
+                $this->conn->delete($beanJob);
+                $id = $this->submit($job, self::PRIO_RETRY, pow(self::RETRY_DELAY, $job->doneTries));
+                $log->warning(
+                    'A job of type ' . $job::class . ' (id ' . $beanJob->getId() . ') failed. '
+                    . 'Resubmitted as ' . $id->getId() . '; '
+                    . $job->remainingTries . ' retries remaining'
+                );
+            }
+        }
+    }
+}
\ No newline at end of file