Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
18 / 18
100.00%
4 / 4
 CRAP
100.00%
1 / 1
Listener
100.00%
18 / 18
100.00%
4 / 4
 5
100.00%
1 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 fromConnection
100.00%
7 / 7
100.00%
1 / 1
 1
 wait
100.00%
7 / 7
100.00%
1 / 1
 2
 listen
100.00%
3 / 3
100.00%
1 / 1
 1
1<?php
2
3declare(strict_types=1);
4
5namespace Cosray\Queue;
6
7use Celema\Quma\Connection;
8use Closure;
9use PDO;
10use Pdo\Pgsql;
11
12/**
13 * Waits for jobs to be pushed. Pushing a job notifies the `cosray_jobs`
14 * channel once its transaction commits; the listener receives it on a
15 * connection of its own, opened on first use.
16 *
17 * @internal
18 */
19final class Listener
20{
21    public const string CHANNEL = 'cosray_jobs';
22
23    private ?Pgsql $pdo = null;
24
25    /** @param Closure(): Pgsql $connect */
26    public function __construct(
27        private readonly Closure $connect,
28    ) {}
29
30    public static function fromConnection(Connection $connection): self
31    {
32        $config = $connection->config;
33
34        return new self(static fn(): Pgsql => new Pgsql(
35            $config->dsn,
36            $config->pdo->username,
37            $config->pdo->password,
38            $config->pdo->effectiveOptions($config->driver),
39        ));
40    }
41
42    /**
43     * Waits until a job is pushed or the time is up, and returns whether
44     * one was pushed. A signal ends the wait early.
45     */
46    public function wait(float $seconds): bool
47    {
48        $pdo = $this->pdo ?? $this->listen();
49        $this->pdo = $pdo;
50
51        if ($pdo->getNotify(PDO::FETCH_ASSOC, max(0, (int) ceil($seconds * 1000))) === false) {
52            return false;
53        }
54
55        // One wake-up serves every job pushed meanwhile.
56        do {
57            $more = $pdo->getNotify(PDO::FETCH_ASSOC);
58        } while ($more !== false);
59
60        return true;
61    }
62
63    private function listen(): Pgsql
64    {
65        $pdo = ($this->connect)();
66        $pdo->exec('LISTEN ' . self::CHANNEL);
67
68        return $pdo;
69    }
70}