diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 5cb12106a..72aa9568e 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -6,14 +6,14 @@ import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.RequestBody import org.slf4j.LoggerFactory +import ru.quipy.common.utils.SlidingWindowRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate import java.net.SocketTimeoutException import java.time.Duration import java.util.* +import java.util.concurrent.Semaphore - -// Advice: always treat time as a Duration class PaymentExternalSystemAdapterImpl( private val properties: PaymentAccountProperties, private val paymentESService: EventSourcingService, @@ -36,13 +36,33 @@ class PaymentExternalSystemAdapterImpl( private val client = OkHttpClient.Builder().build() + private val limiter = SlidingWindowRateLimiter( + rate = parallelRequests.toLong(), + window = Duration.ofMillis(4900) + /* + * rate limit payment system = 3 + * parallel reqs payment system = 5 + * time payment system = 4.9 + * 1 / 4.9 * 5 = 1.02 (сколько всего обрабаиывают все потоки в секунду) + * 3 / 1.02 (за какое время могут пройти 3 запроса) = 2.94 + */ + ) + + private val hardLimiter = SlidingWindowRateLimiter( + rate = rateLimitPerSec.toLong(), + window = Duration.ofSeconds(1) + ) + + // Семафор для ограничения параллельных вызовов + private val semaphore = Semaphore(parallelRequests, true) + override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { logger.warn("[$accountName] Submitting payment request for payment $paymentId") + limiter.tickBlocking() + val transactionId = UUID.randomUUID() - // Вне зависимости от исхода оплаты важно отметить что она была отправлена. - // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. paymentESService.update(paymentId) { it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } @@ -50,26 +70,31 @@ class PaymentExternalSystemAdapterImpl( logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") try { - val request = Request.Builder().run { - url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") - post(emptyBody) - }.build() - - client.newCall(request).execute().use { response -> - val body = try { - mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(),false, e.message) - } + semaphore.acquire() + try { + hardLimiter.tickBlocking() + + val request = Request.Builder().run { + url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") + post(emptyBody) + }.build() + + client.newCall(request).execute().use { response -> + val body = try { + mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) + } - logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) - paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) + paymentESService.update(paymentId) { + it.logProcessing(body.result, now(), transactionId, reason = body.message) + } } + } finally { + semaphore.release() } } catch (e: Exception) { when (e) { @@ -82,7 +107,6 @@ class PaymentExternalSystemAdapterImpl( else -> { logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) - paymentESService.update(paymentId) { it.logProcessing(false, now(), transactionId, reason = e.message) } @@ -96,7 +120,6 @@ class PaymentExternalSystemAdapterImpl( override fun isEnabled() = properties.enabled override fun name() = properties.accountName - } -public fun now() = System.currentTimeMillis() \ No newline at end of file +public fun now() = System.currentTimeMillis()