EP/myep
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: load balancing of api requests, cleanup api client

Browse Source
This commit is contained in:
Björn Fromme
2026-03-16 12:01:09 +01:00
parent ab48b1f867
commit 1fe8a2cb17
4 changed files with 190 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env
+1 -1
View File
@@ -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
config/services.yaml
+1 -1
View File
@@ -60,7 +60,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)%'
src/BusProNet/ApiClient.php
+188 -71
View File
@@ -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,
src/BusProNet/Traits/ApiClientTrait.php
-149
View File
@@ -1,149 +0,0 @@
<?php
namespace App\BusProNet\Traits;
use App\BusProNet\Exception\ApiClientException;
use App\BusProNet\Exception\ImmediateConnectionCloseException;
use App\BusProNet\Exception\TimeoutException;
trait ApiClientTrait
{
private float $operationStartTime;
/**
* @throws ApiClientException
* @throws TimeoutException
*/
private function connect(string $host, int $port, int $maxRetries = 25, int $connectionTimeout = 5, int $streamTimeout = 30, int $totalTimeout = 45)
{
$this->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);
}
}