Skip to content

Commit fd9637f

Browse files
hmacrhmacr
committed
Implement RPUSH for priority messages
1 parent ffdc931 commit fd9637f

9 files changed

Lines changed: 195 additions & 5 deletions

File tree

‎composer.json‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,10 @@
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+
"Tests\\Unit\\": "tests/Queue/Unit"
21+
}
1922
},
2023
"scripts":{
2124
"test": "phpunit",

‎phpunit.xml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,9 @@
99
stopOnFailure="false"
1010
>
1111
<testsuites>
12+
<testsuite name="Unit">
13+
<directory>./tests/Queue/Unit</directory>
14+
</testsuite>
1215
<testsuite name="E2E">
1316
<directory>./tests/Queue/E2E/Adapter</directory>
1417
</testsuite>

‎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: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,17 @@ 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+
sleep(1);
98+
}
99+
89100
/**
90101
* @depends testEvents
91102
*/

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

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
<?php
2+
3+
namespace Tests\E2E\Adapter;
4+
5+
use PHPUnit\Framework\TestCase;
6+
use Utopia\Queue\Broker\Redis as RedisBroker;
7+
use Utopia\Queue\Connection\Redis;
8+
use Utopia\Queue\Queue;
9+
10+
/**
11+
* Verifies that priority jobs (pushed to the tail via rightPushArray) are consumed
12+
* before normal jobs (pushed to the head via leftPushArray) when BRPOP reads from
13+
* the tail.
14+
*
15+
* This test bypasses the worker and reads directly from the queue so it can assert ordering.
16+
*/
17+
class RedisPriorityTest extends TestCase
18+
{
19+
private RedisBroker $broker;
20+
private Queue $queue;
21+
private Redis $connection;
22+
23+
protected function setUp(): void
24+
{
25+
$this->connection = new Redis('redis', 6379);
26+
$this->broker = new RedisBroker($this->connection);
27+
$this->queue = new Queue('priority-e2e-test');
28+
29+
// Flush any leftover state from previous runs.
30+
$key = "{$this->queue->namespace}.queue.{$this->queue->name}";
31+
while ($this->connection->rightPopArray($key, 0) !== false) {
32+
// drain
33+
}
34+
}
35+
36+
public function testPriorityJobIsConsumedBeforeNormalJobs(): void
37+
{
38+
// Enqueue three normal jobs (pushed to head/left).
39+
$this->broker->enqueue($this->queue, ['order' => 'normal-1']);
40+
$this->broker->enqueue($this->queue, ['order' => 'normal-2']);
41+
$this->broker->enqueue($this->queue, ['order' => 'normal-3']);
42+
43+
// Enqueue one priority job (pushed to tail/right — same end BRPOP reads from).
44+
$this->broker->enqueue($this->queue, ['order' => 'priority'], priority: true);
45+
46+
$key = "{$this->queue->namespace}.queue.{$this->queue->name}";
47+
48+
// The first pop should yield the priority job.
49+
$first = $this->connection->rightPopArray($key, 1);
50+
$this->assertNotFalse($first, 'Expected a job but queue was empty');
51+
$this->assertSame('priority', $first['payload']['order'], 'Priority job should be consumed first');
52+
53+
// The remaining three should be normal jobs (consumed oldest-first).
54+
$second = $this->connection->rightPopArray($key, 1);
55+
$this->assertSame('normal-1', $second['payload']['order']);
56+
57+
$third = $this->connection->rightPopArray($key, 1);
58+
$this->assertSame('normal-2', $third['payload']['order']);
59+
60+
$fourth = $this->connection->rightPopArray($key, 1);
61+
$this->assertSame('normal-3', $fourth['payload']['order']);
62+
63+
// Queue should now be empty.
64+
$this->assertFalse($this->connection->rightPopArray($key, 0));
65+
}
66+
67+
public function testEnqueuePriorityReturnsBool(): void
68+
{
69+
$result = $this->broker->enqueue($this->queue, ['check' => 'return-value'], priority: true);
70+
$this->assertIsBool($result);
71+
$this->assertTrue($result);
72+
}
73+
}

‎tests/Queue/Unit/Broker/RedisBrokerTest.php‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
<?php
2+
3+
namespace Tests\Unit\Broker;
4+
5+
use PHPUnit\Framework\TestCase;
6+
use Utopia\Queue\Broker\Redis;
7+
use Utopia\Queue\Connection;
8+
use Utopia\Queue\Queue;
9+
10+
class RedisBrokerTest extends TestCase
11+
{
12+
private Connection $connection;
13+
private Redis $broker;
14+
private Queue $queue;
15+
16+
protected function setUp(): void
17+
{
18+
$this->connection = $this->createMock(Connection::class);
19+
$this->broker = new Redis($this->connection);
20+
$this->queue = new Queue('test');
21+
}
22+
23+
public function testEnqueueNormalUsesLeftPush(): void
24+
{
25+
$this->connection
26+
->expects($this->once())
27+
->method('leftPushArray')
28+
->with(
29+
$this->equalTo('utopia-queue.queue.test'),
30+
$this->callback(fn($p) => $p['queue'] === 'test' && $p['payload'] === ['foo' => 'bar'])
31+
)
32+
->willReturn(true);
33+
34+
$this->connection->expects($this->never())->method('rightPushArray');
35+
36+
$result = $this->broker->enqueue($this->queue, ['foo' => 'bar']);
37+
38+
$this->assertTrue($result);
39+
}
40+
41+
public function testEnqueuePriorityFalseUsesLeftPush(): void
42+
{
43+
$this->connection
44+
->expects($this->once())
45+
->method('leftPushArray')
46+
->willReturn(true);
47+
48+
$this->connection->expects($this->never())->method('rightPushArray');
49+
50+
$result = $this->broker->enqueue($this->queue, ['foo' => 'bar'], priority: false);
51+
52+
$this->assertTrue($result);
53+
}
54+
55+
public function testEnqueuePriorityUsesRightPush(): void
56+
{
57+
$this->connection
58+
->expects($this->once())
59+
->method('rightPushArray')
60+
->with(
61+
$this->equalTo('utopia-queue.queue.test'),
62+
$this->callback(fn($p) => $p['queue'] === 'test' && $p['payload'] === ['urgent' => true])
63+
)
64+
->willReturn(true);
65+
66+
$this->connection->expects($this->never())->method('leftPushArray');
67+
68+
$result = $this->broker->enqueue($this->queue, ['urgent' => true], priority: true);
69+
70+
$this->assertTrue($result);
71+
}
72+
73+
public function testEnqueuePriorityPayloadHasRequiredFields(): void
74+
{
75+
$capturedPayload = null;
76+
77+
$this->connection
78+
->expects($this->once())
79+
->method('rightPushArray')
80+
->with(
81+
$this->anything(),
82+
$this->callback(function ($p) use (&$capturedPayload) {
83+
$capturedPayload = $p;
84+
return true;
85+
})
86+
)
87+
->willReturn(true);
88+
89+
$this->broker->enqueue($this->queue, ['data' => 1], priority: true);
90+
91+
$this->assertArrayHasKey('pid', $capturedPayload);
92+
$this->assertArrayHasKey('queue', $capturedPayload);
93+
$this->assertArrayHasKey('timestamp', $capturedPayload);
94+
$this->assertArrayHasKey('payload', $capturedPayload);
95+
$this->assertNotEmpty($capturedPayload['pid']);
96+
}
97+
}

0 commit comments

Comments
 (0)