Skip to content

Commit 26a744f

Browse files
MeldironMeldiron
authored
Merge pull request #75 from utopia-php/ser-1137
Implement RPUSH for priority messages
2 parents ffdc931 + 45f5a59 commit 26a744f

8 files changed

Lines changed: 109 additions & 9 deletions

File tree

‎composer.json‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,9 @@
1515
"psr-4": {"Utopia\\Queue\\": "src/Queue"}
1616
},
1717
"autoload-dev": {
18-
"psr-4": {"Tests\\E2E\\": "tests/Queue/E2E"}
18+
"psr-4": {
19+
"Tests\\E2E\\": "tests/Queue/E2E"
20+
}
1921
},
2022
"scripts":{
2123
"test": "phpunit",

‎src/Queue/Broker/AMQP.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ public function close(): void
135135
$this->channel?->getConnection()?->close();
136136
}
137137

138-
public function enqueue(Queue $queue, array $payload): bool
138+
public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
139139
{
140140
$payload = [
141141
'pid' => \uniqid(more_entropy: true),

‎src/Queue/Broker/Pool.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ public function __construct(
1515
) {
1616
}
1717

18-
public function enqueue(Queue $queue, array $payload): bool
18+
public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
1919
{
2020
return $this->delegatePublish(__FUNCTION__, \func_get_args());
2121
}

‎src/Queue/Broker/Redis.php‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,14 +104,17 @@ public function close(): void
104104
$this->closed = true;
105105
}
106106

107-
public function enqueue(Queue $queue, array $payload): bool
107+
public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
108108
{
109109
$payload = [
110110
'pid' => \uniqid(more_entropy: true),
111111
'queue' => $queue->name,
112112
'timestamp' => time(),
113113
'payload' => $payload
114114
];
115+
if ($priority) {
116+
return $this->connection->rightPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
117+
}
115118
return $this->connection->leftPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
116119
}
117120

‎src/Queue/Publisher.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ interface Publisher
1111
* @param array $payload
1212
* @return bool
1313
*/
14-
public function enqueue(Queue $queue, array $payload): bool;
14+
public function enqueue(Queue $queue, array $payload, bool $priority = false): bool;
1515

1616
/**
1717
* Retries failed jobs.

‎tests/Queue/E2E/Adapter/Base.php‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,15 @@ public function testConcurrency(): void
8686
});
8787
}
8888

89+
public function testEnqueuePriority(): void
90+
{
91+
$publisher = $this->getPublisher();
92+
93+
$result = $publisher->enqueue($this->getQueue(), ['type' => 'test_string', 'value' => 'priority'], priority: true);
94+
95+
$this->assertTrue($result);
96+
}
97+
8998
/**
9099
* @depends testEvents
91100
*/

‎tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php‎

Lines changed: 47 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,14 +9,59 @@
99

1010
class SwooleRedisClusterTest extends Base
1111
{
12+
private function getConnection(): RedisCluster
13+
{
14+
return new RedisCluster([
15+
'redis-cluster-0:6379',
16+
'redis-cluster-1:6379',
17+
'redis-cluster-2:6379',
18+
]);
19+
}
20+
1221
protected function getPublisher(): Publisher
1322
{
14-
$connection = new RedisCluster(['redis-cluster-0:6379', 'redis-cluster-1:6379', 'redis-cluster-2:6379']);
15-
return new Redis($connection);
23+
return new Redis($this->getConnection());
1624
}
1725

1826
protected function getQueue(): Queue
1927
{
2028
return new Queue('swoole-redis-cluster');
2129
}
30+
31+
public function testPriorityJobIsConsumedBeforeNormalJobs(): void
32+
{
33+
$connection = $this->getConnection();
34+
$key = "{$this->getQueue()->namespace}.queue.{$this->getQueue()->name}";
35+
36+
// Flush any leftover state from previous runs.
37+
while ($connection->rightPopArray($key, 1) !== false) {
38+
// drain
39+
}
40+
41+
// Enqueue three normal jobs (pushed to head/left).
42+
$this->getPublisher()->enqueue($this->getQueue(), ['order' => 'normal-1']);
43+
$this->getPublisher()->enqueue($this->getQueue(), ['order' => 'normal-2']);
44+
$this->getPublisher()->enqueue($this->getQueue(), ['order' => 'normal-3']);
45+
46+
// Enqueue one priority job (pushed to tail/right — same end BRPOP reads from).
47+
$this->getPublisher()->enqueue($this->getQueue(), ['order' => 'priority'], priority: true);
48+
49+
// The first pop should yield the priority job.
50+
$first = $connection->rightPopArray($key, 1);
51+
$this->assertNotFalse($first, 'Expected a job but queue was empty');
52+
$this->assertSame('priority', $first['payload']['order'], 'Priority job should be consumed first');
53+
54+
// The remaining three should be normal jobs (consumed oldest-first).
55+
$second = $connection->rightPopArray($key, 1);
56+
$this->assertSame('normal-1', $second['payload']['order']);
57+
58+
$third = $connection->rightPopArray($key, 1);
59+
$this->assertSame('normal-2', $third['payload']['order']);
60+
61+
$fourth = $connection->rightPopArray($key, 1);
62+
$this->assertSame('normal-3', $fourth['payload']['order']);
63+
64+
// Queue should now be empty.
65+
$this->assertFalse($connection->rightPopArray($key, 1));
66+
}
2267
}

‎tests/Queue/E2E/Adapter/SwooleTest.php‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,14 +9,55 @@
99

1010
class SwooleTest extends Base
1111
{
12+
private function getConnection(): Redis
13+
{
14+
return new Redis('redis', 6379);
15+
}
16+
1217
protected function getPublisher(): Publisher
1318
{
14-
$connection = new Redis('redis', 6379);
15-
return new RedisBroker($connection);
19+
return new RedisBroker($this->getConnection());
1620
}
1721

1822
protected function getQueue(): Queue
1923
{
2024
return new Queue('swoole');
2125
}
26+
27+
public function testPriorityJobIsConsumedBeforeNormalJobs(): void
28+
{
29+
$connection = $this->getConnection();
30+
$key = "{$this->getQueue()->namespace}.queue.{$this->getQueue()->name}";
31+
32+
// Flush any leftover state from previous runs.
33+
while ($connection->rightPopArray($key, 1) !== false) {
34+
// drain
35+
}
36+
37+
// Enqueue three normal jobs (pushed to head/left).
38+
$this->getPublisher()->enqueue($this->getQueue(), ['order' => 'normal-1']);
39+
$this->getPublisher()->enqueue($this->getQueue(), ['order' => 'normal-2']);
40+
$this->getPublisher()->enqueue($this->getQueue(), ['order' => 'normal-3']);
41+
42+
// Enqueue one priority job (pushed to tail/right — same end BRPOP reads from).
43+
$this->getPublisher()->enqueue($this->getQueue(), ['order' => 'priority'], priority: true);
44+
45+
// The first pop should yield the priority job.
46+
$first = $connection->rightPopArray($key, 1);
47+
$this->assertNotFalse($first, 'Expected a job but queue was empty');
48+
$this->assertSame('priority', $first['payload']['order'], 'Priority job should be consumed first');
49+
50+
// The remaining three should be normal jobs (consumed oldest-first).
51+
$second = $connection->rightPopArray($key, 1);
52+
$this->assertSame('normal-1', $second['payload']['order']);
53+
54+
$third = $connection->rightPopArray($key, 1);
55+
$this->assertSame('normal-2', $third['payload']['order']);
56+
57+
$fourth = $connection->rightPopArray($key, 1);
58+
$this->assertSame('normal-3', $fourth['payload']['order']);
59+
60+
// Queue should now be empty.
61+
$this->assertFalse($connection->rightPopArray($key, 1));
62+
}
2263
}

0 commit comments

Comments
 (0)