You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Commit 73aed51
Browse filesBrowse the repository at this point in the historyBrowse files
Copy file name to clipboardExpand all lines: src/Adapter.php
+43-36Lines changed: 43 additions & 36 deletions
Original file line number
Diff line number
Diff line change
@@ -19,6 +19,13 @@
19
19
useYiisoft\Mutex\MutexFactoryInterface;
20
20
useYiisoft\Mutex\MutexInterface;
21
21
useYiisoft\Queue\Provider\QueueProviderInterface;
22
+
useException;
23
+
24
+
usefunctionis_array;
25
+
usefunctionis_resource;
26
+
usefunctionis_string;
27
+
28
+
useconstSORT_ASC;
22
29
23
30
finalclass Adapter implements AdapterInterface
24
31
{
@@ -39,6 +46,11 @@ final class Adapter implements AdapterInterface
39
46
*/
40
47
public$deleteReleased = true;
41
48
49
+
/**
50
+
* @var int reserve time
51
+
*/
52
+
private$reserveTime = 0;
53
+
42
54
publicfunction__construct(
43
55
privateConnectionInterface$db,
44
56
privateMessageSerializerInterface$serializer,
@@ -123,17 +135,42 @@ public function withChannel(BackedEnum|string $channel): self
123
135
return$new;
124
136
}
125
137
138
+
/**
139
+
* Listens queue and runs each job.
140
+
*
141
+
* @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages
142
+
* @param bool $repeat whether to continue listening when queue is empty.
143
+
* @param non-negative-int $timeout number of seconds to sleep before next iteration.
if ($handlerCallback($this->serializer->unserialize($payload['job']))) {
150
+
$this->release($payload);
151
+
}
152
+
continue;
153
+
}
154
+
if (!$repeat) {
155
+
break;
156
+
}
157
+
if ($timeout > 0) {
158
+
sleep($timeout);
159
+
}
160
+
}
161
+
}
162
+
126
163
/**
127
164
* Takes one message from waiting list and reserves it for handling.
128
165
*
129
-
* @throws \Exception in case it hasn't waited the lock
166
+
* @throws Exception in case it hasn't waited the lock
130
167
* @return array|null payload
131
168
*/
132
-
protectedfunctionreserve(): array|null
169
+
protectedfunctionreserve(): ?array
133
170
{
134
171
// TWK TODO what is useMaster in Yii3 return $this->db->useMaster(function () {
135
172
if (!$this->mutex->acquire($this->mutexTimeout)) {
136
-
thrownew\Exception('Has not waited the lock.');
173
+
thrownewException('Has not waited the lock.');
137
174
}
138
175
139
176
try {
@@ -182,13 +219,13 @@ protected function release($payload): void
182
219
if ($this->deleteReleased) {
183
220
$this->db->createCommand()->delete(
184
221
$this->tableName,
185
-
['id' => $payload['id']]
222
+
['id' => $payload['id']],
186
223
)->execute();
187
224
} else {
188
225
$this->db->createCommand()->update(
189
226
$this->tableName,
190
227
['done_at' => time()],
191
-
['id' => $payload['id']]
228
+
['id' => $payload['id']],
192
229
)->execute();
193
230
}
194
231
}
@@ -205,38 +242,8 @@ private function moveExpired(): void
205
242
['reserved_at' => null],
206
243
'[[reserved_at]] < :time - [[ttr]] and [[reserved_at]] is not null and [[done_at]] is null',
207
244
null,
208
-
[':time' => $this->reserveTime]
245
+
[':time' => $this->reserveTime],
209
246
)->execute();
210
247
}
211
248
}
212
-
213
-
/**
214
-
* @var int reserve time
215
-
*/
216
-
private$reserveTime = 0;
217
-
218
-
/**
219
-
* Listens queue and runs each job.
220
-
*
221
-
* @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages
222
-
* @param bool $repeat whether to continue listening when queue is empty.
223
-
* @param non-negative-int $timeout number of seconds to sleep before next iteration.
0 commit comments