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

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

72 lines 2.1 KB

Raw Download Zip

1	<?php
2	declare(strict_types = 1);
3
4	namespace Mgleis\PhpSqliteJobQueue;
5
6	class Worker {
7
8	private Queue $queue;
9	private $maxTotalRuntimeInSeconds = PHP_INT_MAX;
10	private $maxJobCount = PHP_INT_MAX;
11	private $sleepTimeOnEmptyQueueInMilliseconds = 0;
12	private $sleepTimeBetweenJobsInMilliseconds = 0;
13
14	public function __construct(Queue $queue) {
15	$this->queue = $queue;
16	}
17
18	public function process(\Closure $callback) {
19	$jobsExecuted = 0;
20	$startTime = time();
21
22	while (true) {
23
24	$queueSize = $this->queue->size();
25	if ($queueSize == 0) {
26	usleep($this->sleepTimeOnEmptyQueueInMilliseconds * 1000);
27	}
28
29	$elapsedTime = time() - $startTime;
30	if ($elapsedTime > $this->maxTotalRuntimeInSeconds) {
31	break;
32	}
33
34	if ($jobsExecuted >= $this->maxJobCount) {
35	break;
36	}
37
38	$job = $this->queue->pop();
39	if ($job !== null) {
40	try {
41	$callback($job);
42	} catch (\Throwable $e) {
43	$this->queue->error($job);
44	throw $e;
45	}
46	$this->queue->done($job);
47	$jobsExecuted++;
48	}
49	usleep($this->sleepTimeBetweenJobsInMilliseconds * 1000);
50	}
51	}
52
53	public function withSleepTimeOnEmptyQueueInMilliseconds(int $sleepTimeOnEmptyQueueInMilliseconds): self {
54	$this->sleepTimeOnEmptyQueueInMilliseconds = $sleepTimeOnEmptyQueueInMilliseconds;
55	return $this;
56	}
57	public function withSleepTimeBetweenJobsInMilliseconds(int $sleepTimeBetweenJobsInMilliseconds): self {
58	$this->sleepTimeBetweenJobsInMilliseconds = $sleepTimeBetweenJobsInMilliseconds;
59	return $this;
60	}
61	public function withMaxTotalRuntimeInSeconds(int $maxTotalRuntimeInSeconds): self {
62	$this->maxTotalRuntimeInSeconds = $maxTotalRuntimeInSeconds;
63	return $this;
64	}
65
66	public function withMaxJobCount(int $maxJobCount): self {
67	$this->maxJobCount = $maxJobCount;
68	return $this;
69	}
70
71	}
72