Skip to content

Commit e223a69

Browse files
Adam Roses Wightejegg
Adam Roses Wight
authored and
ejegg
committed
Implement IndexedFifoQueue for Predis
Bug: T99152 Change-Id: I2be22aa1441423e58671d58f814259e77372959a
1 parent d76213b commit e223a69

3 files changed

Lines changed: 243 additions & 14 deletions

File tree

‎composer.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@
2626
"mrpoundsign/pheanstalk-5.3": "dev-master",
2727
"aws/aws-sdk-php": "dev-master",
2828
"amazonwebservices/aws-sdk-for-php": "dev-master",
29-
"predis/predis": "v0.8.0",
29+
"predis/predis": "1.*",
3030
"iron-io/iron_mq": "dev-master",
3131
"ext-memcache": "*",
3232
"microsoft/windowsazure": "dev-master"

‎src/PHPQueue/Backend/Predis.php‎

Lines changed: 136 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,22 @@
66
use PHPQueue\Interfaces\FifoQueueStore;
77

88
/**
9-
* NOTE: The FIFO index is not usable as a key-value selector in this backend.
9+
* Wraps several styles of redis use:
10+
* - If constructed with a "order_key" option, the data will be accessible
11+
* as a key-value store, and will also provide pop and push using
12+
* $data[$order_key] as the FIFO ordering. If the ordering value is a
13+
* timestamp, for example, then the queue will have real-world FIFO
14+
* behavior over time, and even if the data comes in out of order, we will
15+
* always pop the true oldest record.
16+
* If you wish to push to this type of store, you'll also need to provide
17+
* the "correlation_key" option so the random-access key can be
18+
* extracted from data.
19+
* - Pushing scalar data will store it as a queue under queue_name.
20+
* - Setting scalar data will store it under the key.
21+
* - If data is an array, setting will store it as a hash, under the key.
22+
*
23+
* TODO: The different behaviors should be modeled as several backends which
24+
* perhaps inherit from an AbstractPredis.
1025
*/
1126
class Predis
1227
extends Base
@@ -18,9 +33,15 @@ class Predis
1833
const TYPE_SET='set';
1934
const TYPE_NONE='none';
2035

36+
// Internal sub-key to hold the ordering.
37+
const FIFO_INDEX = 'fifo';
38+
2139
public $servers;
2240
public $redis_options = array();
2341
public $queue_name;
42+
public $expiry;
43+
public $order_key;
44+
public $correlation_key;
2445

2546
public function __construct($options=array())
2647
{
@@ -34,11 +55,21 @@ public function __construct($options=array())
3455
if (!empty($options['queue'])) {
3556
$this->queue_name = $options['queue'];
3657
}
58+
if (!empty($options['expiry'])) {
59+
$this->expiry = $options['expiry'];
60+
}
61+
if (!empty($options['order_key'])) {
62+
$this->order_key = $options['order_key'];
63+
$this->redis_options['prefix'] = $this->queue_name . ':';
64+
}
65+
if (!empty($options['correlation_key'])) {
66+
$this->correlation_key = $options['correlation_key'];
67+
}
3768
}
3869

3970
public function connect()
4071
{
41-
if (empty($this->servers)) {
72+
if (!$this->servers) {
4273
throw new BackendException("No servers specified");
4374
}
4475
$this->connection = new \Predis\Client($this->servers, $this->redis_options);
@@ -47,7 +78,7 @@ public function connect()
4778
/** @deprecated */
4879
public function add($data=array())
4980
{
50-
if (empty($data)) {
81+
if (!$data) {
5182
throw new BackendException("No data.");
5283
}
5384
$this->push($data);
@@ -61,21 +92,64 @@ public function push($data)
6192
throw new BackendException("No queue specified.");
6293
}
6394
$encoded_data = json_encode($data);
64-
// Note that we're ignoring the "new length" return value, cos I don't
65-
// see how to make it useful.
66-
$this->getConnection()->rpush($this->queue_name, $encoded_data);
95+
if ($this->order_key) {
96+
if (!$this->correlation_key) {
97+
throw new BackendException("Cannot push to indexed fifo queue without a correlation key.");
98+
}
99+
$key = $data[$this->correlation_key];
100+
if (!$key) {
101+
throw new BackendException("Cannot push to indexed fifo queue without correlation data.");
102+
}
103+
$status = $this->addToIndexedFifoQueue($key, $data);
104+
if (!$status) {
105+
throw new BackendException("Couldn't push to indexed fifo queue.");
106+
}
107+
} else {
108+
// Note that we're ignoring the "new length" return value, cos I don't
109+
// see how to make it useful.
110+
$this->getConnection()->rpush($this->queue_name, $encoded_data);
111+
}
67112
}
68113

69114
/**
70115
* @return array|null
71116
*/
72117
public function pop()
73118
{
119+
$data = null;
74120
$this->beforeGet();
75121
if (!$this->hasQueue()) {
76122
throw new BackendException("No queue specified.");
77123
}
78-
$data = $this->getConnection()->lpop($this->queue_name);
124+
if ($this->order_key) {
125+
// Pop the first element.
126+
//
127+
// Adapted from https://github.com/nrk/predis/blob/v1.0/examples/transaction_using_cas.php
128+
$options = array(
129+
'cas' => true,
130+
'watch' => self::FIFO_INDEX,
131+
'retry' => 3,
132+
);
133+
$order_key = $this->order_key;
134+
$this->getConnection()->transaction($options, function ($tx) use ($order_key, &$data) {
135+
// Look up the first element in the FIFO ordering.
136+
$values = $tx->zrange(self::FIFO_INDEX, 0, 0);
137+
if ($values) {
138+
// Use that value as a key into the key-value block, to get the data.
139+
$key = $values[0];
140+
$data = $tx->get($key);
141+
142+
// Begin transaction.
143+
$tx->multi();
144+
145+
// Remove from both indexes.
146+
$tx->zrem(self::FIFO_INDEX, $key);
147+
$tx->del($key);
148+
}
149+
});
150+
} else {
151+
$data = $this->getConnection()->lpop($this->queue_name);
152+
}
79153
if (!$data) {
80154
return null;
81155
}
@@ -111,24 +185,30 @@ public function setKey($key=null, $data=null)
111185
/**
112186
* @param string $key
113187
* @param array|string $data
114-
* @return boolean
115188
* @throws \PHPQueue\Exception
116189
*/
117190
public function set($key, $data)
118191
{
119-
if (empty($key) && !is_string($key)) {
192+
if (!$key || !is_string($key)) {
120193
throw new BackendException("Key is invalid.");
121194
}
122-
if (empty($data)) {
195+
if (!$data) {
123196
throw new BackendException("No data.");
124197
}
125198
$this->beforeAdd();
126199
try {
127200
$status = false;
128-
if (is_array($data)) {
201+
if ($this->order_key) {
202+
$status = $this->addToIndexedFifoQueue($key, $data);
203+
} elseif (is_array($data)) {
204+
// FIXME: Assert
129205
$status = $this->getConnection()->hmset($key, $data);
130206
} elseif (is_string($data) || is_numeric($data)) {
131-
$status = $this->getConnection()->set($key, $data);
207+
if ($this->expiry) {
208+
$status = $this->getConnection()->setex($key, $this->expiry, $data);
209+
} else {
210+
$status = $this->getConnection()->set($key, $data);
211+
}
132212
}
133213
if (!$status) {
134214
throw new BackendException("Unable to save data.");
@@ -138,6 +218,35 @@ public function set($key, $data)
138218
}
139219
}
140220

221+
/**
222+
* Store the data under its order and correlation keys
223+
*
224+
* @param string $key
225+
* @param array $data
226+
*/
227+
protected function addToIndexedFifoQueue($key, $data)
228+
{
229+
$options = array(
230+
'cas' => true,
231+
'watch' => self::FIFO_INDEX,
232+
'retry' => 3,
233+
);
234+
$score = $data[$this->order_key];
235+
$encoded_data = json_encode($data);
236+
$status = false;
237+
$expiry = $this->expiry;
238+
$this->getConnection()->transaction($options, function ($tx) use ($key, $score, $encoded_data, $expiry, &$status) {
239+
$tx->multi();
240+
$tx->zadd(self::FIFO_INDEX, $score, $key);
241+
if ($expiry) {
242+
$status = $tx->setex($key, $expiry, $encoded_data);
243+
} else {
244+
$status = $tx->set($key, $encoded_data);
245+
}
246+
});
247+
return $status;
248+
}
249+
141250
/** @deprecated */
142251
public function getKey($key=null)
143252
{
@@ -159,6 +268,10 @@ public function get($key=null)
159268
return null;
160269
}
161270
$this->beforeGet($key);
271+
if ($this->order_key) {
272+
$data = $this->getConnection()->get($key);
273+
return json_decode($data, true);
274+
}
162275
$type = $this->getConnection()->type($key);
163276
switch ($type) {
164277
case self::TYPE_STRING:
@@ -193,7 +306,17 @@ public function clearKey($key=null)
193306
public function clear($key)
194307
{
195308
$this->beforeClear($key);
196-
$num_removed = $this->getConnection()->del($key);
309+
310+
if ($this->order_key) {
311+
$result = $this->getConnection()->pipeline()
312+
->zrem(self::FIFO_INDEX, $key)
313+
->del($key)
314+
->execute();
315+
316+
$num_removed = $result[1];
317+
} else {
318+
$num_removed = $this->getConnection()->del($key);
319+
}
197320

198321
$this->afterClearRelease();
199322

‎test/PHPQueue/Backend/PredisZsetTest.php‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
<?php
2+
namespace PHPQueue\Backend;
3+
class PredisZsetTest extends \PHPUnit_Framework_TestCase
4+
{
5+
private $object;
6+
7+
public function setUp()
8+
{
9+
parent::setUp();
10+
if (!class_exists('\Predis\Client')) {
11+
$this->markTestSkipped('Predis not installed');
12+
} else {
13+
$options = array(
14+
'servers' => array('host' => '127.0.0.1', 'port' => 6379)
15+
, 'queue' => 'testqueue-' . mt_rand()
16+
, 'order_key' => 'timestamp'
17+
, 'correlation_key' => 'txn_id'
18+
);
19+
$this->object = new Predis($options);
20+
}
21+
}
22+
23+
public function tearDown()
24+
{
25+
if ($this->object) {
26+
$this->object->getConnection()->flushall();
27+
}
28+
parent::tearDown();
29+
}
30+
31+
public function testSet()
32+
{
33+
$key = 'A0001';
34+
$data = array('name' => 'Michael', 'timestamp' => 1);
35+
$this->object->set($key, $data);
36+
37+
$key = 'A0001';
38+
$data = array('name' => 'Michael Cheng', 'timestamp' => 2);
39+
$this->object->set($key, $data);
40+
41+
$key = 'A0002';
42+
$data = array('name' => 'Michael Cheng', 'timestamp' => 3);
43+
$this->object->set($key, $data);
44+
}
45+
46+
public function testGet()
47+
{
48+
$key = 'A0001';
49+
$data1 = array('name' => 'Michael', 'timestamp' => 1);
50+
$this->object->set($key, $data1);
51+
52+
$key = 'A0001';
53+
$data2 = array('name' => 'Michael Cheng', 'timestamp' => 2);
54+
$this->object->set($key, $data2);
55+
56+
$key = 'A0002';
57+
$data3 = array('name' => 'Michael Cheng', 'timestamp' => 3);
58+
$this->object->set($key, $data3);
59+
60+
$result = $this->object->get('A0001');
61+
$this->assertEquals($data2, $result);
62+
63+
$result = $this->object->getKey('A0002');
64+
$this->assertEquals($data3, $result);
65+
}
66+
67+
public function testClear()
68+
{
69+
$key = 'A0002';
70+
$data = array('name' => 'Adam Wight', 'timestamp' => 2718);
71+
$result = $this->object->set($key, $data);
72+
73+
$result = $this->object->clear($key);
74+
$this->assertTrue($result);
75+
76+
$result = $this->object->get($key);
77+
$this->assertNull($result);
78+
}
79+
80+
public function testClearEmpty()
81+
{
82+
$jobId = 'xxx';
83+
$this->assertFalse($this->object->clear($jobId));
84+
}
85+
86+
public function testPushPop()
87+
{
88+
$data = array(
89+
'name' => 'Weezle-' . mt_rand(),
90+
'timestamp' => mt_rand(),
91+
'txn_id' => mt_rand(),
92+
);
93+
$this->object->push($data);
94+
95+
$this->assertEquals($data, $this->object->get($data['txn_id']));
96+
97+
$this->assertEquals($data, $this->object->pop());
98+
99+
$this->assertNull($this->object->get($data['txn_id']));
100+
}
101+
102+
public function testPopEmpty()
103+
{
104+
$this->assertNull($this->object->pop());
105+
}
106+
}

0 commit comments

Comments
 (0)