From 3c81c79819e72ca742a721056c9d177aed624779 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Fromme?= Date: Fri, 2 Oct 2026 13:51:48 +0200 Subject: [PATCH] refactor: move buspro socket framing into SocketTransport --- config/services.yaml | 14 +- src/BusProNet/ApiClient.php | 241 +--------------- src/BusProNet/Transport/SocketTransport.php | 265 ++++++++++++++++++ .../SocketTransportTest.php} | 46 +-- 4 files changed, 289 insertions(+), 277 deletions(-) create mode 100644 src/BusProNet/Transport/SocketTransport.php rename tests/BusProNet/{ApiClientReceiveTest.php => Transport/SocketTransportTest.php} (68%) diff --git a/config/services.yaml b/config/services.yaml index 94e1566f..85e6a630 100644 --- a/config/services.yaml +++ b/config/services.yaml @@ -134,12 +134,16 @@ services: $options: bpn_username: '%env(APP_BPN_USER)%' bpn_password: '%env(APP_BPN_PASSWORD)%' - bpn_api_ip: '%env(APP_BPN_IP)%' - bpn_api_ports: '%env(csv:APP_BPN_PORTS)%' debug: '%env(bool:APP_BPN_DEBUG)%' - connection_timeout: '%env(int:APP_BPN_CONNECTION_TIMEOUT)%' - stream_timeout: '%env(int:APP_BPN_STREAM_TIMEOUT)%' - total_timeout: '%env(int:APP_BPN_TOTAL_TIMEOUT)%' + + App\BusProNet\Transport\SocketTransport: + arguments: + $logger: '@monolog.logger.bpn' + $host: '%env(APP_BPN_IP)%' + $ports: '%env(csv:APP_BPN_PORTS)%' + $connectionTimeout: '%env(int:APP_BPN_CONNECTION_TIMEOUT)%' + $streamTimeout: '%env(int:APP_BPN_STREAM_TIMEOUT)%' + $totalTimeout: '%env(int:APP_BPN_TOTAL_TIMEOUT)%' App\Twig\AppRuntime: arguments: diff --git a/src/BusProNet/ApiClient.php b/src/BusProNet/ApiClient.php index d54120df..5977d50f 100644 --- a/src/BusProNet/ApiClient.php +++ b/src/BusProNet/ApiClient.php @@ -7,7 +7,6 @@ use App\BusProNet\DataProcessor\VoucherSalePayloadBuilder; use App\BusProNet\Exception\ApiClientException; use App\BusProNet\Exception\ImmediateConnectionCloseException; use App\BusProNet\Exception\ResponseParserException; -use App\BusProNet\Exception\TimeoutException; use App\BusProNet\Model\Agency; use App\BusProNet\Model\BaseData; use App\BusProNet\Model\Booking; @@ -27,6 +26,7 @@ use App\BusProNet\Model\VoucherPaymentMethod; use App\BusProNet\Model\VoucherPreview; use App\BusProNet\Model\VoucherSaleResponse; use App\BusProNet\Model\VoucherTemplates; +use App\BusProNet\Transport\SocketTransport; use App\BusProNet\XmlParser\ApiResponseParser; use App\Form\Model\BookingDto; use App\Form\Model\RegistrationDto; @@ -64,8 +64,6 @@ class ApiClient /** @var array */ private array $config; - private float $operationStartTime; - private int $selectedPort; private int $requestCounter = 0; /** @param array $options */ @@ -78,6 +76,7 @@ class ApiClient private readonly VoucherSalePayloadBuilder $voucherSalePayloadBuilder, private readonly RequestStack $requestStack, private readonly RequestIdGenerator $requestIdGenerator, + private readonly SocketTransport $transport, array $options, ) { $this->config = $this->resolveOptions($options); @@ -558,8 +557,6 @@ class ApiClient } /** - * One request and its response over a fresh connection, both dumped if $dump. - * * @throws ApiClientException * @throws ImmediateConnectionCloseException */ @@ -569,22 +566,7 @@ class ApiClient $this->dumpXmlToFile('request', $requestId, $body); } - $socket = $this->connect(); - - $this->logger->info($logMessage, [ - 'requestId' => $requestId, - 'type' => $type, - 'port' => $this->selectedPort, - ]); - - $this->send($socket, $body); - - try { - // the length header is consumed and verified by receive(), this is the payload - $response = $this->receive($socket); - } finally { - $this->disconnect($socket); - } + $response = $this->transport->exchange($body, $logMessage, ['requestId' => $requestId, 'type' => $type]); if ($dump) { $this->dumpXmlToFile('response', $requestId, $response); @@ -652,216 +634,6 @@ class ApiClient return md5($this->config['bpn_username'].$this->config['bpn_password'].$date.$type); } - /** - * @return resource - * - * @throws ApiClientException - * @throws TimeoutException - */ - private function connect() - { - // Randomly select a port from the available ports for load balancing - $this->selectedPort = $this->config['bpn_api_ports'][array_rand($this->config['bpn_api_ports'])]; - $this->operationStartTime = microtime(true); - - $tries = 1; - $errNo = $errStr = ''; - $errorCodesForRetry = [ - SOCKET_ECONNREFUSED, - SOCKET_EBADF, - ]; - - $openSocket = function (&$errNo, &$errStr) { - return @fsockopen( - $this->config['bpn_api_ip'], - $this->selectedPort, - $errNo, - $errStr, - $this->config['connection_timeout'] - ); - }; - - $socket = $openSocket($errNo, $errStr); - - while (false === $socket && true === in_array($errNo, $errorCodesForRetry) && $this->config['max_retries'] > $tries) { - // Check if we've exceeded total timeout during retries - if (microtime(true) - $this->operationStartTime > $this->config['total_timeout']) { - $this->logger->error('Connection retry timeout exceeded', [ - 'elapsed_time' => microtime(true) - $this->operationStartTime, - 'total_timeout' => $this->config['total_timeout'], - '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->config['stream_timeout']); - } 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->config['total_timeout']) { - $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->config['total_timeout']; - - 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); - } - /** * @param array $options * @@ -873,16 +645,9 @@ class ApiClient $optionsResolver->setRequired([ 'bpn_username', 'bpn_password', - 'bpn_api_ip', - 'bpn_api_ports', ]); - $optionsResolver->setAllowedTypes('bpn_api_ports', ['array']); $optionsResolver->setDefaults([ - 'max_retries' => 25, 'debug' => false, - 'connection_timeout' => 5, - 'stream_timeout' => 30, - 'total_timeout' => 45, 'busy_retry_attempts' => 3, 'busy_retry_delay' => 1, ]); diff --git a/src/BusProNet/Transport/SocketTransport.php b/src/BusProNet/Transport/SocketTransport.php new file mode 100644 index 00000000..4dbf8dc3 --- /dev/null +++ b/src/BusProNet/Transport/SocketTransport.php @@ -0,0 +1,265 @@ + $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 $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); + } +} diff --git a/tests/BusProNet/ApiClientReceiveTest.php b/tests/BusProNet/Transport/SocketTransportTest.php similarity index 68% rename from tests/BusProNet/ApiClientReceiveTest.php rename to tests/BusProNet/Transport/SocketTransportTest.php index 74a22189..88fb31c6 100644 --- a/tests/BusProNet/ApiClientReceiveTest.php +++ b/tests/BusProNet/Transport/SocketTransportTest.php @@ -2,24 +2,17 @@ declare(strict_types=1); -namespace App\Tests\BusProNet; +namespace App\Tests\BusProNet\Transport; -use App\BusProNet\ApiClient; -use App\BusProNet\DataProcessor\BookingDataProcessor; -use App\BusProNet\DataProcessor\VoucherSalePayloadBuilder; use App\BusProNet\Exception\ApiClientException; use App\BusProNet\Exception\ImmediateConnectionCloseException; -use App\BusProNet\XmlParser\ApiResponseParser; -use App\Service\RequestIdGenerator; -use League\Flysystem\FilesystemOperator; +use App\BusProNet\Transport\SocketTransport; use PHPUnit\Framework\TestCase; use Psr\Log\LoggerInterface; -use Symfony\Component\HttpFoundation\RequestStack; -use Symfony\Component\Serializer\SerializerInterface; -class ApiClientReceiveTest extends TestCase +class SocketTransportTest extends TestCase { - private ?ApiClient $apiClient = null; + private ?SocketTransport $transport = null; private LoggerInterface $logger; protected function setUp(): void @@ -27,24 +20,9 @@ class ApiClientReceiveTest extends TestCase $this->logger = $this->createStub(LoggerInterface::class); } - private function apiClient(): ApiClient + private function transport(): SocketTransport { - return $this->apiClient ??= new ApiClient( - $this->createStub(SerializerInterface::class), - $this->createStub(ApiResponseParser::class), - $this->createStub(FilesystemOperator::class), - $this->logger, - $this->createStub(BookingDataProcessor::class), - new VoucherSalePayloadBuilder(), - new RequestStack(), - new RequestIdGenerator(), - [ - 'bpn_username' => 'test', - 'bpn_password' => 'test', - 'bpn_api_ip' => '127.0.0.1', - 'bpn_api_ports' => [9000], - ], - ); + return $this->transport ??= new SocketTransport($this->logger, '127.0.0.1', [9000]); } public function testEmptyResponseThrowsImmediateConnectionCloseException(): void @@ -137,20 +115,20 @@ class ApiClientReceiveTest extends TestCase private function invokeReceive($socket): string { - $method = new \ReflectionMethod(ApiClient::class, 'receive'); + $method = new \ReflectionMethod(SocketTransport::class, 'receive'); - return $method->invoke($this->apiClient(), $socket); + return $method->invoke($this->transport(), $socket); } private function setOperationStartTime(): void { - $property = new \ReflectionProperty(ApiClient::class, 'operationStartTime'); - $property->setValue($this->apiClient(), microtime(true)); + $property = new \ReflectionProperty(SocketTransport::class, 'operationStartTime'); + $property->setValue($this->transport(), microtime(true)); } private function setSelectedPort(int $port): void { - $property = new \ReflectionProperty(ApiClient::class, 'selectedPort'); - $property->setValue($this->apiClient(), $port); + $property = new \ReflectionProperty(SocketTransport::class, 'selectedPort'); + $property->setValue($this->transport(), $port); } }