queue/src/Subqueue.php at 8175cf0acf1d2f9cf1a464cb3a60a743471e305d · tarantool-php/queue · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
<?php

/**
 * This file is part of the tarantool/queue package.
 *
 * (c) Eugene Leonovich <gen.work@gmail.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

declare(strict_types=1);

namespace Tarantool\Queue;

use Tarantool\Client\Client;

final class Subqueue extends AbstractQueue
{
    private readonly string $subqueueName;

    public function __construct(Client $client, string $queueName, string $subqueueName)
    {
        parent::__construct($client, $queueName);
        $this->subqueueName = $subqueueName;
    }

    public function put(
        mixed $data,
        ?int $pri = null,
        ?float $ttl = null,
        ?float $ttr = null,
        ?float $delay = null,
    ) : Task {
        $options = ['subqueue' => $this->subqueueName] + array_filter([
            'pri' => $pri,
            'ttl' => $ttl,
            'ttr' => $ttr,
            'delay' => $delay,
        ], static fn ($value) => null !== $value);

        return Task::fromTuple($this->call('put', $data, $options)[0]);
    }

    public function release(int $taskId, ?float $delay = null) : Task
    {
        $options = null === $delay
            ? ['subqueue' => $this->subqueueName]
            : ['subqueue' => $this->subqueueName, 'delay' => $delay];

        return Task::fromTuple($this->call('release', $taskId, $options)[0]);
    }

    #[\Override]
    public function take(?float $timeout = null) : ?Task
    {
        $result = $this->call('take', $timeout, ['subqueue' => $this->subqueueName]);

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

    #[\Override]
    public function touch(int $taskId, float $increment) : ?Task
    {
        $result = $this->call('touch', $taskId, $increment, ['subqueue' => $this->subqueueName]);

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

    #[\Override]
    public function ack(int $taskId) : Task
    {
        return Task::fromTuple($this->call('ack', $taskId, ['subqueue' => $this->subqueueName])[0]);
    }

    #[\Override]
    public function peek(int $taskId) : Task
    {
        return Task::fromTuple($this->call('peek', $taskId, ['subqueue' => $this->subqueueName])[0]);
    }

    #[\Override]
    public function bury(int $taskId) : Task
    {
        return Task::fromTuple($this->call('bury', $taskId, ['subqueue' => $this->subqueueName])[0]);
    }

    #[\Override]
    public function kick(int $count) : int
    {
        return $this->call('kick', $count, ['subqueue' => $this->subqueueName])[0];
    }

    #[\Override]
    public function delete(int $taskId) : Task
    {
        return Task::fromTuple($this->call('delete', $taskId, ['subqueue' => $this->subqueueName])[0]);
    }

    #[\Override]
    public function truncate() : void
    {
        $this->call('truncate', ['subqueue' => $this->subqueueName]);
    }
}