Server::start() - Code Metrics - Inspection of "Implement blocking http php server for master" - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 67f1c0...e9d99c )

by Anton

created 2024-10-22 20:19 UTC

Server::start() A

↳ Parent: Project

Complexity

Conditions	2
Paths	1

Size

Total Lines	20
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	15
c	0
b	0
f	0
nc	1
nop	0
dl	0
loc	20
rs	9.7666
ccs	13
cts	13
cp	1
crap	2

<?php

declare(strict_types=1);

/* (c) Anton Medvedev <[email protected]>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Deployer\Executor;

use Closure;
use Deployer\Exception\Exception;
use Symfony\Component\Console\Output\OutputInterface;

class Server
{
    private string $host;
    private int $port;

    private OutputInterface $output;

    private bool $stop = false;

    /**
     * @var ?resource
     */
    private $socket;

    /**
     * @var resource[]
     */
    private array $clientSockets = [];

    private Closure $afterCallback;
    private Closure $tickerCallback;
    private Closure $routerCallback;

    public function __construct($host, $port, OutputInterface $output)
    {
        self::checkRequiredExtensionsExists();
        $this->host = $host;
        $this->port = $port;
        $this->output = $output;
    }

    public static function checkRequiredExtensionsExists(): void
    {
        if (!function_exists('socket_import_stream')) {
            throw new Exception('Required PHP extension "sockets" is not loaded');
        }
        if (!function_exists('stream_set_blocking')) {
            throw new Exception('Required PHP extension "stream" is not loaded');
        }
    }

    public function run(): void
    {
        try {
            $this->socket = $this->createServerSocket();
            $this->updatePort();
            if ($this->output->isDebug()) {
                $this->output->writeln("[master] Starting server at http://{$this->host}:{$this->port}");
            }

            ($this->afterCallback)($this->port);

            while (true) {
                $this->acceptNewConnections();
                $this->handleClientRequests();

                // Prevent CPU exhaustion and 60fps ticker.
                usleep(16_000); // 16ms


                ($this->tickerCallback)();

                if ($this->stop) {
                    break;
                }
            }

            if ($this->output->isDebug()) {
                $this->output->writeln("[master] Stopping server at http://{$this->host}:{$this->port}");
            }
        } finally {
            if (isset($this->socket)) {
                fclose($this->socket);
            }
        }
    }

    /**
     * @return resource
     * @throws Exception
     */
    private function createServerSocket()
    {
        $server = stream_socket_server("tcp://{$this->host}:{$this->port}", $errno, $errstr);
        if (!$server) {
            throw new Exception("Socket creation failed: $errstr ($errno)");
        }

        if (!stream_set_blocking($server, false)) {
            throw new Exception("Failed to set server socket to non-blocking mode");
        }

        return $server;
    }

    private function updatePort(): void
    {
        $name = stream_socket_get_name($this->socket, false);
        if ($name) {
            list(, $port) = explode(':', $name);
            $this->port = (int) $port;
        } else {
            throw new Exception("Failed to get the assigned port");
        }
    }

    private function acceptNewConnections(): void
    {
        $newClientSocket = @stream_socket_accept($this->socket, 0);
        if ($newClientSocket) {
            if (!stream_set_blocking($newClientSocket, false)) {
                throw new Exception("Failed to set client socket to non-blocking mode");
            }
            $this->clientSockets[] = $newClientSocket;
        }
    }

    private function handleClientRequests(): void
    {
        foreach ($this->clientSockets as $key => $clientSocket) {
            if (feof($clientSocket)) {
                $this->closeClientSocket($clientSocket, $key);
                continue;
            }

            $request = $this->readClientRequest($clientSocket);
            list($path, $payload) = $this->parseRequest($request);
            $response = ($this->routerCallback)($path, $payload);

            $this->sendResponse($clientSocket, $response);
            $this->closeClientSocket($clientSocket, $key);
        }
    }

    private function readClientRequest($clientSocket)
    {
        $request = '';
        while (($chunk = @fread($clientSocket, 1024)) !== false) {
            $request .= $chunk;
            if (strpos($request, "\r\n\r\n") !== false) {
                break;
            }
        }

        if ($chunk === false && !feof($clientSocket)) {
            throw new Exception("Socket read failed");
        }

        return $request;
    }

    private function parseRequest($request)
    {
        $lines = explode("\r\n", $request);
        $requestLine = $lines[0];
        $parts = explode(' ', $requestLine);
        if (count($parts) !== 3) {
            throw new Exception("Malformed request line: $requestLine");
        }
        $path = $parts[1];

        $headers = [];
        for ($i = 1; $i < count($lines); $i++) {
            $line = $lines[$i];
            if (empty($line)) {
                break;
            }
            [$key, $value] = explode(':', $line, 2);
            $headers[$key] = trim($value);
        }
        if (empty($headers['Content-Type']) || $headers['Content-Type'] !== 'application/json') {
            throw new Exception("Malformed request: invalid Content-Type");
        }

        $payload = json_decode(implode("\n", array_slice($lines, $i + 1)), true, flags: JSON_THROW_ON_ERROR);
        return [$path, $payload];
    }

    private function sendResponse($clientSocket, Response $response)
    {
        $code = $response->getStatus();
        $content = json_encode($response->getBody(), flags: JSON_PRETTY_PRINT);
        $headers = "HTTP/1.1 $code OK\r\n" .
            "Content-Type: application/json\r\n" .
            "Content-Length: " . strlen($content) . "\r\n" .
            "Connection: close\r\n\r\n";
        fwrite($clientSocket, $headers . $content);
    }

    private function closeClientSocket($clientSocket, $key): void
    {
        fclose($clientSocket);
        unset($this->clientSockets[$key]);
    }

    public function afterRun(Closure $param): void
    {
        $this->afterCallback = $param;
    }

    public function ticker(Closure $param): void
    {
        $this->tickerCallback = $param;
    }

    public function router(Closure $param)
    {
        $this->routerCallback = $param;
    }

    public function stop(): void
    {
        $this->stop = true;
    }
}


1		<?php
2
3		declare(strict_types=1);
4
5		/* (c) Anton Medvedev <[email protected]>
6		*
7		* For the full copyright and license information, please view the LICENSE
8		* file that was distributed with this source code.
9		*/
10
11		namespace Deployer\Executor;
12
13		use Closure;
14		use Deployer\Exception\Exception;
15		use Symfony\Component\Console\Output\OutputInterface;
16
17		class Server
18		{
19		private string $host;
20		private int $port;
21
22		private OutputInterface $output;
23
24		private bool $stop = false;
25
26		/**
27		* @var ?resource
28		*/
29	12	private $socket;
30
31		/**
32		* @var resource[]
33		*/
34		private array $clientSockets = [];
35	12
36	12	private Closure $afterCallback;
37	12	private Closure $tickerCallback;
38	12	private Closure $routerCallback;
39
40	12	public function __construct($host, $port, OutputInterface $output)
41		{
42	12	self::checkRequiredExtensionsExists();
43		$this->host = $host;
44		$this->port = $port;
45	4	$this->output = $output;
46		}
47
48		public static function checkRequiredExtensionsExists(): void
49		{
50	12	if (!function_exists('socket_import_stream')) {
51	12	throw new Exception('Required PHP extension "sockets" is not loaded');
52	12	}
53	12	if (!function_exists('stream_set_blocking')) {
54	12	throw new Exception('Required PHP extension "stream" is not loaded');
55	12	}
56		}
57	4
58		public function run(): void
59	4	{
60	4	try {
61	4	$this->socket = $this->createServerSocket();
62	4	$this->updatePort();
63		if ($this->output->isDebug()) {
64	4	$this->output->writeln("[master] Starting server at http://{$this->host}:{$this->port}");
65	4	}
66
67	4	($this->afterCallback)($this->port);
68
69	4	while (true) {
70	4	$this->acceptNewConnections();
71		$this->handleClientRequests();
72	4
73	4	// Prevent CPU exhaustion and 60fps ticker.
74		usleep(16_000); // 16ms
		0 ignored issues – show Bug introduced 2024-10-22 20:22 UTC by Report Bug Copy Issue Report A parse error occurred: Syntax error, unexpected T_STRING, expecting ',' or ')' on line 74 at column 25 Loading history...
75	4
76		($this->tickerCallback)();
77	1
78	1	if ($this->stop) {
79		break;
80	1	}
81	1	}
82	1
83		if ($this->output->isDebug()) {
84	1	$this->output->writeln("[master] Stopping server at http://{$this->host}:{$this->port}");
85		}
86		} finally {
87		if (isset($this->socket)) {
88		fclose($this->socket);
89		}
90		}
91	4	}
92
93	4	/**
94	4	* @return resource
95		* @throws Exception
96	4	*/
97		private function createServerSocket()
98	4	{
99	4	$server = stream_socket_server("tcp://{$this->host}:{$this->port}", $errno, $errstr);
100		if (!$server) {
101	4	throw new Exception("Socket creation failed: $errstr ($errno)");
102		}
103	4
104	4	if (!stream_set_blocking($server, false)) {
105		throw new Exception("Failed to set server socket to non-blocking mode");
106	4	}
107
108	4	return $server;
109	4	}
110
111	4	private function updatePort(): void
112		{
113	4	$name = stream_socket_get_name($this->socket, false);
114	4	if ($name) {
115		list(, $port) = explode(':', $name);
116	4	$this->port = (int) $port;
117		} else {
118	4	throw new Exception("Failed to get the assigned port");
119		}
120		}
121
122		private function acceptNewConnections(): void
123		{
124		$newClientSocket = @stream_socket_accept($this->socket, 0);
125		if ($newClientSocket) {
126		if (!stream_set_blocking($newClientSocket, false)) {
127		throw new Exception("Failed to set client socket to non-blocking mode");
128		}
129		$this->clientSockets[] = $newClientSocket;
130		}
131		}
132
133		private function handleClientRequests(): void
134		{
135		foreach ($this->clientSockets as $key => $clientSocket) {
136		if (feof($clientSocket)) {
137		$this->closeClientSocket($clientSocket, $key);
138		continue;
139		}
140
141		$request = $this->readClientRequest($clientSocket);
142		list($path, $payload) = $this->parseRequest($request);
143		$response = ($this->routerCallback)($path, $payload);
144
145		$this->sendResponse($clientSocket, $response);
146		$this->closeClientSocket($clientSocket, $key);
147		}
148		}
149
150		private function readClientRequest($clientSocket)
151		{
152		$request = '';
153		while (($chunk = @fread($clientSocket, 1024)) !== false) {
154		$request .= $chunk;
155		if (strpos($request, "\r\n\r\n") !== false) {
156		break;
157		}
158		}
159
160		if ($chunk === false && !feof($clientSocket)) {
161		throw new Exception("Socket read failed");
162		}
163
164		return $request;
165		}
166
167		private function parseRequest($request)
168		{
169		$lines = explode("\r\n", $request);
170		$requestLine = $lines[0];
171		$parts = explode(' ', $requestLine);
172		if (count($parts) !== 3) {
173		throw new Exception("Malformed request line: $requestLine");
174		}
175		$path = $parts[1];
176
177		$headers = [];
178		for ($i = 1; $i < count($lines); $i++) {
179		$line = $lines[$i];
180		if (empty($line)) {
181		break;
182		}
183		[$key, $value] = explode(':', $line, 2);
184		$headers[$key] = trim($value);
185		}
186		if (empty($headers['Content-Type']) \|\| $headers['Content-Type'] !== 'application/json') {
187		throw new Exception("Malformed request: invalid Content-Type");
188		}
189
190		$payload = json_decode(implode("\n", array_slice($lines, $i + 1)), true, flags: JSON_THROW_ON_ERROR);
191		return [$path, $payload];
192		}
193
194		private function sendResponse($clientSocket, Response $response)
195		{
196		$code = $response->getStatus();
197		$content = json_encode($response->getBody(), flags: JSON_PRETTY_PRINT);
198		$headers = "HTTP/1.1 $code OK\r\n" .
199		"Content-Type: application/json\r\n" .
200		"Content-Length: " . strlen($content) . "\r\n" .
201		"Connection: close\r\n\r\n";
202		fwrite($clientSocket, $headers . $content);
203		}
204
205		private function closeClientSocket($clientSocket, $key): void
206		{
207		fclose($clientSocket);
208		unset($this->clientSockets[$key]);
209		}
210
211		public function afterRun(Closure $param): void
212		{
213		$this->afterCallback = $param;
214		}
215
216		public function ticker(Closure $param): void
217		{
218		$this->tickerCallback = $param;
219		}
220
221		public function router(Closure $param)
222		{
223		$this->routerCallback = $param;
224		}
225
226		public function stop(): void
227		{
228		$this->stop = true;
229		}
230		}
231

deployphp / deployer

GitHub Access Token became invalid

Push — master ( 67f1c0...e9d99c )

Server::start() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like