120 lines
3.4 KiB
PHP
120 lines
3.4 KiB
PHP
<?php
|
|
|
|
namespace App\BusProNet;
|
|
|
|
trait ApiClientTrait
|
|
{
|
|
/**
|
|
* @throws ApiClientException
|
|
*/
|
|
private function connect(string $host, int $port, int $maxRetries = 25)
|
|
{
|
|
$tries = 1;
|
|
$errNo = $errStr = '';
|
|
$errorCodesForRetry = [
|
|
SOCKET_ECONNREFUSED,
|
|
SOCKET_EBADF,
|
|
];
|
|
|
|
$openSocket = function (&$errNo, &$errStr) use ($host, $port) {
|
|
return @fsockopen(
|
|
$host,
|
|
$port,
|
|
$errNo,
|
|
$errStr,
|
|
10
|
|
);
|
|
};
|
|
|
|
$socket = $openSocket($errNo, $errStr);
|
|
|
|
while (false === $socket && true === in_array($errNo, $errorCodesForRetry) && $maxRetries > $tries) {
|
|
$this->logger->warning('Could not connect to socket, retrying', [
|
|
'error_message' => $errStr,
|
|
'error_number' => $errNo,
|
|
]);
|
|
++$tries;
|
|
sleep(1);
|
|
$socket = $openSocket($errNo, $errStr);
|
|
}
|
|
|
|
if (false !== $socket) {
|
|
stream_set_timeout($socket, 60);
|
|
} else {
|
|
$this->logger->error('Unable to open socket', [
|
|
'error_message' => $errStr,
|
|
'error_number' => $errNo,
|
|
]);
|
|
throw new ApiClientException('Unable to open socket');
|
|
}
|
|
|
|
return $socket;
|
|
}
|
|
|
|
private function send($socket, string $data): void
|
|
{
|
|
// message length is prepended to actual message
|
|
$send = sprintf('%010s', strlen($data)).$data;
|
|
fwrite($socket, $send);
|
|
}
|
|
|
|
/**
|
|
* Reads a single response message. The protocol prepends the payload length as a
|
|
* 10 byte header, so read exactly that many bytes rather than guessing at EOF:
|
|
* a peer that closes mid-stream would otherwise yield a silently truncated body.
|
|
*
|
|
* @throws ApiClientException
|
|
*/
|
|
private function receive($socket): string
|
|
{
|
|
$header = $this->readBytes($socket, 10);
|
|
|
|
if (10 !== strlen($header)) {
|
|
throw new ApiClientException(sprintf('Incomplete response header, got %d of 10 bytes', strlen($header)));
|
|
}
|
|
|
|
$expectedLength = (int) trim($header);
|
|
|
|
if (1 > $expectedLength) {
|
|
throw new ApiClientException(sprintf('Response announced an empty body (header "%s")', trim($header)));
|
|
}
|
|
|
|
$body = $this->readBytes($socket, $expectedLength);
|
|
|
|
if (strlen($body) !== $expectedLength) {
|
|
throw new ApiClientException(sprintf('Truncated response, got %d of %d announced bytes', strlen($body), $expectedLength));
|
|
}
|
|
|
|
return $body;
|
|
}
|
|
|
|
/**
|
|
* @throws ApiClientException
|
|
*/
|
|
private function readBytes($socket, int $length): string
|
|
{
|
|
$buffer = '';
|
|
|
|
while (strlen($buffer) < $length && false === feof($socket)) {
|
|
$chunk = fread($socket, min(4096, $length - strlen($buffer)));
|
|
|
|
if (false === $chunk || '' === $chunk) {
|
|
if (true === (stream_get_meta_data($socket)['timed_out'] ?? false)) {
|
|
throw new ApiClientException(sprintf('Timed out reading response after %d of %d bytes', strlen($buffer), $length));
|
|
}
|
|
|
|
break;
|
|
}
|
|
|
|
$buffer .= $chunk;
|
|
}
|
|
|
|
return $buffer;
|
|
}
|
|
|
|
private function disconnect($socket): void
|
|
{
|
|
@fclose($socket);
|
|
}
|
|
}
|