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

88.1

/src/Psl/Network/Internal/AbstractStreamServer.php

<?php

declare(strict_types=1);

namespace Psl\Network\Internal;

use Generator;
use Override;
use Psl\Channel;
use Psl\Network;
use Psl\Network\StreamServerInterface;
use Revolt\EventLoop;

use function error_get_last;
use function fclose;
use function is_resource;
use function stream_socket_accept;

abstract class AbstractStreamServer implements StreamServerInterface
{
    private const int DEFAULT_IDLE_CONNECTIONS = 256;

    /**
     * @var closed-resource|resource|null $impl
     */
    private mixed $impl;

    /**
     * @var string
     */
    private string $watcher;

    /**
     * @var Channel\ReceiverInterface<array{true, Socket}|array{false, Network\Exception\RuntimeException}>
     */
    private Channel\ReceiverInterface $receiver;

    /**
     * @param resource $impl
     * @param int<1, max> $idleConnections
     */
    protected function __construct(mixed $impl, int $idleConnections = self::DEFAULT_IDLE_CONNECTIONS)
    {
        $this->impl = $impl;

        /**
         * @var Channel\ReceiverInterface<array{true, Socket}|array{false, Network\Exception\RuntimeException}> $receiver
         * @var Channel\SenderInterface<array{true, Socket}|array{false, Network\Exception\RuntimeException}> $sender
         */
        [$receiver, $sender] = Channel\bounded($idleConnections);

        $this->receiver = $receiver;
        $this->watcher = EventLoop::onReadable(
            $impl,
            /**
             * @param resource $resource
             */
            static function (string $watcher, mixed $resource) use ($sender): void {
                try {
                    $sock = @stream_socket_accept($resource, timeout: 0.0);
                    if ($sock !== false) {
                        $sender->send([true, new Socket($sock)]);

                        return;
                    }

                    // @codeCoverageIgnoreStart
                    /** @var array{file: string, line: int, message: string, type: int} $err */
                    $err = error_get_last();
                    $sender->send([
                        false,
                        new Network\Exception\RuntimeException(
                            'Failed to accept incoming connection: ' . $err['message'],
                            $err['type'],
                        ),
                    ]);
                    // @codeCoverageIgnoreEnd
                } catch (Channel\Exception\ClosedChannelException) {
                    EventLoop::cancel($watcher);

                    return;
                }
            },
        );
    }

    /**
     * {@inheritDoc}
     */
    #[Override]
    public function nextConnection(): Network\StreamSocketInterface
    {
        try {
            [$success, $result] = $this->receiver->receive();
        } catch (Channel\Exception\ClosedChannelException) {
            throw new Network\Exception\AlreadyStoppedException('Server socket has already been stopped.');
        }

        if ($success) {
            /** @var Socket $result */
            return $result;
        }

        /** @var Network\Exception\RuntimeException $result */
        throw $result;
    }

    /**
     * @return Generator<null, Network\StreamSocketInterface, void, null>
     */
    #[Override]
    public function incoming(): Generator
    {
        try {
            while (true) {
                [$success, $result] = $this->receiver->receive();
                if ($success) {
                    /** @var Socket $result */
                    yield null => $result;
                    continue;
                }

                /** @var Network\Exception\RuntimeException $result */
                throw $result;
            }
        } catch (Channel\Exception\ClosedChannelException) {
            return;
        }
    }

    /**
     * {@inheritDoc}
     */
    #[\Override]
    public function getLocalAddress(): Network\Address
    {
        if (!is_resource($this->impl)) {
            throw new Network\Exception\AlreadyStoppedException('Server socket has already been stopped.');
        }

        return Network\Internal\get_sock_name($this->impl);
    }

    public function __destruct()
    {
        $this->close();
    }

    /**
     * {@inheritDoc}
     */
    #[\Override]
    public function close(): void
    {
        EventLoop::disable($this->watcher);
        if (null === $this->impl) {
            return;
        }

        $this->receiver->close();
        $resource = $this->impl;
        $this->impl = null;
        if (is_resource($resource)) {
            fclose($resource);
        }
    }

    /**
     * {@inheritDoc}
     */
    #[\Override]
    public function getStream(): mixed
    {
        /** @var resource */
        return $this->impl;
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	namespace Psl\Network\Internal;
6
7	use Generator;
8	use Override;
9	use Psl\Channel;
10	use Psl\Network;
11	use Psl\Network\StreamServerInterface;
12	use Revolt\EventLoop;
13
14	use function error_get_last;
15	use function fclose;
16	use function is_resource;
17	use function stream_socket_accept;
18
19	abstract class AbstractStreamServer implements StreamServerInterface
20	{
21	private const int DEFAULT_IDLE_CONNECTIONS = 256;
22
23	/**
24	* @var closed-resource\|resource\|null $impl
25	*/
26	private mixed $impl;
27
28	/**
29	* @var string
30	*/
31	private string $watcher;
32
33	/**
34	* @var Channel\ReceiverInterface<array{true, Socket}\|array{false, Network\Exception\RuntimeException}>
35	*/
36	private Channel\ReceiverInterface $receiver;
37
38	/**
39	* @param resource $impl
40	* @param int<1, max> $idleConnections
41	*/
42	protected function __construct(mixed $impl, int $idleConnections = self::DEFAULT_IDLE_CONNECTIONS)
43	{
44	$this->impl = $impl;	11✔
45
46	/**
47	* @var Channel\ReceiverInterface<array{true, Socket}\|array{false, Network\Exception\RuntimeException}> $receiver
48	* @var Channel\SenderInterface<array{true, Socket}\|array{false, Network\Exception\RuntimeException}> $sender
49	*/
50	[$receiver, $sender] = Channel\bounded($idleConnections);	11✔
51
52	$this->receiver = $receiver;	11✔
53	$this->watcher = EventLoop::onReadable(	11✔
54	$impl,	11✔
55	/**
56	* @param resource $resource
57	*/
58	static function (string $watcher, mixed $resource) use ($sender): void {	11✔
59	try {
60	$sock = @stream_socket_accept($resource, timeout: 0.0);	7✔
61	if ($sock !== false) {	7✔
62	$sender->send([true, new Socket($sock)]);	7✔
63
64	return;	7✔
65	}
66
67	// @codeCoverageIgnoreStart
68	/** @var array{file: string, line: int, message: string, type: int} $err */
69	$err = error_get_last();
70	$sender->send([
71	false,
72	new Network\Exception\RuntimeException(
73	'Failed to accept incoming connection: ' . $err['message'],
74	$err['type'],
75	),
76	]);
77	// @codeCoverageIgnoreEnd
NEW 78	} catch (Channel\Exception\ClosedChannelException) {	×
NEW 79	EventLoop::cancel($watcher);	×
80
UNCOV 81	return;	×
82	}
83	},	11✔
84	);	11✔
85	}
86
87	/**
88	* {@inheritDoc}
89	*/
90	#[Override]
91	public function nextConnection(): Network\StreamSocketInterface
92	{
93	try {
94	[$success, $result] = $this->receiver->receive();	8✔
95	} catch (Channel\Exception\ClosedChannelException) {	2✔
96	throw new Network\Exception\AlreadyStoppedException('Server socket has already been stopped.');	2✔
97	}
98
99	if ($success) {	6✔
100	/** @var Socket $result */
101	return $result;	6✔
102	}
103
104	/** @var Network\Exception\RuntimeException $result */
105	throw $result;	×
106	}
107
108	/**
109	* @return Generator<null, Network\StreamSocketInterface, void, null>
110	*/
111	#[Override]
112	public function incoming(): Generator
113	{
114	try {
115	while (true) {	1✔
116	[$success, $result] = $this->receiver->receive();	1✔
117	if ($success) {	1✔
118	/** @var Socket $result */
119	yield null => $result;	1✔
120	continue;	1✔
121	}
122
123	/** @var Network\Exception\RuntimeException $result */
124	throw $result;	×
125	}
126	} catch (Channel\Exception\ClosedChannelException) {	1✔
127	return;	1✔
128	}
129	}
130
131	/**
132	* {@inheritDoc}
133	*/
134	#[\Override]
135	public function getLocalAddress(): Network\Address
136	{
137	if (!is_resource($this->impl)) {	7✔
138	throw new Network\Exception\AlreadyStoppedException('Server socket has already been stopped.');	2✔
139	}
140
141	return Network\Internal\get_sock_name($this->impl);	5✔
142	}
143
144	public function __destruct()
145	{
146	$this->close();	11✔
147	}
148
149	/**
150	* {@inheritDoc}
151	*/
152	#[\Override]
153	public function close(): void
154	{
155	EventLoop::disable($this->watcher);	11✔
156	if (null === $this->impl) {	11✔
157	return;	11✔
158	}
159
160	$this->receiver->close();	11✔
161	$resource = $this->impl;	11✔
162	$this->impl = null;	11✔
163	if (is_resource($resource)) {	11✔
164	fclose($resource);	11✔
165	}
166	}
167
168	/**
169	* {@inheritDoc}
170	*/
171	#[\Override]
172	public function getStream(): mixed
173	{
174	/** @var resource */
175	return $this->impl;	2✔
176	}
177	}

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