queue/src/Queue/Client.php at 88f2b402cd421e771ec395a8c17e1306f9658d56 · utopia-php/queue · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
<?php

namespace Utopia\Queue;

class Client
{
    protected string $queue;
    protected string $namespace;
    protected Connection $connection;
    public function __construct(string $queue, Connection $connection, string $namespace = 'utopia-queue')
    {
        $this->queue = $queue;
        $this->namespace = $namespace;
        $this->connection = $connection;
    }

    public function enqueue(array $payload): bool
    {
        $payload = [
            'pid' => \uniqid(more_entropy: true),
            'queue' => $this->queue,
            'timestamp' => time(),
            'payload' => $payload
        ];

        return $this->connection->leftPushArray("{$this->namespace}.queue.{$this->queue}", $payload);
    }

    /**
     * Take all jobs from the failed queue and re-enqueue them.
     * @param int|null $limit The amount of jobs to retry
     */
    public function retry(int $limit = null): void
    {
        $start = \time();
        $processed = 0;

        while (true) {
            $pid = $this->connection->rightPop("{$this->namespace}.failed.{$this->queue}", 5);

            // No more jobs to retry
            if ($pid === false) {
                break;
            }

            $job = $this->getJob($pid);

            // Job doesn't exist
            if ($job === false) {
                break;
            }

            // Job was already retried
            if ($job->getTimestamp() >= $start) {
                break;
            }

            // We're reached the max amount of jobs to retry
            if ($limit !== null && $processed >= $limit) {
                break;
            }

            $this->enqueue($job->getPayload());
            $processed++;
        }
    }

    public function getJob(string $pid): Message|false
    {
        $value = $this->connection->get("{$this->namespace}.jobs.{$this->queue}.{$pid}");

        if ($value === false) {
            return false;
        }

        $job = json_decode($value, true);

        return new Message($job);
    }

    public function listJobs(int $total = 50, int $offset = 0): array
    {
        return $this->connection->listRange("{$this->namespace}.queue.{$this->queue}", $total, $offset);
    }

    public function getQueueSize(): int
    {
        return $this->connection->listSize("{$this->namespace}.queue.{$this->queue}");
    }

    public function countTotalJobs(): int
    {
        return (int)($this->connection->get("{$this->namespace}.stats.{$this->queue}.total") ?? 0);
    }

    public function countSuccessfulJobs(): int
    {
        return (int)($this->connection->get("{$this->namespace}.stats.{$this->queue}.success") ?? 0);
    }

    public function countFailedJobs(): int
    {
        return (int)($this->connection->get("{$this->namespace}.stats.{$this->queue}.failed") ?? 0);
    }

    public function countProcessingJobs(): int
    {
        return (int)($this->connection->get("{$this->namespace}.stats.{$this->queue}.processing") ?? 0);
    }

    public function resetStats(): void
    {
        $this->connection->set("{$this->namespace}.stats.{$this->queue}.total", 0);
        $this->connection->set("{$this->namespace}.stats.{$this->queue}.success", 0);
        $this->connection->set("{$this->namespace}.stats.{$this->queue}.failed", 0);
        $this->connection->set("{$this->namespace}.stats.{$this->queue}.processing", 0);
    }
}