|
1
|
|
|
<?php |
|
2
|
|
|
declare(strict_types=1); |
|
3
|
|
|
|
|
4
|
|
|
namespace Genkgo\Mail\Queue; |
|
5
|
|
|
|
|
6
|
|
|
use Genkgo\Mail\Exception\EmptyQueueException; |
|
7
|
|
|
use Genkgo\Mail\Exception\QueueStoreException; |
|
8
|
|
|
use Genkgo\Mail\GenericMessage; |
|
9
|
|
|
use Genkgo\Mail\MessageInterface; |
|
10
|
|
|
use Predis\ClientInterface; |
|
11
|
|
|
use Predis\Connection\ConnectionException; |
|
12
|
|
|
|
|
13
|
|
|
final class RedisQueue implements QueueInterface, \Countable |
|
14
|
|
|
{ |
|
15
|
|
|
|
|
16
|
|
|
/** |
|
17
|
|
|
* @var ClientInterface |
|
18
|
|
|
*/ |
|
19
|
|
|
private $client; |
|
20
|
|
|
/** |
|
21
|
|
|
* @var string |
|
22
|
|
|
*/ |
|
23
|
|
|
private $key; |
|
24
|
|
|
|
|
25
|
|
|
/** |
|
26
|
|
|
* RedisQueue constructor. |
|
27
|
|
|
* @param ClientInterface $client |
|
28
|
|
|
* @param string $key |
|
29
|
|
|
*/ |
|
30
|
7 |
|
public function __construct(ClientInterface $client, string $key) |
|
31
|
|
|
{ |
|
32
|
7 |
|
$this->client = $client; |
|
33
|
7 |
|
$this->key = $key; |
|
34
|
7 |
|
} |
|
35
|
|
|
|
|
36
|
|
|
/** |
|
37
|
|
|
* @param MessageInterface $message |
|
38
|
|
|
* @throws QueueStoreException |
|
39
|
|
|
*/ |
|
40
|
4 |
|
public function store(MessageInterface $message): void |
|
41
|
|
|
{ |
|
42
|
|
|
try { |
|
43
|
4 |
|
$this->client->rpush($this->key, (string)$message); |
|
44
|
1 |
|
} catch (ConnectionException $e) { |
|
|
|
|
|
|
45
|
1 |
|
throw new QueueStoreException('Cannot add message to redis queue: ' . $e->getMessage()); |
|
46
|
|
|
} |
|
47
|
3 |
|
} |
|
48
|
|
|
|
|
49
|
|
|
/** |
|
50
|
|
|
* @return MessageInterface |
|
51
|
|
|
* @throws EmptyQueueException |
|
52
|
|
|
* @throws QueueStoreException |
|
53
|
|
|
*/ |
|
54
|
3 |
|
public function fetch(): MessageInterface |
|
55
|
|
|
{ |
|
56
|
|
|
try { |
|
57
|
3 |
|
$message = $this->client->lpop($this->key); |
|
58
|
2 |
|
if ($message) { |
|
59
|
2 |
|
return GenericMessage::fromString($message); |
|
60
|
|
|
} |
|
61
|
|
|
|
|
62
|
|
|
throw new EmptyQueueException(); |
|
63
|
2 |
|
} catch (ConnectionException $e) { |
|
|
|
|
|
|
64
|
1 |
|
throw new QueueStoreException('Cannot add message to redis queue ' . $e->getMessage()); |
|
65
|
|
|
} |
|
66
|
|
|
} |
|
67
|
|
|
|
|
68
|
|
|
|
|
69
|
|
|
/** |
|
70
|
|
|
* @return int |
|
71
|
|
|
* @throws QueueStoreException |
|
72
|
|
|
*/ |
|
73
|
2 |
|
public function count(): int |
|
74
|
|
|
{ |
|
75
|
|
|
try { |
|
76
|
2 |
|
return $this->client->llen($this->key); |
|
77
|
1 |
|
} catch (ConnectionException $e) { |
|
|
|
|
|
|
78
|
1 |
|
throw new QueueStoreException('Cannot get messages from redis queue ' . $e->getMessage()); |
|
79
|
|
|
} |
|
80
|
|
|
} |
|
81
|
|
|
} |
genkgo /
mail
Loading history...
Scrutinizer analyzes your
composer.json/composer.lockfile if available to determine the classes, and functions that are defined by your dependencies.It seems like the listed class was neither found in your dependencies, nor was it found in the analyzed files in your repository. If you are using some other form of dependency management, you might want to disable this analysis.