1 | <?php |
||
22 | class BatchQueue extends DatabaseQueue |
||
23 | { |
||
24 | /** |
||
25 | * The AWS Batch client. |
||
26 | * |
||
27 | * @var BatchClient |
||
28 | */ |
||
29 | protected $batch; |
||
30 | |||
31 | protected $jobDefinition; |
||
32 | |||
33 | 5 | public function __construct( |
|
34 | Connection $database, |
||
35 | $table, |
||
36 | $default, |
||
37 | $expire, |
||
38 | $jobDefinition, |
||
39 | BatchClient $batch |
||
40 | ) { |
||
41 | 5 | $this->jobDefinition = $jobDefinition; |
|
42 | 5 | $this->batch = $batch; |
|
43 | 5 | parent::__construct($database, $table, $default, $expire); |
|
44 | 5 | } |
|
45 | |||
46 | 2 | public function push($job, $data = '', $queue = null) |
|
47 | { |
||
48 | 2 | $queue = $this->getQueue($queue); |
|
49 | 2 | $payload = $this->createPayload($job, $queue, $data); |
|
50 | 2 | return $this->pushToBatch($queue, $payload, $this->getBatchDisplayName($job)); |
|
51 | } |
||
52 | |||
53 | public function pushRaw($payload, $queue = null, array $options = []) |
||
54 | { |
||
55 | return $this->pushToBatch($queue, $payload, 'raw-job'); |
||
56 | } |
||
57 | |||
58 | /** |
||
59 | * Get the display name for the given job. |
||
60 | * |
||
61 | * @param mixed $job |
||
62 | * @return string |
||
63 | */ |
||
64 | 2 | protected function getBatchDisplayName($job) |
|
65 | { |
||
66 | 2 | if (is_object($job)) { |
|
67 | 1 | return method_exists($job, 'displayName') |
|
68 | 1 | ? $job->displayName() : str_replace('\\', '_', (string) get_class($job)); |
|
69 | } else { |
||
70 | 1 | return is_string($job) ? explode('@', $job)[0] : null; |
|
71 | } |
||
72 | } |
||
73 | |||
74 | /** |
||
75 | * Push a raw payload to the database, then to AWS Batch, with a given delay. |
||
76 | * |
||
77 | * @param string|null $queue |
||
78 | * @param string $payload |
||
79 | * @param string $jobName |
||
80 | * |
||
81 | * @return int |
||
82 | */ |
||
83 | 2 | protected function pushToBatch($queue, $payload, $jobName) |
|
84 | { |
||
85 | 2 | $jobId = $this->pushToDatabase($queue, $payload); |
|
86 | |||
87 | 2 | $this->batch->submitJob([ |
|
88 | 2 | 'jobDefinition' => $this->jobDefinition, |
|
89 | 2 | 'jobName' => $jobName, |
|
90 | 2 | 'jobQueue' => $this->getQueue($queue), |
|
91 | 'parameters' => [ |
||
92 | 2 | 'jobId' => $jobId, |
|
93 | ] |
||
94 | ]); |
||
95 | |||
96 | 2 | return $jobId; |
|
97 | } |
||
98 | |||
99 | public function getJobById($id) |
||
100 | { |
||
101 | $this->database->beginTransaction(); |
||
102 | |||
103 | $job = $this->database->table($this->table) |
||
104 | ->lockForUpdate() |
||
105 | ->where('id', $id) |
||
106 | ->first(); |
||
107 | |||
108 | |||
109 | if (!isset($job)) { |
||
110 | throw new JobNotFoundException('Could not find the job'); |
||
111 | } |
||
112 | |||
113 | $job = new DatabaseJobRecord($job); |
||
114 | |||
115 | return $this->marshalJob($job->queue, $job); |
||
116 | } |
||
117 | |||
118 | protected function marshalJob($queue, $job) |
||
119 | { |
||
120 | $job = $this->markJobAsReserved($job); |
||
121 | |||
122 | $this->database->commit(); |
||
123 | |||
124 | return new BatchJob( |
||
125 | $this->container, |
||
126 | $this, |
||
127 | $job, |
||
|
|||
128 | $this->connectionName, |
||
129 | $queue |
||
130 | ); |
||
131 | } |
||
132 | |||
133 | /** |
||
134 | * Release the job, without deleting first from the Queue |
||
135 | * |
||
136 | * @param string $queue |
||
137 | * @param \StdClass $job |
||
138 | * @param int $delay |
||
139 | * |
||
140 | * @return int |
||
141 | * @throws UnsupportedException |
||
142 | */ |
||
143 | 2 | public function release($queue, $job, $delay) |
|
152 | ]); |
||
153 | } |
||
154 | |||
155 | /* |
||
156 | * Function pop() from parent class need to be executed on Laravel 6.x |
||
157 | */ |
||
158 | // public function pop($queue = null) |
||
159 | // { |
||
160 | // throw new UnsupportedException('The BatchQueue does not support running via a regular worker. ' . |
||
161 | // 'Instead, you should use the queue:batch-work command with a job id.'); |
||
162 | // } |
||
163 | |||
164 | public function bulk($jobs, $data = '', $queue = null) |
||
168 | } |
||
169 | |||
170 | 1 | public function later($delay, $job, $data = '', $queue = null) |
|
171 | { |
||
172 | 1 | throw new UnsupportedException('The BatchQueue does not support the later() operation.'); |
|
173 | } |
||
175 |