-
Notifications
You must be signed in to change notification settings - Fork 68
Expand file tree
/
Copy pathCli.php
More file actions
99 lines (87 loc) · 3.03 KB
/
Copy pathCli.php
File metadata and controls
99 lines (87 loc) · 3.03 KB
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
<?php
namespace PHPQueue;
use PHPQueue\Exception\Exception;
class Cli
{
public $queue_name;
public function __construct($options=array())
{
if ( !empty($options['queue']) ) {
$this->queue_name = $options['queue'];
}
}
public function add($payload=array())
{
fwrite(STDOUT, "===========================================================\n");
fwrite(STDOUT, "Adding Job...");
$status = false;
try {
$queue = Base::getQueue($this->queue_name);
$status = Base::addJob($queue, $payload);
fwrite(STDOUT, "Done.\n");
} catch (\Exception $ex) {
fwrite(STDOUT, sprintf("Error: %s\n", $ex->getMessage()));
throw $ex;
}
return $status;
}
public function peek()
{
$newJob = null;
$queue = Base::getQueue($this->queue_name);
try {
$newJob = Base::getJob($queue);
fwrite(STDOUT, "===========================================================\n");
fwrite(STDOUT, "Next Job:\n");
var_dump($newJob);
fwrite(STDOUT, "\nReleasing Job...\n");
$queue->releaseJob($newJob->job_id);
} catch (\Exception $ex) {
fwrite(STDOUT, "Error: " . $ex->getMessage() . "\n");
}
}
public function work()
{
$newJob = null;
$queue = Base::getQueue($this->queue_name);
try {
$newJob = Base::getJob($queue);
fwrite(STDOUT, "===========================================================\n");
fwrite(STDOUT, "Next Job:\n");
var_dump($newJob);
} catch (\Exception $ex) {
fwrite(STDOUT, "Error: " . $ex->getMessage() . "\n");
}
if (empty($newJob)) {
fwrite(STDOUT, "Notice: No Job found.\n");
return;
}
try {
if (empty($newJob->worker)) {
throw new Exception("No worker declared.");
}
if (is_string($newJob->worker)) {
$result_data = $this->processWorker($newJob->worker, $newJob);
} elseif (is_array($newJob->worker)) {
foreach ($newJob->worker as $worker_name) {
$result_data = $this->processWorker($worker_name, $newJob);
$newJob->data = $result_data;
}
}
fwrite(STDOUT, "Updating job... \n");
return Base::updateJob($queue, $newJob->job_id, $result_data);
} catch (\Exception $ex) {
fwrite(STDOUT, sprintf("\nError occured: %s\n", $ex->getMessage()));
$queue->releaseJob($newJob->job_id);
throw $ex;
}
}
protected function processWorker($worker_name, $new_job)
{
fwrite(STDOUT, sprintf("Running worker (%s) now... ", $worker_name));
$newWorker = Base::getWorker($worker_name);
Base::workJob($newWorker, $new_job);
fwrite(STDOUT, "Done.\n");
return $newWorker->result_data;
}
}