Skip to content

Commit 8175cf0

Browse files
rybakitrybakit
committed
refactor: move subqueue context handling from AbstractQueue to Subqueue
Remove the generic context machinery ($context property, callWithContext(), withContext()) from AbstractQueue and make Subqueue explicitly manage its own subqueue parameter routing per method. This makes AbstractQueue simpler and context-free, while Subqueue becomes self-contained with explicit per-method parameter passing that matches the Tarantool subqueuettl Lua driver contract: - put/release: merge subqueue opts into the options array - take/touch/ack/peek/bury/kick/delete/truncate: append subqueue opts as a trailing argument
1 parent 8576238 commit 8175cf0

3 files changed

Lines changed: 164 additions & 59 deletions

File tree

‎src/AbstractQueue.php‎

Lines changed: 8 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,6 @@
1717

1818
abstract class AbstractQueue
1919
{
20-
/** @var array<string, mixed> */
21-
protected array $context = [];
22-
2320
public function __construct(
2421
private readonly Client $client,
2522
private readonly string $name,
@@ -38,60 +35,46 @@ public function getName() : string
3835

3936
public function take(?float $timeout = null) : ?Task
4037
{
41-
$args = [$timeout];
42-
if ([] !== $this->context) {
43-
$args[] = $this->context;
44-
}
45-
46-
$result = $this->call('take', ...$args);
38+
$result = $this->call('take', $timeout);
4739

4840
return empty($result[0]) ? null : Task::fromTuple($result[0]);
4941
}
5042

5143
public function touch(int $taskId, float $increment) : ?Task
5244
{
53-
$args = [$taskId, $increment];
54-
if ([] !== $this->context) {
55-
$args[] = $this->context;
56-
}
57-
58-
$result = $this->call('touch', ...$args);
45+
$result = $this->call('touch', $taskId, $increment);
5946

6047
return empty($result[0]) ? null : Task::fromTuple($result[0]);
6148
}
6249

6350
public function ack(int $taskId) : Task
6451
{
65-
return Task::fromTuple($this->callWithContext('ack', $taskId)[0]);
52+
return Task::fromTuple($this->call('ack', $taskId)[0]);
6653
}
6754

6855
public function peek(int $taskId) : Task
6956
{
70-
return Task::fromTuple($this->callWithContext('peek', $taskId)[0]);
57+
return Task::fromTuple($this->call('peek', $taskId)[0]);
7158
}
7259

7360
public function bury(int $taskId) : Task
7461
{
75-
return Task::fromTuple($this->callWithContext('bury', $taskId)[0]);
62+
return Task::fromTuple($this->call('bury', $taskId)[0]);
7663
}
7764

7865
public function kick(int $count) : int
7966
{
80-
return $this->callWithContext('kick', $count)[0];
67+
return $this->call('kick', $count)[0];
8168
}
8269

8370
public function delete(int $taskId) : Task
8471
{
85-
return Task::fromTuple($this->callWithContext('delete', $taskId)[0]);
72+
return Task::fromTuple($this->call('delete', $taskId)[0]);
8673
}
8774

8875
public function truncate() : void
8976
{
90-
$args = [];
91-
if ([] !== $this->context) {
92-
$args[] = $this->context;
93-
}
94-
$this->call('truncate', ...$args);
77+
$this->call('truncate');
9578
}
9679

9780
/**
@@ -121,23 +104,4 @@ public function call(string $methodName, ...$args) : array
121104
{
122105
return $this->client->call("queue.tube.$this->name:$methodName", ...$args);
123106
}
124-
125-
private function callWithContext(string $methodName, mixed ...$args) : array
126-
{
127-
if ([] !== $this->context) {
128-
$args[] = $this->context;
129-
}
130-
131-
return $this->call($methodName, ...$args);
132-
}
133-
134-
/**
135-
* @param array<string, mixed> $options
136-
*
137-
* @return array<string, mixed>
138-
*/
139-
protected function withContext(array $options) : array
140-
{
141-
return $this->context + $options;
142-
}
143107
}

‎src/Subqueue.php‎

Lines changed: 61 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,12 @@
1717

1818
final class Subqueue extends AbstractQueue
1919
{
20+
private readonly string $subqueueName;
21+
2022
public function __construct(Client $client, string $queueName, string $subqueueName)
2123
{
2224
parent::__construct($client, $queueName);
23-
$this->context = ['subqueue' => $subqueueName];
25+
$this->subqueueName = $subqueueName;
2426
}
2527

2628
public function put(
@@ -30,20 +32,74 @@ public function put(
3032
?float $ttr = null,
3133
?float $delay = null,
3234
) : Task {
33-
$options = array_filter([
35+
$options = ['subqueue' => $this->subqueueName] + array_filter([
3436
'pri' => $pri,
3537
'ttl' => $ttl,
3638
'ttr' => $ttr,
3739
'delay' => $delay,
3840
], static fn ($value) => null !== $value);
3941

40-
return Task::fromTuple($this->call('put', $data, $this->withContext($options))[0]);
42+
return Task::fromTuple($this->call('put', $data, $options)[0]);
4143
}
4244

4345
public function release(int $taskId, ?float $delay = null) : Task
4446
{
45-
$options = null === $delay ? [] : ['delay' => $delay];
47+
$options = null === $delay
48+
? ['subqueue' => $this->subqueueName]
49+
: ['subqueue' => $this->subqueueName, 'delay' => $delay];
50+
51+
return Task::fromTuple($this->call('release', $taskId, $options)[0]);
52+
}
53+
54+
#[\Override]
55+
public function take(?float $timeout = null) : ?Task
56+
{
57+
$result = $this->call('take', $timeout, ['subqueue' => $this->subqueueName]);
58+
59+
return empty($result[0]) ? null : Task::fromTuple($result[0]);
60+
}
61+
62+
#[\Override]
63+
public function touch(int $taskId, float $increment) : ?Task
64+
{
65+
$result = $this->call('touch', $taskId, $increment, ['subqueue' => $this->subqueueName]);
66+
67+
return empty($result[0]) ? null : Task::fromTuple($result[0]);
68+
}
69+
70+
#[\Override]
71+
public function ack(int $taskId) : Task
72+
{
73+
return Task::fromTuple($this->call('ack', $taskId, ['subqueue' => $this->subqueueName])[0]);
74+
}
4675

47-
return Task::fromTuple($this->call('release', $taskId, $this->withContext($options))[0]);
76+
#[\Override]
77+
public function peek(int $taskId) : Task
78+
{
79+
return Task::fromTuple($this->call('peek', $taskId, ['subqueue' => $this->subqueueName])[0]);
80+
}
81+
82+
#[\Override]
83+
public function bury(int $taskId) : Task
84+
{
85+
return Task::fromTuple($this->call('bury', $taskId, ['subqueue' => $this->subqueueName])[0]);
86+
}
87+
88+
#[\Override]
89+
public function kick(int $count) : int
90+
{
91+
return $this->call('kick', $count, ['subqueue' => $this->subqueueName])[0];
92+
}
93+
94+
#[\Override]
95+
public function delete(int $taskId) : Task
96+
{
97+
return Task::fromTuple($this->call('delete', $taskId, ['subqueue' => $this->subqueueName])[0]);
98+
}
99+
100+
#[\Override]
101+
public function truncate() : void
102+
{
103+
$this->call('truncate', ['subqueue' => $this->subqueueName]);
48104
}
49105
}

‎tests/Integration/SubqueueTtlQueueTest.php‎

Lines changed: 95 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -16,26 +16,111 @@
1616
use Tarantool\PhpUnit\Attribute\Lua;
1717
use Tarantool\Queue\Queue;
1818
use Tarantool\Queue\States;
19+
use Tarantool\Queue\Subqueue;
1920

2021
final class SubqueueTtlQueueTest extends TestCase
2122
{
22-
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
23+
private function createSubqueue(string $name = 'emails') : Subqueue
24+
{
25+
return Queue::subqueueTtl($this->getClient(), $this->getQueueName())
26+
->subqueue($name);
27+
}
28+
29+
#[Lua("create_tube('%tube_name%', '%tube_type%')")]
2330
public function testTakeFromNewSubqueueReturnsNull() : void
2431
{
25-
$subqueue = Queue::subqueueTtl($this->getClient(), $this->getQueueName())
26-
->subqueue('created-on-first-put');
32+
self::assertNull($this->createSubqueue('created-on-first-put')->take(0.0));
33+
}
34+
35+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
36+
public function testPut() : void
37+
{
38+
$subqueue = $this->createSubqueue();
39+
40+
self::assertTask($subqueue->put('payload', ttl: 60.0), 0, States::READY, 'payload');
41+
}
2742

28-
self::assertNull($subqueue->take(0.0));
43+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
44+
#[Lua("tube:put('peek_0', {subqueue = 'emails'})")]
45+
public function testPeek() : void
46+
{
47+
self::assertTask($this->createSubqueue()->peek(0), 0, States::READY, 'peek_0');
48+
}
49+
50+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
51+
#[Lua("tube:put('take_0', {subqueue = 'emails'})")]
52+
public function testTake() : void
53+
{
54+
self::assertTask($this->createSubqueue()->take(1.0), 0, States::TAKEN, 'take_0');
55+
}
56+
57+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
58+
#[Lua("tube:put('ack_0', {subqueue = 'emails'})")]
59+
#[Lua("tube:take(0.001, {subqueue = 'emails'})")]
60+
public function testAck() : void
61+
{
62+
self::assertTask($this->createSubqueue()->ack(0), 0, States::DONE, 'ack_0');
63+
}
64+
65+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
66+
#[Lua("tube:put('release_0', {subqueue = 'emails'})")]
67+
#[Lua("tube:take(0.001, {subqueue = 'emails'})")]
68+
public function testRelease() : void
69+
{
70+
self::assertTask($this->createSubqueue()->release(0), 0, States::READY, 'release_0');
71+
}
72+
73+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
74+
#[Lua("tube:put('release_delay_0', {subqueue = 'emails'})")]
75+
#[Lua("tube:take(0.001, {subqueue = 'emails'})")]
76+
public function testReleaseWithDelay() : void
77+
{
78+
self::assertTask($this->createSubqueue()->release(0, delay: 0.1), 0, States::DELAYED, 'release_delay_0');
79+
}
80+
81+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
82+
#[Lua("tube:put('touch_0', {subqueue = 'emails', ttr = 2.0})")]
83+
#[Lua("tube:take(0.001, {subqueue = 'emails'})")]
84+
public function testTouch() : void
85+
{
86+
$task = $this->createSubqueue()->touch(0, 5.0);
87+
88+
self::assertNotNull($task);
89+
self::assertSame(0, $task->getId());
90+
self::assertSame(States::TAKEN, $task->getState());
91+
}
92+
93+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
94+
#[Lua("tube:put('bury_0', {subqueue = 'emails'})")]
95+
public function testBury() : void
96+
{
97+
self::assertTask($this->createSubqueue()->bury(0), 0, States::BURIED, 'bury_0');
98+
}
99+
100+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
101+
#[Lua("tube:put('kick_0', {subqueue = 'emails'})")]
102+
#[Lua('tube:bury(0)')]
103+
public function testKick() : void
104+
{
105+
self::assertSame(1, $this->createSubqueue()->kick(1));
106+
}
107+
108+
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
109+
#[Lua("tube:put('delete_0', {subqueue = 'emails'})")]
110+
public function testDelete() : void
111+
{
112+
self::assertTask($this->createSubqueue()->delete(0), 0, States::DONE, 'delete_0');
29113
}
30114

31115
#[Lua("tube = create_tube('%tube_name%', '%tube_type%')")]
32-
public function testSubqueueRoutesOperations() : void
116+
#[Lua("tube:put('truncate_0', {subqueue = 'emails'})")]
117+
#[Lua("tube:put('truncate_1', {subqueue = 'emails'})")]
118+
public function testTruncate() : void
33119
{
34-
$subqueue = Queue::subqueueTtl($this->getClient(), $this->getQueueName())
35-
->subqueue('emails');
120+
$subqueue = $this->createSubqueue();
36121

37-
self::assertTask($subqueue->put('typed', ttl: 60.0), 0, States::READY, 'typed');
38-
self::assertTask($subqueue->take(0.0), 0, States::TAKEN, 'typed');
39-
self::assertTask($subqueue->ack(0), 0, States::DONE, 'typed');
122+
self::assertSame(2, $this->queue->stats('tasks.total'));
123+
$subqueue->truncate();
124+
self::assertSame(0, $this->queue->stats('tasks.total'));
40125
}
41126
}

0 commit comments

Comments
 (0)