disk-usage-insights / vendor / mgleis / php-sqlite-job-queue / src / Queue.php

Queue.php in Disk Usage Insights 1.9, at vendor/mgleis/php-sqlite-job-queue/src/Queue.php

132 lines 3.9 KB

Raw Download Zip

1	<?php
2	declare(strict_types = 1);
3
4	namespace Mgleis\PhpSqliteJobQueue;
5
6	class Queue {
7
8	public \PDO $db;
9	private string $table;
10
11	public function __construct(string $filename, string $table = 'jobs') {
12	$this->db = new \PDO("sqlite:$filename");
13	$this->db->setAttribute(\PDO::ATTR_ERRMODE, \PDO::ERRMODE_EXCEPTION);
14	$this->table = $this->db->quote($table);
15	$this->initializeDatabase();
16	}
17
18	private function initializeDatabase() {
19	$this->db->exec(sprintf("
20	CREATE TABLE IF NOT EXISTS %s (
21	id INTEGER PRIMARY KEY AUTOINCREMENT,
22	payload TEXT NOT NULL,
23	status TEXT NOT NULL DEFAULT 'queued',
24	reserved_at TIMESTAMP DEFAULT NULL
25	);
26	", $this->table));
27	}
28
29	public function push($payload) {
30	$strPayload = json_encode($payload, JSON_THROW_ON_ERROR);
31
32	$stmt = $this->db->prepare(sprintf("INSERT INTO %s (payload, status) VALUES (:payload, 'queued')", $this->table));
33	$stmt->bindValue(':payload', $strPayload);
34	$stmt->execute();
35	}
36
37	public function pop(): ?Job {
38	try {
39
40	// find the next job
41	$stmt = $this->db->prepare(sprintf("
42	SELECT id, payload
43	FROM %s
44	WHERE status = 'queued'
45	ORDER BY id ASC
46	LIMIT 1
47	", $this->table));
48	$stmt->execute();
49	$jobData = $stmt->fetch(\PDO::FETCH_ASSOC);
50
51	if (!$jobData) {
52	return null;
53	}
54
55	// mark the job as "working"
56	$updateStmt = $this->db->prepare(sprintf("
57	UPDATE %s
58	SET status = 'working', reserved_at = CURRENT_TIMESTAMP
59	WHERE id = :id AND status = 'queued'
60	", $this->table));
61	$updateStmt->bindValue(':id', (int) $jobData['id'], \PDO::PARAM_INT);
62	$updateStmt->execute();
63
64	return new Job(
65	(int) $jobData['id'],
66	json_decode($jobData['payload'], true, JSON_THROW_ON_ERROR)
67	);
68	} catch (\Throwable $t) {
69	$this->throwIfNotALockError($t);
70	return null;
71	}
72	}
73
74	public function peek(): ?Job {
75	try {
76	$stmt = $this->db->prepare(sprintf("
77	SELECT id, payload
78	FROM %s
79	ORDER BY id ASC
80	LIMIT 1
81	", $this->table));
82	$stmt->execute();
83	$jobData = $stmt->fetch(\PDO::FETCH_ASSOC);
84
85	if (!$jobData) {
86	return null;
87	}
88
89	return new Job(
90	(int) $jobData['id'],
91	json_decode($jobData['payload'], true, JSON_THROW_ON_ERROR)
92	);
93	} catch (\Throwable $t) {
94	$this->throwIfNotALockError($t);
95	return null;
96	}
97	}
98
99	private function throwIfNotALockError(\Throwable $t) {
100	if ($t instanceof \PDOException
101	&& str_contains($t->getMessage(), 'database is locked')) {
102	return;
103	}
104	throw $t;
105	}
106
107	public function done(Job $job) {
108	$updateStmt = $this->db->prepare(sprintf("
109	DELETE FROM %s
110	WHERE id = :id
111	", $this->table));
112	$updateStmt->bindValue(':id', $job->id, \PDO::PARAM_INT);
113	$updateStmt->execute();
114	}
115
116	public function error(Job $job) {
117	$updateStmt = $this->db->prepare(sprintf("
118	UPDATE %s
119	SET status = 'failed'
120	WHERE id = :id
121	", $this->table));
122	$updateStmt->bindValue(':id', $job->id, \PDO::PARAM_INT);
123	$updateStmt->execute();
124	}
125
126	public function size(): int {
127	$stmt = $this->db->query(sprintf("SELECT COUNT(*) FROM %s WHERE status = 'queued'", $this->table));
128	return (int)$stmt->fetchColumn();
129	}
130
131	}
132