17432557268

Committed 03 Sep 2025 11:52AM UTC coverage: 98.446% (-0.07%) from 98.513%

Build # 17432557268

Build Type

push

github

Committed by

web-flow

Commit Message

chore: migrate from `psalm` to `mago` (#527)

Run Details

232 of 241 new or added lines in 81 files covered. (96.27%)

14 existing lines in 12 files now uncovered.

5510 of 5597 relevant lines covered (98.45%)

52.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.3

/src/Psl/Async/Sequence.php

<?php

declare(strict_types=1);

namespace Psl\Async;

use Closure;
use Exception;
use Revolt\EventLoop;
use Revolt\EventLoop\Suspension;

use function array_slice;

/**
 * Run an operation with a limit on number of ongoing asynchronous jobs of 1.
 *
 * Just like {@see Semaphore}, all operations must have the same input type (Tin) and output type (Tout), and be processed by the same function;
 *
 * @template Tin
 * @template Tout
 *
 * @see Semaphore
 *
 * @mago-expect lint:no-else-clause
 */
final class Sequence
{
    private bool $ingoing = false;

    /**
     * @var list<Suspension>
     */
    private array $pending = [];

    /**
     * @var list<Suspension>
     */
    private array $waits = [];

    /**
     * @param (Closure(Tin): Tout) $operation
     */
    public function __construct(
        private readonly Closure $operation,
    ) {}

    /**
     * Run the operation using the given `$input`, after all previous operations have completed.
     *
     * @param Tin $input
     *
     * @return Tout
     *
     * @see Sequence::cancel()
     */
    public function waitFor(mixed $input): mixed
    {
        if ($this->ingoing) {
            $suspension = EventLoop::getSuspension();
            $this->pending[] = $suspension;

            $suspension->suspend();
        }

        $this->ingoing = true;

        try {
            return ($this->operation)($input);
        } finally {
            $suspension = $this->pending[0] ?? null;
            if ($suspension !== null) {
                $this->pending = array_slice($this->pending, 1);
                $suspension->resume();
            } else {
                foreach ($this->waits as $suspension) {
                    $suspension->resume();
                }

                $this->waits = [];

                $this->ingoing = false;
            }
        }
    }

    /**
     * Cancel all pending operations.
     *
     * Any pending operation will fail with the given exception.
     *
     * Future operations will continue execution as usual.
     */
    public function cancel(Exception $exception): void
    {
        $suspensions = $this->pending;
        $this->pending = [];
        foreach ($suspensions as $suspension) {
            $suspension->throw($exception);
        }
    }

    /**
     * Get the number of operations pending execution.
     *
     * @return int<0, max>
     */
    public function getPendingOperations(): int
    {
        /** @var int<0, max> */
        return count($this->pending);
    }

    /**
     * Check if there's any operations pending execution.
     *
     * If this method returns `true`, it means future calls to `waitFor` will wait.
     */
    public function hasPendingOperations(): bool
    {
        return $this->pending !== [];
    }

    /**
     * Check if the sequence has any ingoing operations.
     *
     * If this method returns `true`, it means future calls to `waitFor` will wait.
     * If this method returns `false`, it means future calls to `waitFor` will execute immediately.
     */
    public function hasIngoingOperations(): bool
    {
        return $this->ingoing;
    }

    /**
     * Wait for all pending operations to finish execution.
     */
    public function waitForPending(): void
    {
        if (!$this->ingoing) {
            return;
        }

        $suspension = EventLoop::getSuspension();
        $this->waits[] = $suspension;
        $suspension->suspend();
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	namespace Psl\Async;
6
7	use Closure;
8	use Exception;
9	use Revolt\EventLoop;
10	use Revolt\EventLoop\Suspension;
11
12	use function array_slice;
13
14	/**
15	* Run an operation with a limit on number of ongoing asynchronous jobs of 1.
16	*
17	* Just like {@see Semaphore}, all operations must have the same input type (Tin) and output type (Tout), and be processed by the same function;
18	*
19	* @template Tin
20	* @template Tout
21	*
22	* @see Semaphore
23	*
24	* @mago-expect lint:no-else-clause
25	*/
26	final class Sequence
27	{
28	private bool $ingoing = false;
29
30	/**
31	* @var list<Suspension>
32	*/
33	private array $pending = [];
34
35	/**
36	* @var list<Suspension>
37	*/
38	private array $waits = [];
39
40	/**
41	* @param (Closure(Tin): Tout) $operation
42	*/
43	public function __construct(
44	private readonly Closure $operation,
45	) {}	85✔
46
47	/**
48	* Run the operation using the given `$input`, after all previous operations have completed.
49	*
50	* @param Tin $input
51	*
52	* @return Tout
53	*
54	* @see Sequence::cancel()
55	*/
56	public function waitFor(mixed $input): mixed
57	{
58	if ($this->ingoing) {	50✔
59	$suspension = EventLoop::getSuspension();	6✔
60	$this->pending[] = $suspension;	6✔
61
62	$suspension->suspend();	6✔
63	}
64
65	$this->ingoing = true;	50✔
66
67	try {
68	return ($this->operation)($input);	50✔
69	} finally {
70	$suspension = $this->pending[0] ?? null;	50✔
71	if ($suspension !== null) {	50✔
72	$this->pending = array_slice($this->pending, 1);	5✔
73	$suspension->resume();	5✔
74	} else {
75	foreach ($this->waits as $suspension) {	50✔
76	$suspension->resume();	1✔
77	}
78
79	$this->waits = [];	50✔
80
81	$this->ingoing = false;	50✔
82	}
83	}
84	}
85
86	/**
87	* Cancel all pending operations.
88	*
89	* Any pending operation will fail with the given exception.
90	*
91	* Future operations will continue execution as usual.
92	*/
93	public function cancel(Exception $exception): void
94	{
95	$suspensions = $this->pending;	73✔
96	$this->pending = [];	73✔
97	foreach ($suspensions as $suspension) {	73✔
98	$suspension->throw($exception);	1✔
99	}
100	}
101
102	/**
103	* Get the number of operations pending execution.
104	*
105	* @return int<0, max>
106	*/
107	public function getPendingOperations(): int
108	{
109	/** @var int<0, max> */
110	return count($this->pending);	2✔
111	}
112
113	/**
114	* Check if there's any operations pending execution.
115	*
116	* If this method returns `true`, it means future calls to `waitFor` will wait.
117	*/
118	public function hasPendingOperations(): bool
119	{
120	return $this->pending !== [];	2✔
121	}
122
123	/**
124	* Check if the sequence has any ingoing operations.
125	*
126	* If this method returns `true`, it means future calls to `waitFor` will wait.
127	* If this method returns `false`, it means future calls to `waitFor` will execute immediately.
128	*/
129	public function hasIngoingOperations(): bool
130	{
131	return $this->ingoing;	2✔
132	}
133
134	/**
135	* Wait for all pending operations to finish execution.
136	*/
137	public function waitForPending(): void
138	{
139	if (!$this->ingoing) {	1✔
UNCOV 140	return;	×
141	}
142
143	$suspension = EventLoop::getSuspension();	1✔
144	$this->waits[] = $suspension;	1✔
145	$suspension->suspend();	1✔
146	}
147	}

azjezz / psl / 17432557268

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous