|
1
|
|
|
<?php |
|
2
|
|
|
|
|
3
|
|
|
namespace Bernard\Driver\IronMQ; |
|
4
|
|
|
|
|
5
|
|
|
use Bernard\Driver\AbstractPrefetchDriver; |
|
6
|
|
|
use IronMQ\IronMQ; |
|
7
|
|
|
|
|
8
|
|
|
/** |
|
9
|
|
|
* Implements a Driver for use with Iron MQ: |
|
10
|
|
|
* https://github.com/iron-io/iron_mq_php |
|
11
|
|
|
* |
|
12
|
|
|
* @package Bernard |
|
13
|
|
|
*/ |
|
14
|
|
|
final class Driver extends AbstractPrefetchDriver |
|
15
|
|
|
{ |
|
16
|
|
|
private $ironmq; |
|
17
|
|
|
|
|
18
|
|
|
/** |
|
19
|
|
|
* @param IronMQ $ironmq |
|
20
|
|
|
* @param int|null $prefetch |
|
21
|
|
|
*/ |
|
22
|
|
|
public function __construct(IronMQ $ironmq, $prefetch = null) |
|
23
|
|
|
{ |
|
24
|
|
|
parent::__construct($prefetch); |
|
25
|
|
|
|
|
26
|
|
|
$this->ironmq = $ironmq; |
|
27
|
|
|
} |
|
28
|
|
|
|
|
29
|
|
|
/** |
|
30
|
|
|
* {@inheritdoc} |
|
31
|
|
|
*/ |
|
32
|
|
|
public function listQueues() |
|
33
|
|
|
{ |
|
34
|
|
|
$queueNames = array(); |
|
35
|
|
|
$page = 0; |
|
36
|
|
|
|
|
37
|
|
|
while ($queues = $this->ironmq->getQueues($page, 100)) { |
|
38
|
|
|
$queueNames += $this->pluck($queues, 'name'); |
|
39
|
|
|
|
|
40
|
|
|
// If we get 100 results the probability of another page is high. |
|
41
|
|
|
if (count($queues) < 100) { |
|
42
|
|
|
break; |
|
43
|
|
|
} |
|
44
|
|
|
|
|
45
|
|
|
$page++; |
|
46
|
|
|
} |
|
47
|
|
|
|
|
48
|
|
|
return $queueNames; |
|
49
|
|
|
} |
|
50
|
|
|
|
|
51
|
|
|
/** |
|
52
|
|
|
* {@inheritdoc} |
|
53
|
|
|
*/ |
|
54
|
|
|
public function createQueue($queueName) |
|
55
|
|
|
{ |
|
56
|
|
|
} |
|
57
|
|
|
|
|
58
|
|
|
/** |
|
59
|
|
|
* {@inheritdoc} |
|
60
|
|
|
*/ |
|
61
|
|
|
public function countMessages($queueName) |
|
62
|
|
|
{ |
|
63
|
|
|
if ($info = $this->ironmq->getQueue($queueName)) { |
|
64
|
|
|
return $info->size; |
|
65
|
|
|
} |
|
66
|
|
|
} |
|
67
|
|
|
|
|
68
|
|
|
/** |
|
69
|
|
|
* {@inheritdoc} |
|
70
|
|
|
*/ |
|
71
|
|
|
public function pushMessage($queueName, $message) |
|
72
|
|
|
{ |
|
73
|
|
|
$this->ironmq->postMessage($queueName, $message); |
|
74
|
|
|
} |
|
75
|
|
|
|
|
76
|
|
|
/** |
|
77
|
|
|
* {@inheritdoc} |
|
78
|
|
|
*/ |
|
79
|
|
|
public function popMessage($queueName, $duration = 5) |
|
80
|
|
|
{ |
|
81
|
|
|
if ($message = $this->cache->pop($queueName)) { |
|
82
|
|
|
return $message; |
|
83
|
|
|
} |
|
84
|
|
|
|
|
85
|
|
|
$timeout = IronMQ::GET_MESSAGE_TIMEOUT; |
|
86
|
|
|
|
|
87
|
|
|
$messages = $this->ironmq->getMessages($queueName, $this->prefetch, $timeout, $duration); |
|
|
|
|
|
|
88
|
|
|
|
|
89
|
|
|
if (!$messages) { |
|
90
|
|
|
return array(null, null); |
|
91
|
|
|
} |
|
92
|
|
|
|
|
93
|
|
|
foreach ($messages as $message) { |
|
94
|
|
|
$this->cache->push($queueName, array($message->body, $message->id)); |
|
95
|
|
|
} |
|
96
|
|
|
|
|
97
|
|
|
return $this->cache->pop($queueName); |
|
98
|
|
|
} |
|
99
|
|
|
|
|
100
|
|
|
/** |
|
101
|
|
|
* {@inheritdoc} |
|
102
|
|
|
*/ |
|
103
|
|
|
public function acknowledgeMessage($queueName, $receipt) |
|
104
|
|
|
{ |
|
105
|
|
|
$this->ironmq->deleteMessage($queueName, $receipt); |
|
106
|
|
|
} |
|
107
|
|
|
|
|
108
|
|
|
/** |
|
109
|
|
|
* IronMQ does not support an offset when peeking messages. |
|
110
|
|
|
* |
|
111
|
|
|
* {@inheritdoc} |
|
112
|
|
|
*/ |
|
113
|
|
|
public function peekQueue($queueName, $index = 0, $limit = 20) |
|
114
|
|
|
{ |
|
115
|
|
|
if ($messages = $this->ironmq->peekMessages($queueName, $limit)) { |
|
116
|
|
|
return $this->pluck($messages, 'body'); |
|
117
|
|
|
} |
|
118
|
|
|
|
|
119
|
|
|
return []; |
|
120
|
|
|
} |
|
121
|
|
|
|
|
122
|
|
|
/** |
|
123
|
|
|
* {@inheritdoc} |
|
124
|
|
|
*/ |
|
125
|
|
|
public function removeQueue($queueName) |
|
126
|
|
|
{ |
|
127
|
|
|
$this->ironmq->deleteQueue($queueName); |
|
128
|
|
|
} |
|
129
|
|
|
|
|
130
|
|
|
/** |
|
131
|
|
|
* {@inheritdoc} |
|
132
|
|
|
*/ |
|
133
|
|
|
public function info() |
|
134
|
|
|
{ |
|
135
|
|
|
return [ |
|
136
|
|
|
'prefetch' => $this->prefetch, |
|
137
|
|
|
]; |
|
138
|
|
|
} |
|
139
|
|
|
|
|
140
|
|
|
/** |
|
141
|
|
|
* The missing array_pluck but for objects array |
|
142
|
|
|
* |
|
143
|
|
|
* @param array $objects |
|
144
|
|
|
* @param string $property |
|
145
|
|
|
* |
|
146
|
|
|
* @return array |
|
147
|
|
|
*/ |
|
148
|
|
|
protected function pluck(array $objects, $property) |
|
149
|
|
|
{ |
|
150
|
|
|
$function = function ($object) use ($property) { |
|
151
|
|
|
return $object->$property; |
|
152
|
|
|
}; |
|
153
|
|
|
|
|
154
|
|
|
return array_map($function, $objects); |
|
155
|
|
|
} |
|
156
|
|
|
} |
|
157
|
|
|
|
bernardphp /
bernard
This method has been deprecated. The supplier of the class has supplied an explanatory message.
The explanatory message should give you some clue as to whether and when the method will be removed from the class and what other method or class to use instead.