refactor: move buspro socket framing into SocketTransport
This commit is contained in:
fromme committed
1 parent
e51664df6c
commit
3c81c79819
4 files changed
+289
-277
No files matched your search
@@ -0,0 +1,265 @@
|
||||
<?php
|
||||
|
||||
namespace App\BusProNet\Transport;
|
||||
|
||||
use App\BusProNet\Exception\ApiClientException;
|
||||
use App\BusProNet\Exception\ImmediateConnectionCloseException;
|
||||
use App\BusProNet\Exception\TimeoutException;
|
||||
use Psr\Log\LoggerInterface;
|
||||
|
||||
/**
|
||||
* BusPro's socket framing: one connection per request, each message prefixed with its length
|
||||
* as a 10 byte header. Knows nothing about what the messages contain.
|
||||
*/
|
||||
class SocketTransport
|
||||
{
|
||||
private float $operationStartTime;
|
||||
private int $selectedPort;
|
||||
|
||||
/** @param list<int|string> $ports */
|
||||
public function __construct(
|
||||
private readonly LoggerInterface $logger,
|
||||
private readonly string $host,
|
||||
private readonly array $ports,
|
||||
private readonly int $connectionTimeout = 5,
|
||||
private readonly int $streamTimeout = 30,
|
||||
private readonly int $totalTimeout = 45,
|
||||
private readonly int $maxRetries = 25,
|
||||
) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends $body over a fresh connection and returns the response payload. $logMessage is
|
||||
* logged with $context and the chosen port once the connection is open.
|
||||
*
|
||||
* @param array<string, mixed> $context
|
||||
*
|
||||
* @throws ApiClientException
|
||||
* @throws ImmediateConnectionCloseException
|
||||
*/
|
||||
public function exchange(string $body, string $logMessage, array $context = []): string
|
||||
{
|
||||
$socket = $this->connect();
|
||||
|
||||
$this->logger->info($logMessage, [...$context, 'port' => $this->selectedPort]);
|
||||
|
||||
$this->send($socket, $body);
|
||||
|
||||
try {
|
||||
// the length header is consumed and verified by receive(), this is the payload
|
||||
return $this->receive($socket);
|
||||
} finally {
|
||||
$this->disconnect($socket);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @return resource
|
||||
*
|
||||
* @throws ApiClientException
|
||||
* @throws TimeoutException
|
||||
*/
|
||||
private function connect()
|
||||
{
|
||||
// Randomly select a port from the available ports for load balancing
|
||||
$this->selectedPort = $this->ports[array_rand($this->ports)];
|
||||
$this->operationStartTime = microtime(true);
|
||||
|
||||
$tries = 1;
|
||||
$errNo = $errStr = '';
|
||||
$errorCodesForRetry = [
|
||||
SOCKET_ECONNREFUSED,
|
||||
SOCKET_EBADF,
|
||||
];
|
||||
|
||||
$openSocket = function (&$errNo, &$errStr) {
|
||||
return @fsockopen(
|
||||
$this->host,
|
||||
$this->selectedPort,
|
||||
$errNo,
|
||||
$errStr,
|
||||
$this->connectionTimeout
|
||||
);
|
||||
};
|
||||
|
||||
$socket = $openSocket($errNo, $errStr);
|
||||
|
||||
while (false === $socket && true === in_array($errNo, $errorCodesForRetry) && $this->maxRetries > $tries) {
|
||||
// Check if we've exceeded total timeout during retries
|
||||
if (microtime(true) - $this->operationStartTime > $this->totalTimeout) {
|
||||
$this->logger->error('Connection retry timeout exceeded', [
|
||||
'elapsed_time' => microtime(true) - $this->operationStartTime,
|
||||
'total_timeout' => $this->totalTimeout,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
throw new TimeoutException('Connection timeout exceeded during retries');
|
||||
}
|
||||
|
||||
$this->logger->warning('Could not connect to socket, retrying', [
|
||||
'error_message' => $errStr,
|
||||
'error_number' => $errNo,
|
||||
'attempt' => $tries,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
++$tries;
|
||||
sleep(1);
|
||||
$socket = $openSocket($errNo, $errStr);
|
||||
}
|
||||
|
||||
if (false !== $socket) {
|
||||
stream_set_timeout($socket, $this->streamTimeout);
|
||||
} else {
|
||||
$this->logger->error('Unable to open socket', [
|
||||
'error_message' => $errStr,
|
||||
'error_number' => $errNo,
|
||||
'elapsed_time' => microtime(true) - $this->operationStartTime,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
throw new ApiClientException('Unable to open socket');
|
||||
}
|
||||
|
||||
return $socket;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param resource $socket
|
||||
*
|
||||
* @throws TimeoutException
|
||||
*/
|
||||
private function send($socket, string $data): void
|
||||
{
|
||||
// Check total timeout before sending
|
||||
if (microtime(true) - $this->operationStartTime > $this->totalTimeout) {
|
||||
$this->logger->error('Total timeout exceeded before send', [
|
||||
'elapsed_time' => microtime(true) - $this->operationStartTime,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
throw new TimeoutException('Total operation timeout exceeded before send');
|
||||
}
|
||||
|
||||
// message length is prepended to actual message
|
||||
$send = sprintf('%010s', strlen($data)).$data;
|
||||
fwrite($socket, $send);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads a single response message and returns its payload. The protocol prepends the
|
||||
* payload length as a 10 byte header, so the announced length is compared against what
|
||||
* actually arrived: a peer closing mid-stream would otherwise yield a silently
|
||||
* truncated body that only surfaces later as an unexplained parse failure.
|
||||
*
|
||||
* @param resource $socket
|
||||
*
|
||||
* @throws ApiClientException
|
||||
* @throws TimeoutException
|
||||
* @throws ImmediateConnectionCloseException
|
||||
*/
|
||||
private function receive($socket): string
|
||||
{
|
||||
$response = '';
|
||||
$readAttempts = 0;
|
||||
$totalTimeout = $this->totalTimeout;
|
||||
|
||||
while (false === feof($socket)) {
|
||||
// Check total timeout before each read
|
||||
$elapsedTime = microtime(true) - $this->operationStartTime;
|
||||
if ($elapsedTime > $totalTimeout) {
|
||||
$this->logger->error('Total timeout exceeded during receive', [
|
||||
'elapsed_time' => $elapsedTime,
|
||||
'total_timeout' => $totalTimeout,
|
||||
'bytes_received' => strlen($response),
|
||||
'read_attempts' => $readAttempts,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
throw new TimeoutException('Total operation timeout exceeded while receiving data');
|
||||
}
|
||||
|
||||
$chunk = fread($socket, 4096);
|
||||
++$readAttempts;
|
||||
|
||||
// Check if stream timed out on this specific read
|
||||
$metadata = stream_get_meta_data($socket);
|
||||
if (true === $metadata['timed_out']) {
|
||||
// Check if this is an immediate rejection (0 bytes, < 1 second)
|
||||
// This indicates the server is busy rather than a true timeout
|
||||
if (0 === strlen($response) && $elapsedTime < 1.0) {
|
||||
$this->logger->warning('Server closed connection immediately', [
|
||||
'elapsed_time' => $elapsedTime,
|
||||
'bytes_received' => 0,
|
||||
'read_attempts' => $readAttempts,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
throw new ImmediateConnectionCloseException('Server closed connection immediately - server may be busy');
|
||||
}
|
||||
|
||||
$this->logger->error('Stream read timeout detected', [
|
||||
'elapsed_time' => $elapsedTime,
|
||||
'bytes_received' => strlen($response),
|
||||
'read_attempts' => $readAttempts,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
throw new TimeoutException('Stream timeout while reading from socket');
|
||||
}
|
||||
|
||||
$response .= $chunk;
|
||||
}
|
||||
|
||||
if (0 === strlen($response)) {
|
||||
$elapsedTime = microtime(true) - $this->operationStartTime;
|
||||
$this->logger->warning('Server closed connection without sending data', [
|
||||
'elapsed_time' => $elapsedTime,
|
||||
'read_attempts' => $readAttempts,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
throw new ImmediateConnectionCloseException('Server closed connection without sending data');
|
||||
}
|
||||
|
||||
return $this->extractPayload($response);
|
||||
}
|
||||
|
||||
/**
|
||||
* @throws ApiClientException
|
||||
*/
|
||||
private function extractPayload(string $response): string
|
||||
{
|
||||
if (10 > strlen($response)) {
|
||||
$this->logger->error('Incomplete response header received from API', [
|
||||
'bytes_received' => strlen($response),
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
|
||||
throw new ApiClientException(sprintf('Incomplete response header, got %d of 10 bytes', strlen($response)));
|
||||
}
|
||||
|
||||
$header = substr($response, 0, 10);
|
||||
$announcedLength = (int) trim($header);
|
||||
$payload = substr($response, 10);
|
||||
|
||||
if (1 > $announcedLength) {
|
||||
$this->logger->error('Empty response body announced by API', [
|
||||
'header' => trim($header),
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
|
||||
throw new ApiClientException(sprintf('Response announced an empty body (header "%s")', trim($header)));
|
||||
}
|
||||
|
||||
if (strlen($payload) !== $announcedLength) {
|
||||
$this->logger->error('Truncated response received from API', [
|
||||
'bytes_received' => strlen($payload),
|
||||
'announced_length' => $announcedLength,
|
||||
'port' => $this->selectedPort,
|
||||
]);
|
||||
|
||||
throw new ApiClientException(sprintf('Truncated response, got %d of %d announced bytes', strlen($payload), $announcedLength));
|
||||
}
|
||||
|
||||
return $payload;
|
||||
}
|
||||
|
||||
/** @param resource $socket */
|
||||
private function disconnect($socket): void
|
||||
{
|
||||
@fclose($socket);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user