From 69df93b7737e9c8edb295158f48f04465d5763bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Fromme?= Date: Wed, 7 Jan 2026 08:33:16 +0100 Subject: [PATCH] feat: load balancing of api requests, cleanup api client --- .env | 2 +- config/services.yaml | 2 +- src/BusProNet/ApiClient.php | 259 +++++++++++++++++------- src/BusProNet/Traits/ApiClientTrait.php | 149 -------------- 4 files changed, 190 insertions(+), 222 deletions(-) delete mode 100644 src/BusProNet/Traits/ApiClientTrait.php diff --git a/.env b/.env index 33b8056..e12f3d2 100644 --- a/.env +++ b/.env @@ -43,7 +43,7 @@ MAILER_DSN=null://null APP_BPN_USER= APP_BPN_PASSWORD= APP_BPN_IP= -APP_BPN_PORT= +APP_BPN_PORTS= APP_BPN_DEBUG=false APP_BPN_CONNECTION_TIMEOUT=5 APP_BPN_STREAM_TIMEOUT=30 diff --git a/config/services.yaml b/config/services.yaml index 00a24c3..2f4de95 100644 --- a/config/services.yaml +++ b/config/services.yaml @@ -61,7 +61,7 @@ services: bpn_username: '%env(APP_BPN_USER)%' bpn_password: '%env(APP_BPN_PASSWORD)%' bpn_api_ip: '%env(APP_BPN_IP)%' - bpn_api_port: '%env(APP_BPN_PORT)%' + 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)%' diff --git a/src/BusProNet/ApiClient.php b/src/BusProNet/ApiClient.php index 890a196..b83f9df 100644 --- a/src/BusProNet/ApiClient.php +++ b/src/BusProNet/ApiClient.php @@ -6,6 +6,7 @@ use App\BusProNet\DataProcessor\BookingDataProcessor; use App\BusProNet\Exception\ApiClientException; use App\BusProNet\Exception\ImmediateConnectionCloseException; use App\BusProNet\Exception\ResponseParserException; +use App\BusProNet\Exception\TimeoutException; use App\BusProNet\Model\BaseData; use App\BusProNet\Model\Booking; use App\BusProNet\Model\BookingResponse; @@ -18,7 +19,6 @@ use App\BusProNet\Model\PurchaseVoucher; use App\BusProNet\Model\RegistrationResponse; use App\BusProNet\Model\ServiceAvailabilityResponse; use App\BusProNet\Model\Travel; -use App\BusProNet\Traits\ApiClientTrait; use App\BusProNet\XmlParser\ApiResponseParser; use App\Form\Model\BookingDto; use App\Form\Model\RegistrationDto; @@ -31,8 +31,6 @@ use Symfony\Component\Serializer\SerializerInterface; class ApiClient { - use ApiClientTrait; - public const TYPE_NOTIFICATION = 'HINWEIS'; public const TYPE_CUSTOMER_DATA = 'KUNDENKONTO'; public const TYPE_BASE_DATA_COUNTRIES = 'STAMMLAENDER'; @@ -48,6 +46,8 @@ class ApiClient public const TYPE_PROMO_VOUCHER = 'AKTIONSGUTSCHEIN'; private array $config; + private float $operationStartTime; + private int $selectedPort; public function __construct( private readonly SerializerInterface $serializer, @@ -505,32 +505,7 @@ class ApiClient */ public function sendRawXml(string $xml, bool $debug = false): string { - $maxAttempts = $this->config['busy_retry_attempts']; - $retryDelay = $this->config['busy_retry_delay']; - $lastException = null; - - for ($attempt = 1; $attempt <= $maxAttempts; ++$attempt) { - try { - return $this->doSendRawXml($xml, $debug); - } catch (ImmediateConnectionCloseException $e) { - $lastException = $e; - - if ($attempt < $maxAttempts) { - $this->logger->warning('BPN server busy, retrying raw XML request', [ - 'attempt' => $attempt, - 'maxAttempts' => $maxAttempts, - 'retryDelay' => $retryDelay, - ]); - sleep($retryDelay); - } - } - } - - $this->logger->error('BPN server busy after all retry attempts (raw XML)', [ - 'attempts' => $maxAttempts, - ]); - - throw $lastException; + return $this->executeWithRetry(fn () => $this->doSendRawXml($xml, $debug)); } /** @@ -572,25 +547,19 @@ class ApiClient $body = $doc->saveXML(); - $this->logger->info('Sending raw XML request to BPN API', [ - 'requestId' => $requestId, - 'type' => $type, - ]); - if (true === $debug || true === $this->config['debug']) { $this->dumpXmlToFile('request', $requestId, $body); } - $socket = $this->connect( - $this->config['bpn_api_ip'], - $this->config['bpn_api_port'], - $this->config['max_retries'], - $this->config['connection_timeout'], - $this->config['stream_timeout'], - $this->config['total_timeout'] - ); - $this->send($socket, $body, $this->config['total_timeout']); - $response = $this->receive($socket, $this->config['total_timeout']); + $socket = $this->connect(); + + $this->logger->info('Sending raw XML request to BPN API', [ + 'requestId' => $requestId, + 'type' => $type, + 'port' => $this->selectedPort, + ]); + $this->send($socket, $body); + $response = $this->receive($socket); $this->disconnect($socket); $responseXml = substr($response, 10); @@ -603,14 +572,22 @@ class ApiClient } /** - * Sends a request to the BPN API with automatic retry on immediate connection close. + * @throws ApiClientException + */ + private function sendRequest(string $type, array $data, array $additionalArgs = [], bool $debug = false): mixed + { + return $this->executeWithRetry(fn () => $this->doSendRequest($type, $data, $additionalArgs, $debug), $type); + } + + /** + * Executes an operation with automatic retry on immediate connection close. * * When the BPN server is busy, it may close connections immediately without responding. * This method detects such conditions and automatically retries after a short delay. * - * @throws ApiClientException + * @throws ImmediateConnectionCloseException If all retry attempts fail */ - private function sendRequest(string $type, array $data, array $additionalArgs = [], bool $debug = false): mixed + private function executeWithRetry(callable $operation, ?string $type = null): mixed { $maxAttempts = $this->config['busy_retry_attempts']; $retryDelay = $this->config['busy_retry_delay']; @@ -618,26 +595,30 @@ class ApiClient for ($attempt = 1; $attempt <= $maxAttempts; ++$attempt) { try { - return $this->doSendRequest($type, $data, $additionalArgs, $debug); + return $operation(); } catch (ImmediateConnectionCloseException $e) { $lastException = $e; if ($attempt < $maxAttempts) { - $this->logger->warning('BPN server busy, retrying request', [ + $context = [ 'attempt' => $attempt, 'maxAttempts' => $maxAttempts, 'retryDelay' => $retryDelay, - 'type' => $type, - ]); + ]; + if (null !== $type) { + $context['type'] = $type; + } + $this->logger->warning('BPN server busy, retrying request', $context); sleep($retryDelay); } } } - $this->logger->error('BPN server busy after all retry attempts', [ - 'attempts' => $maxAttempts, - 'type' => $type, - ]); + $context = ['attempts' => $maxAttempts]; + if (null !== $type) { + $context['type'] = $type; + } + $this->logger->error('BPN server busy after all retry attempts', $context); throw $lastException; } @@ -660,25 +641,20 @@ class ApiClient ]) ; - $this->logger->info('Sending request to BPN API', [ - 'requestId' => $requestId, - 'type' => $data['satz']['@typ'], - ]); - if (true === $debug || true === $this->config['debug']) { $this->dumpXmlToFile('request', $requestId, $body); } - $socket = $this->connect( - $this->config['bpn_api_ip'], - $this->config['bpn_api_port'], - $this->config['max_retries'], - $this->config['connection_timeout'], - $this->config['stream_timeout'], - $this->config['total_timeout'] - ); - $this->send($socket, $body, $this->config['total_timeout']); - $response = $this->receive($socket, $this->config['total_timeout']); + $socket = $this->connect(); + + $this->logger->info('Sending request to BPN API', [ + 'requestId' => $requestId, + 'type' => $data['satz']['@typ'], + 'port' => $this->selectedPort, + ]); + + $this->send($socket, $body); + $response = $this->receive($socket); $this->disconnect($socket); // message length (10 bytes) is prepended to actual message @@ -716,6 +692,146 @@ class ApiClient return md5($username.$password.$date.$type); } + /** + * @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'], + ]); + 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, + ]); + ++$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, + ]); + throw new ApiClientException('Unable to open socket'); + } + + return $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, + ]); + 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); + } + + /** + * @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, + ]); + 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, + ]); + 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, + ]); + throw new TimeoutException('Stream timeout while reading from socket'); + } + + $response .= $chunk; + } + + return $response; + } + + private function disconnect($socket): void + { + @fclose($socket); + } + private function resolveOptions(array $options): array { $optionsResolver = new OptionsResolver(); @@ -723,8 +839,9 @@ class ApiClient 'bpn_username', 'bpn_password', 'bpn_api_ip', - 'bpn_api_port', + 'bpn_api_ports', ]); + $optionsResolver->setAllowedTypes('bpn_api_ports', ['array']); $optionsResolver->setDefaults([ 'max_retries' => 25, 'debug' => false, diff --git a/src/BusProNet/Traits/ApiClientTrait.php b/src/BusProNet/Traits/ApiClientTrait.php deleted file mode 100644 index 5ea6b85..0000000 --- a/src/BusProNet/Traits/ApiClientTrait.php +++ /dev/null @@ -1,149 +0,0 @@ -operationStartTime = microtime(true); - - $tries = 1; - $errNo = $errStr = ''; - $errorCodesForRetry = [ - SOCKET_ECONNREFUSED, - SOCKET_EBADF, - ]; - - $openSocket = function (&$errNo, &$errStr) use ($host, $port, $connectionTimeout) { - return @fsockopen( - $host, - $port, - $errNo, - $errStr, - $connectionTimeout - ); - }; - - $socket = $openSocket($errNo, $errStr); - - while (false === $socket && true === in_array($errNo, $errorCodesForRetry) && $maxRetries > $tries) { - // Check if we've exceeded total timeout during retries - if (microtime(true) - $this->operationStartTime > $totalTimeout) { - $this->logger->error('Connection retry timeout exceeded', [ - 'elapsed_time' => microtime(true) - $this->operationStartTime, - 'total_timeout' => $totalTimeout, - ]); - 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, - ]); - ++$tries; - sleep(1); - $socket = $openSocket($errNo, $errStr); - } - - if (false !== $socket) { - stream_set_timeout($socket, $streamTimeout); - } else { - $this->logger->error('Unable to open socket', [ - 'error_message' => $errStr, - 'error_number' => $errNo, - 'elapsed_time' => microtime(true) - $this->operationStartTime, - ]); - throw new ApiClientException('Unable to open socket'); - } - - return $socket; - } - - /** - * @throws TimeoutException - */ - private function send($socket, string $data, int $totalTimeout): void - { - // Check total timeout before sending - if (microtime(true) - $this->operationStartTime > $totalTimeout) { - $this->logger->error('Total timeout exceeded before send', [ - 'elapsed_time' => microtime(true) - $this->operationStartTime, - ]); - 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); - } - - /** - * @throws TimeoutException - * @throws ImmediateConnectionCloseException - */ - private function receive($socket, int $totalTimeout): string - { - $response = ''; - $readAttempts = 0; - - 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, - ]); - 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, - ]); - 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, - ]); - throw new TimeoutException('Stream timeout while reading from socket'); - } - - $response .= $chunk; - } - - return $response; - } - - private function disconnect($socket): void - { - @fclose($socket); - } -}