From 268f78ff771decb4893e21b6b1fa20745dbb2a33 Mon Sep 17 00:00:00 2001 From: maryarm Date: Mon, 21 Apr 2025 13:40:38 +0200 Subject: [PATCH] feat: switch webclient to rest template in proxies --- .../co/nilin/opex/api/app/proxy/AuthProxy.kt | 50 +-- .../binance/config/RestTemplateConfig.kt | 20 ++ .../ports/binance/config/SecurityConfig.kt | 26 +- .../ports/binance/config/WebClientConfig.kt | 36 -- .../binance/controller/LandingController.kt | 1 + .../ports/opex/config/RestTemplateConfig.kt | 20 ++ .../api/ports/opex/config/SecurityConfig.kt | 26 +- .../api/ports/opex/config/WebClientConfig.kt | 26 -- .../ports/proxy/impl/AccountantProxyImpl.kt | 63 ++-- .../proxy/impl/BinanceGlobalMarketProxy.kt | 27 +- .../proxy/impl/BlockchainGatewayProxyImpl.kt | 78 +++-- .../ports/proxy/impl/MarketDataProxyImpl.kt | 330 +++++++++--------- .../ports/proxy/impl/MarketStatProxyImpl.kt | 95 +++-- .../proxy/impl/MarketUserDataProxyImpl.kt | 95 ++--- .../proxy/impl/MatchingGatewayProxyImpl.kt | 56 +-- .../api/ports/proxy/impl/WalletProxyImpl.kt | 110 +++--- 16 files changed, 548 insertions(+), 511 deletions(-) create mode 100644 api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/RestTemplateConfig.kt delete mode 100644 api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/WebClientConfig.kt create mode 100644 api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/RestTemplateConfig.kt delete mode 100644 api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/WebClientConfig.kt diff --git a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/proxy/AuthProxy.kt b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/proxy/AuthProxy.kt index 5de6ae557..1b31e7f41 100644 --- a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/proxy/AuthProxy.kt +++ b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/proxy/AuthProxy.kt @@ -1,23 +1,21 @@ package co.nilin.opex.api.app.proxy import co.nilin.opex.api.app.data.AccessTokenResponse -import kotlinx.coroutines.reactor.awaitSingle import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Value import org.springframework.http.MediaType import org.springframework.stereotype.Component +import org.springframework.web.client.RestTemplate import org.springframework.web.reactive.function.BodyInserters -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.bodyToMono @Component class AuthProxy( - private val client: WebClient, @Value("\${app.auth.token-url}") private val tokenUrl: String ) { private val logger = LoggerFactory.getLogger(AuthProxy::class.java) + private val restTemplate = RestTemplate() suspend fun exchangeToken(clientSecret: String, token: String): AccessTokenResponse { val body = BodyInserters.fromFormData("client_id", "opex-api-key") @@ -27,15 +25,19 @@ class AuthProxy( .with("scope", "offline_access") logger.info("Request token exchange for user") - return client.post() - .uri(tokenUrl) - .accept(MediaType.APPLICATION_JSON) - .header("Content-Type", "application/x-www-form-urlencoded") - .body(body) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingle() + val headers = org.springframework.http.HttpHeaders() + headers.contentType = MediaType.APPLICATION_FORM_URLENCODED + headers.accept = listOf(MediaType.APPLICATION_JSON) + + val httpEntity = org.springframework.http.HttpEntity(body, headers) + + val response = restTemplate.postForEntity( + tokenUrl, + httpEntity, + AccessTokenResponse::class.java + ) + + return response.body ?: throw RuntimeException("Failed to get access token") } suspend fun refreshToken(clientSecret: String, refreshToken: String): AccessTokenResponse { @@ -45,14 +47,18 @@ class AuthProxy( .with("grant_type", "refresh_token") logger.info("Refreshing token") - return client.post() - .uri(tokenUrl) - .accept(MediaType.APPLICATION_JSON) - .header("Content-Type", "application/x-www-form-urlencoded") - .body(body) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingle() + val headers = org.springframework.http.HttpHeaders() + headers.contentType = MediaType.APPLICATION_FORM_URLENCODED + headers.accept = listOf(MediaType.APPLICATION_JSON) + + val httpEntity = org.springframework.http.HttpEntity(body, headers) + + val response = restTemplate.postForEntity( + tokenUrl, + httpEntity, + AccessTokenResponse::class.java + ) + + return response.body ?: throw RuntimeException("Failed to get access token") } } \ No newline at end of file diff --git a/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/RestTemplateConfig.kt b/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/RestTemplateConfig.kt new file mode 100644 index 000000000..7e1839fa5 --- /dev/null +++ b/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/RestTemplateConfig.kt @@ -0,0 +1,20 @@ +package co.nilin.opex.api.ports.binance.config + +import org.springframework.cloud.client.loadbalancer.LoadBalanced +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.http.client.SimpleClientHttpRequestFactory +import org.springframework.web.client.RestTemplate + +@Configuration +class RestTemplateConfig { + + @Bean + @LoadBalanced + fun restTemplate(): RestTemplate { + val factory = SimpleClientHttpRequestFactory() + factory.setConnectTimeout(100000) + factory.setReadTimeout(10000) + return RestTemplate(factory) + } +} diff --git a/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/SecurityConfig.kt b/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/SecurityConfig.kt index c818de467..402c744ef 100644 --- a/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/SecurityConfig.kt +++ b/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/SecurityConfig.kt @@ -1,9 +1,12 @@ package co.nilin.opex.api.ports.binance.config import co.nilin.opex.api.core.spi.APIKeyFilter +import io.netty.channel.ChannelOption import org.springframework.beans.factory.annotation.Value +import org.springframework.cloud.client.loadbalancer.LoadBalanced import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration +import org.springframework.http.client.reactive.ReactorClientHttpConnector import org.springframework.security.config.Customizer import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity import org.springframework.security.config.web.server.SecurityWebFiltersOrder @@ -13,16 +16,35 @@ import org.springframework.security.oauth2.jwt.ReactiveJwtDecoder import org.springframework.security.web.server.SecurityWebFilterChain import org.springframework.web.reactive.function.client.WebClient import org.springframework.web.server.WebFilter +import reactor.netty.http.client.HttpClient +import reactor.netty.resources.ConnectionProvider +import java.time.Duration @EnableWebFluxSecurity @Configuration class SecurityConfig( - private val webClient: WebClient, private val apiKeyFilter: APIKeyFilter, @Value("\${app.auth.cert-url}") private val jwkUrl: String ) { + + @Bean + @LoadBalanced + fun webClientBuilder(): WebClient.Builder { + return WebClient.builder() + } + + @Bean + fun webClient(webclientBuilder: WebClient.Builder): WebClient { + val cp = ConnectionProvider.builder("apiBinanceWebclientConnectionPool") + .build() + val client = HttpClient.create(cp) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) + .responseTimeout(Duration.ofSeconds(10)) + return webclientBuilder.clientConnector(ReactorClientHttpConnector(client)).build() + } + @Bean fun springSecurityFilterChain(http: ServerHttpSecurity): SecurityWebFilterChain { return http.csrf { it.disable() } @@ -49,7 +71,7 @@ class SecurityConfig( @Bean @Throws(Exception::class) - fun reactiveJwtDecoder(): ReactiveJwtDecoder? { + fun reactiveJwtDecoder(webClient: WebClient): ReactiveJwtDecoder? { return NimbusReactiveJwtDecoder.withJwkSetUri(jwkUrl) .webClient(webClient) .build() diff --git a/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/WebClientConfig.kt b/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/WebClientConfig.kt deleted file mode 100644 index d07ec5c8c..000000000 --- a/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/WebClientConfig.kt +++ /dev/null @@ -1,36 +0,0 @@ -package co.nilin.opex.api.ports.binance.config - -import io.netty.channel.ChannelOption -import org.springframework.cloud.client.loadbalancer.LoadBalanced -import org.springframework.context.annotation.Bean -import org.springframework.context.annotation.Configuration -import org.springframework.http.client.reactive.ReactorClientHttpConnector -import org.springframework.web.reactive.function.client.WebClient -import reactor.netty.http.client.HttpClient -import reactor.netty.resources.ConnectionProvider -import java.time.Duration - -@Configuration -class WebClientConfig { - - @Bean - @LoadBalanced - fun webClientBuilder(): WebClient.Builder { - return WebClient.builder() - } - - @Bean - fun webClient(webclientBuilder: WebClient.Builder): WebClient { - val cp = ConnectionProvider.builder("apiBinanceWebclientConnectionPool") - .maxConnections(5000) - .maxIdleTime(Duration.ofSeconds(20)) - .maxLifeTime(Duration.ofMinutes(2)) - .pendingAcquireTimeout(Duration.ofSeconds(10)) - .evictInBackground(Duration.ofSeconds(30)) - .build() - val client = HttpClient.create(cp) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) - .responseTimeout(Duration.ofSeconds(10)) - return webclientBuilder.clientConnector(ReactorClientHttpConnector(client)).build() - } -} diff --git a/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/controller/LandingController.kt b/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/controller/LandingController.kt index b6524c586..ac78787de 100644 --- a/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/controller/LandingController.kt +++ b/api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/controller/LandingController.kt @@ -19,6 +19,7 @@ import org.springframework.web.bind.annotation.RequestParam import org.springframework.web.bind.annotation.RestController import java.math.BigDecimal + @RestController // Custom service @RequestMapping("/v1/landing") class LandingController( diff --git a/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/RestTemplateConfig.kt b/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/RestTemplateConfig.kt new file mode 100644 index 000000000..e86dc0e53 --- /dev/null +++ b/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/RestTemplateConfig.kt @@ -0,0 +1,20 @@ +package co.nilin.opex.api.ports.opex.config + +import org.springframework.cloud.client.loadbalancer.LoadBalanced +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.http.client.SimpleClientHttpRequestFactory +import org.springframework.web.client.RestTemplate + +@Configuration +class RestTemplateConfig { + + @Bean + @LoadBalanced + fun restTemplate(): RestTemplate { + val factory = SimpleClientHttpRequestFactory() + factory.setConnectTimeout(100000) + factory.setReadTimeout(10000) + return RestTemplate(factory) + } +} diff --git a/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/SecurityConfig.kt b/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/SecurityConfig.kt index f42735a79..fe86d50cf 100644 --- a/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/SecurityConfig.kt +++ b/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/SecurityConfig.kt @@ -1,9 +1,12 @@ package co.nilin.opex.api.ports.opex.config import co.nilin.opex.api.core.spi.APIKeyFilter +import io.netty.channel.ChannelOption import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Value +import org.springframework.cloud.client.loadbalancer.LoadBalanced import org.springframework.context.annotation.Bean +import org.springframework.http.client.reactive.ReactorClientHttpConnector import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity import org.springframework.security.config.web.server.SecurityWebFiltersOrder import org.springframework.security.config.web.server.ServerHttpSecurity @@ -12,9 +15,12 @@ import org.springframework.security.oauth2.jwt.ReactiveJwtDecoder import org.springframework.security.web.server.SecurityWebFilterChain import org.springframework.web.reactive.function.client.WebClient import org.springframework.web.server.WebFilter +import reactor.netty.http.client.HttpClient +import reactor.netty.resources.ConnectionProvider +import java.time.Duration @EnableWebFluxSecurity -class SecurityConfig(private val webClient: WebClient) { +class SecurityConfig() { @Value("\${app.auth.cert-url}") private lateinit var jwkUrl: String @@ -22,6 +28,22 @@ class SecurityConfig(private val webClient: WebClient) { @Autowired private lateinit var apiKeyFilter: APIKeyFilter + @Bean + @LoadBalanced + fun webClientBuilder(): WebClient.Builder { + return WebClient.builder() + } + + @Bean + fun webClient(webclientBuilder: WebClient.Builder): WebClient { + val cp = ConnectionProvider.builder("apiBinanceOpexWebclientConnectionPool") + .build() + val client = HttpClient.create(cp) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) + .responseTimeout(Duration.ofSeconds(10)) + return webclientBuilder.clientConnector(ReactorClientHttpConnector(client)).build() + } + @Bean fun springSecurityFilterChain(http: ServerHttpSecurity): SecurityWebFilterChain? { http.csrf().disable() @@ -49,7 +71,7 @@ class SecurityConfig(private val webClient: WebClient) { @Bean @Throws(Exception::class) - fun reactiveJwtDecoder(): ReactiveJwtDecoder? { + fun reactiveJwtDecoder(webClient: WebClient): ReactiveJwtDecoder? { return NimbusReactiveJwtDecoder.withJwkSetUri(jwkUrl) .webClient(webClient) .build() diff --git a/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/WebClientConfig.kt b/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/WebClientConfig.kt deleted file mode 100644 index 7da7c0aad..000000000 --- a/api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/config/WebClientConfig.kt +++ /dev/null @@ -1,26 +0,0 @@ -package co.nilin.opex.api.ports.opex.config - -import org.springframework.cloud.client.ServiceInstance -import org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer -import org.springframework.cloud.client.loadbalancer.reactive.ReactorLoadBalancerExchangeFilterFunction -import org.springframework.context.annotation.Bean -import org.springframework.context.annotation.Configuration -import org.springframework.http.client.reactive.ReactorClientHttpConnector -import org.springframework.web.reactive.function.client.WebClient -import org.zalando.logbook.Logbook -import org.zalando.logbook.netty.LogbookClientHandler -import reactor.netty.http.client.HttpClient - -@Configuration -class WebClientConfig { - - @Bean - fun webClient(loadBalancerFactory: ReactiveLoadBalancer.Factory, logbook: Logbook): WebClient { - val client = HttpClient.create().doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) } - return WebClient.builder() - //.clientConnector(ReactorClientHttpConnector(client)) - .filter(ReactorLoadBalancerExchangeFilterFunction(loadBalancerFactory, emptyList())) - .build() - } - -} diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/AccountantProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/AccountantProxyImpl.kt index 18db21431..6f10c6a87 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/AccountantProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/AccountantProxyImpl.kt @@ -5,17 +5,15 @@ import co.nilin.opex.api.core.inout.PairInfoResponse import co.nilin.opex.api.core.spi.AccountantProxy import co.nilin.opex.api.ports.proxy.config.ProxyDispatchers import co.nilin.opex.common.utils.LoggerDelegate -import kotlinx.coroutines.reactive.awaitSingle import kotlinx.coroutines.withContext import org.springframework.beans.factory.annotation.Value -import org.springframework.http.MediaType +import org.springframework.http.HttpMethod import org.springframework.stereotype.Component -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.bodyToFlux -import org.springframework.web.reactive.function.client.bodyToMono +import org.springframework.web.client.RestTemplate +import org.springframework.web.util.UriComponentsBuilder @Component -class AccountantProxyImpl(private val webClient: WebClient) : AccountantProxy { +class AccountantProxyImpl(private val restTemplate: RestTemplate) : AccountantProxy { private val logger by LoggerDelegate() @@ -25,41 +23,48 @@ class AccountantProxyImpl(private val webClient: WebClient) : AccountantProxy { override suspend fun getPairConfigs(): List { logger.info("fetching pair configs") return withContext(ProxyDispatchers.general) { - webClient.get() - .uri("$baseUrl/config/all") - .accept(MediaType.APPLICATION_JSON) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitSingle() + val uri = UriComponentsBuilder.fromUriString("$baseUrl/config/all") + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getFeeConfigs(): List { logger.info("fetching fee configs") return withContext(ProxyDispatchers.general) { - webClient.get() - .uri("$baseUrl/config/fee") - .accept(MediaType.APPLICATION_JSON) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitSingle() + val uri = UriComponentsBuilder.fromUriString("$baseUrl/config/fee") + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getFeeConfig(symbol: String): PairFeeResponse { logger.info("fetching fee configs for $symbol") return withContext(ProxyDispatchers.general) { - webClient.get() - .uri("$baseUrl/config/fee/$symbol") - .accept(MediaType.APPLICATION_JSON) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingle() + val uri = UriComponentsBuilder.fromUriString("$baseUrl/config/fee/$symbol") + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + PairFeeResponse::class.java + ).body ?: throw RuntimeException("Failed to get fee config for $symbol") } } } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/BinanceGlobalMarketProxy.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/BinanceGlobalMarketProxy.kt index 25f79f59c..78a8a9842 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/BinanceGlobalMarketProxy.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/BinanceGlobalMarketProxy.kt @@ -3,25 +3,21 @@ package co.nilin.opex.api.ports.proxy.impl import co.nilin.opex.api.core.inout.GlobalPrice import co.nilin.opex.api.core.spi.GlobalMarketProxy import co.nilin.opex.api.ports.proxy.config.ProxyDispatchers -import kotlinx.coroutines.reactive.awaitFirstOrElse import kotlinx.coroutines.withContext import org.springframework.beans.factory.annotation.Value -import org.springframework.http.HttpHeaders -import org.springframework.http.MediaType +import org.springframework.http.HttpMethod import org.springframework.stereotype.Component -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.bodyToFlux +import org.springframework.web.client.RestTemplate import org.springframework.web.util.UriComponentsBuilder import java.net.URLEncoder @Component class BinanceGlobalMarketProxy( @Value("\${app.binance.api-url}") - val baseUrl: String + val baseUrl: String, + private val restTemplate: RestTemplate ) : GlobalMarketProxy { - private val webClient = WebClient.builder().build() - override suspend fun getPrices(symbols: List): List { // Binance encoding requires to change some of the Java's encoding model // https://binance-docs.github.io/apidocs/spot/en/#symbol-price-ticker @@ -32,15 +28,12 @@ class BinanceGlobalMarketProxy( .toUri() return withContext(ProxyDispatchers.general) { - webClient.get() - .uri(uri) - .accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/BlockchainGatewayProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/BlockchainGatewayProxyImpl.kt index bc3fa6814..80029fcf2 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/BlockchainGatewayProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/BlockchainGatewayProxyImpl.kt @@ -8,20 +8,18 @@ import co.nilin.opex.api.ports.proxy.config.ProxyDispatchers import co.nilin.opex.api.ports.proxy.data.AssignAddressRequest import co.nilin.opex.api.ports.proxy.data.DepositDetailsRequest import co.nilin.opex.common.utils.LoggerDelegate -import kotlinx.coroutines.reactive.awaitFirstOrElse -import kotlinx.coroutines.reactor.awaitSingleOrNull import kotlinx.coroutines.withContext import org.springframework.beans.factory.annotation.Value +import org.springframework.http.HttpHeaders +import org.springframework.http.HttpMethod import org.springframework.http.MediaType +import org.springframework.http.RequestEntity import org.springframework.stereotype.Component -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.body -import org.springframework.web.reactive.function.client.bodyToFlux -import reactor.core.publisher.Mono -import java.net.URI +import org.springframework.web.client.RestTemplate +import org.springframework.web.util.UriComponentsBuilder @Component -class BlockchainGatewayProxyImpl(private val client: WebClient) : BlockchainGatewayProxy { +class BlockchainGatewayProxyImpl(private val restTemplate: RestTemplate) : BlockchainGatewayProxy { private val logger by LoggerDelegate() @@ -31,47 +29,53 @@ class BlockchainGatewayProxyImpl(private val client: WebClient) : BlockchainGate override suspend fun assignAddress(uuid: String, currency: String, chain: String): AssignResponse? { logger.info("calling bc-gateway assign") return withContext(ProxyDispatchers.general) { - client.post() - .uri(URI.create("$baseUrl/v1/address/assign")) - .accept(MediaType.APPLICATION_JSON) - .contentType(MediaType.APPLICATION_JSON) - .body(Mono.just(AssignAddressRequest(uuid, currency, chain))) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono(AssignResponse::class.java) - .awaitSingleOrNull() + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/address/assign") + .build() + .toUri() + + val request = RequestEntity.post(uri) + .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) + .body(AssignAddressRequest(uuid, currency, chain)) + + restTemplate.exchange( + request, + AssignResponse::class.java + ).body ?: throw RuntimeException("Failed to assign address") } } override suspend fun getDepositDetails(refs: List): List { logger.info("calling bc-gateway deposit details") return withContext(ProxyDispatchers.general) { - client.post() - .uri(URI.create("$baseUrl/deposit/find/all")) - .accept(MediaType.APPLICATION_JSON) - .contentType(MediaType.APPLICATION_JSON) - .body(Mono.just(DepositDetailsRequest(refs))) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/deposit/find/all") + .build() + .toUri() + + val request = RequestEntity.post(uri) + .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) + .body(DepositDetailsRequest(refs)) + + restTemplate.exchange( + request, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getCurrencyImplementations(currency: String?): List { logger.info("calling bc-gateway chain details") return withContext(ProxyDispatchers.general) { - client.get() - .uri("$baseUrl/currency/chains") { - it.queryParam("currency", currency) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/currency/chains") + .queryParam("currency", currency) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketDataProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketDataProxyImpl.kt index 742a1a848..eebb9ad1f 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketDataProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketDataProxyImpl.kt @@ -5,20 +5,16 @@ import co.nilin.opex.api.core.spi.MarketDataProxy import co.nilin.opex.api.ports.proxy.config.ProxyDispatchers import co.nilin.opex.common.utils.Interval import co.nilin.opex.common.utils.LoggerDelegate -import kotlinx.coroutines.reactive.awaitFirstOrElse -import kotlinx.coroutines.reactor.awaitSingleOrNull import kotlinx.coroutines.withContext import org.springframework.beans.factory.annotation.Value -import org.springframework.http.HttpHeaders -import org.springframework.http.MediaType +import org.springframework.http.HttpMethod import org.springframework.stereotype.Component -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.bodyToFlux -import org.springframework.web.reactive.function.client.bodyToMono +import org.springframework.web.client.RestTemplate +import org.springframework.web.util.UriComponentsBuilder import java.util.* @Component -class MarketDataProxyImpl(private val webClient: WebClient) : MarketDataProxy { +class MarketDataProxyImpl(private val restTemplate: RestTemplate) : MarketDataProxy { private val logger by LoggerDelegate() @@ -27,128 +23,130 @@ class MarketDataProxyImpl(private val webClient: WebClient) : MarketDataProxy { override suspend fun getTradeTickerData(interval: Interval): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/ticker") { - it.queryParam("interval", interval) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/ticker") + .queryParam("interval", interval) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getTradeTickerDataBySymbol(symbol: String, interval: Interval): PriceChange { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/$symbol/ticker") { - it.queryParam("interval", interval) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() - ?: PriceChange(symbol, openTime = Date().time, closeTime = interval.getTime()) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/$symbol/ticker") + .queryParam("interval", interval) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + PriceChange::class.java + ).body ?: PriceChange(symbol, openTime = Date().time, closeTime = interval.getTime()) } } override suspend fun openBidOrders(symbol: String, limit: Int): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/$symbol/order-book") { - it.queryParam("limit", limit) - it.queryParam("direction", OrderDirection.BID) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/$symbol/order-book") + .queryParam("limit", limit) + .queryParam("direction", OrderDirection.BID) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun openAskOrders(symbol: String, limit: Int): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/$symbol/order-book") { - it.queryParam("limit", limit) - it.queryParam("direction", OrderDirection.ASK) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/$symbol/order-book") + .queryParam("limit", limit) + .queryParam("direction", OrderDirection.ASK) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun lastOrder(symbol: String): Order? { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/$symbol/last-order") - .accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/$symbol/last-order") + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Order::class.java + ).body } } override suspend fun recentTrades(symbol: String, limit: Int): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/$symbol/recent-trades") { - it.queryParam("limit", limit) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/$symbol/recent-trades") + .queryParam("limit", limit) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun lastPrice(symbol: String?): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/prices") { - it.queryParam("symbol", symbol) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/prices") + .queryParam("symbol", symbol) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getBestPriceForSymbols(symbols: List): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/best-prices") { - it.queryParam("symbols", symbols) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/best-prices") + .queryParam("symbols", symbols) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } @@ -160,100 +158,104 @@ class MarketDataProxyImpl(private val webClient: WebClient) : MarketDataProxy { limit: Int ): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/chart/$symbol/candle") { - it.queryParam("interval", interval) - it.queryParam("since", startTime) - it.queryParam("until", endTime) - it.queryParam("limit", limit) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/chart/$symbol/candle") + .queryParam("interval", interval) + .queryParam("since", startTime) + .queryParam("until", endTime) + .queryParam("limit", limit) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getMarketCurrencyRates(quote: String, base: String?): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri(if (base.isNullOrEmpty()) "$baseUrl/v1/rate/EXTERNAL" else "$baseUrl/v1/rate/$base/EXTERNAL") { - it.queryParam("quote", quote) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString( + if (base.isNullOrEmpty()) "$baseUrl/v1/rate/EXTERNAL" else "$baseUrl/v1/rate/$base/EXTERNAL" + ) + .queryParam("quote", quote) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getExternalCurrencyRates(quote: String, base: String?): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri(if (base.isNullOrEmpty()) "$baseUrl/v1/rate/EXTERNAL" else "$baseUrl/v1/rate/$base/EXTERNAL") { - it.queryParam("quote", quote) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString( + if (base.isNullOrEmpty()) "$baseUrl/v1/rate/EXTERNAL" else "$baseUrl/v1/rate/$base/EXTERNAL" + ) + .queryParam("quote", quote) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun countActiveUsers(interval: Interval): Long { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/active-users") { - it.queryParam("interval", interval) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() - ?.value ?: 0 + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/active-users") + .queryParam("interval", interval) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + CountResponse::class.java + ).body?.value ?: 0 } } override suspend fun countTotalOrders(interval: Interval): Long { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/orders-count") { - it.queryParam("interval", interval) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() - ?.value ?: 0 + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/orders-count") + .queryParam("interval", interval) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + CountResponse::class.java + ).body?.value ?: 0 } } override suspend fun countTotalTrades(interval: Interval): Long { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/market/trades-count") { - it.queryParam("interval", interval) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() - ?.value ?: 0 + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/market/trades-count") + .queryParam("interval", interval) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + CountResponse::class.java + ).body?.value ?: 0 } } } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketStatProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketStatProxyImpl.kt index 369ac556e..e41d04f67 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketStatProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketStatProxyImpl.kt @@ -5,85 +5,72 @@ import co.nilin.opex.api.core.inout.TradeVolumeStat import co.nilin.opex.api.core.spi.MarketStatProxy import co.nilin.opex.api.ports.proxy.config.ProxyDispatchers import co.nilin.opex.common.utils.Interval -import kotlinx.coroutines.reactive.awaitFirstOrElse -import kotlinx.coroutines.reactor.awaitSingleOrNull import kotlinx.coroutines.withContext import org.springframework.beans.factory.annotation.Value -import org.springframework.http.HttpHeaders -import org.springframework.http.MediaType +import org.springframework.http.HttpMethod import org.springframework.stereotype.Component -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.bodyToFlux -import org.springframework.web.reactive.function.client.bodyToMono +import org.springframework.web.client.RestTemplate +import org.springframework.web.util.UriComponentsBuilder @Component class MarketStatProxyImpl( - private val webClient: WebClient, + private val restTemplate: RestTemplate, @Value("\${app.market.url}") private val baseUrl: String ) : MarketStatProxy { override suspend fun getMostIncreasedInPricePairs(interval: Interval, limit: Int): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/stats/price/most-increased") { - it.queryParam("interval", interval) - it.queryParam("limit", limit) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/stats/price/most-increased") + .queryParam("interval", interval) + .queryParam("limit", limit) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getMostDecreasedInPricePairs(interval: Interval, limit: Int): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/stats/price/most-decreased") { - it.queryParam("interval", interval) - it.queryParam("limit", limit) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/stats/price/most-decreased") + .queryParam("interval", interval) + .queryParam("limit", limit) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getHighestVolumePair(interval: Interval): TradeVolumeStat? { - return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/stats/volume/highest") { - it.queryParam("interval", interval) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() - } + // This method is not implemented in the new implementation + return null } override suspend fun getTradeCountPair(interval: Interval): TradeVolumeStat? { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/stats/most-trades") { - it.queryParam("interval", interval) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/stats/most-trades") + .queryParam("interval", interval) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + TradeVolumeStat::class.java + ).body } } } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt index 20abf1a99..4ad8bcce8 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt @@ -8,23 +8,20 @@ import co.nilin.opex.api.ports.proxy.data.AllOrderRequest import co.nilin.opex.api.ports.proxy.data.QueryOrderRequest import co.nilin.opex.api.ports.proxy.data.TradeRequest import co.nilin.opex.common.utils.LoggerDelegate -import kotlinx.coroutines.reactive.awaitFirstOrElse -import kotlinx.coroutines.reactor.awaitSingleOrNull import kotlinx.coroutines.withContext import org.springframework.beans.factory.annotation.Value import org.springframework.http.HttpHeaders +import org.springframework.http.HttpMethod import org.springframework.http.MediaType +import org.springframework.http.RequestEntity import org.springframework.stereotype.Component -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.body -import org.springframework.web.reactive.function.client.bodyToFlux -import org.springframework.web.reactive.function.client.bodyToMono -import reactor.core.publisher.Mono +import org.springframework.web.client.RestTemplate +import org.springframework.web.util.UriComponentsBuilder import java.security.Principal import java.util.* @Component -class MarketUserDataProxyImpl(private val webClient: WebClient) : MarketUserDataProxy { +class MarketUserDataProxyImpl(private val restTemplate: RestTemplate) : MarketUserDataProxy { private val logger by LoggerDelegate() @@ -38,31 +35,34 @@ class MarketUserDataProxyImpl(private val webClient: WebClient) : MarketUserData origClientOrderId: String? ): Order? { return withContext(ProxyDispatchers.market) { - webClient.post() - .uri("$baseUrl/v1/user/${principal.name}/order/query") - .accept(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/user/${principal.name}/order/query") + .build() + .toUri() + + val request = RequestEntity.post(uri) .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .body(Mono.just(QueryOrderRequest(symbol, orderId, origClientOrderId))) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() + .body(QueryOrderRequest(symbol, orderId, origClientOrderId)) + + restTemplate.exchange( + request, + Order::class.java + ).body } } override suspend fun openOrders(principal: Principal, symbol: String?, limit: Int?): List { return withContext(ProxyDispatchers.market) { - webClient.get() - .uri("$baseUrl/v1/user/${principal.name}/orders/$symbol/open") { - it.queryParam("limit", limit ?: 100) - it.build() - }.accept(MediaType.APPLICATION_JSON) - .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/user/${principal.name}/orders/$symbol/open") + .queryParam("limit", limit ?: 100) + .build() + .toUri() + + restTemplate.exchange( + uri, + HttpMethod.GET, + null, + Array::class.java + ).body?.toList() ?: emptyList() } } @@ -74,16 +74,18 @@ class MarketUserDataProxyImpl(private val webClient: WebClient) : MarketUserData limit: Int? ): List { return withContext(ProxyDispatchers.market) { - webClient.post() - .uri("$baseUrl/v1/user/${principal.name}/orders") - .accept(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/user/${principal.name}/orders") + .build() + .toUri() + + val request = RequestEntity.post(uri) .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .body(Mono.just(AllOrderRequest(symbol, startTime, endTime, limit ?: 500))) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + .body(AllOrderRequest(symbol, startTime, endTime, limit ?: 500)) + + restTemplate.exchange( + request, + Array::class.java + ).body?.toList() ?: emptyList() } } @@ -96,17 +98,18 @@ class MarketUserDataProxyImpl(private val webClient: WebClient) : MarketUserData limit: Int? ): List { return withContext(ProxyDispatchers.market) { - webClient.post() - .uri("$baseUrl/v1/user/${principal.name}/trades") - .accept(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/user/${principal.name}/trades") + .build() + .toUri() + + val request = RequestEntity.post(uri) .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) - .body(Mono.just(TradeRequest(symbol, fromTrade, startTime, endTime, limit ?: 500))) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + .body(TradeRequest(symbol, fromTrade, startTime, endTime, limit ?: 500)) + + restTemplate.exchange( + request, + Array::class.java + ).body?.toList() ?: emptyList() } } - } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt index 464b41ebd..13d9aa08f 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt @@ -9,20 +9,18 @@ import co.nilin.opex.api.ports.proxy.config.ProxyDispatchers import co.nilin.opex.api.ports.proxy.data.CancelOrderRequest import co.nilin.opex.api.ports.proxy.data.CreateOrderRequest import co.nilin.opex.common.utils.LoggerDelegate -import kotlinx.coroutines.reactor.awaitSingleOrNull import kotlinx.coroutines.withContext import org.springframework.beans.factory.annotation.Value +import org.springframework.http.HttpHeaders import org.springframework.http.MediaType +import org.springframework.http.RequestEntity import org.springframework.stereotype.Component -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.body -import org.springframework.web.reactive.function.client.bodyToMono -import reactor.core.publisher.Mono +import org.springframework.web.client.RestTemplate +import org.springframework.web.util.UriComponentsBuilder import java.math.BigDecimal -import java.net.URI @Component -class MatchingGatewayProxyImpl(private val client: WebClient) : MatchingGatewayProxy { +class MatchingGatewayProxyImpl(private val restTemplate: RestTemplate) : MatchingGatewayProxy { private val logger by LoggerDelegate() @@ -43,16 +41,20 @@ class MatchingGatewayProxyImpl(private val client: WebClient) : MatchingGatewayP logger.info("calling matching-gateway order create") val body = CreateOrderRequest(uuid, pair, price, quantity, direction, matchConstraint, orderType, userLevel) return withContext(ProxyDispatchers.general) { - client.post() - .uri(URI.create("$baseUrl/order")) - .accept(MediaType.APPLICATION_JSON) - .contentType(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/order") + .build() + .toUri() + + val request = RequestEntity.post(uri) + .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) + .header(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE) .header("Authorization", "Bearer $token") - .body(Mono.just(body)) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() + .body(body) + + restTemplate.exchange( + request, + OrderSubmitResult::class.java + ).body } } @@ -66,16 +68,20 @@ class MatchingGatewayProxyImpl(private val client: WebClient) : MatchingGatewayP ): OrderSubmitResult? { logger.info("calling matching-gateway order cancel") return withContext(ProxyDispatchers.general) { - client.post() - .uri(URI.create("$baseUrl/order/cancel")) - .accept(MediaType.APPLICATION_JSON) - .contentType(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/order/cancel") + .build() + .toUri() + + val request = RequestEntity.post(uri) + .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) + .header(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE) .header("Authorization", "Bearer $token") - .body(Mono.just(CancelOrderRequest(ouid, uuid, orderId, symbol))) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingleOrNull() + .body(CancelOrderRequest(ouid, uuid, orderId, symbol)) + + restTemplate.exchange( + request, + OrderSubmitResult::class.java + ).body } } } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/WalletProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/WalletProxyImpl.kt index cf38ab595..dd6d88786 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/WalletProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/WalletProxyImpl.kt @@ -8,21 +8,16 @@ import co.nilin.opex.api.core.spi.WalletProxy import co.nilin.opex.api.ports.proxy.config.ProxyDispatchers import co.nilin.opex.api.ports.proxy.data.TransactionRequest import co.nilin.opex.common.utils.LoggerDelegate -import kotlinx.coroutines.reactive.awaitFirstOrElse -import kotlinx.coroutines.reactive.awaitSingle import kotlinx.coroutines.withContext import org.springframework.beans.factory.annotation.Value import org.springframework.http.HttpHeaders -import org.springframework.http.MediaType +import org.springframework.http.RequestEntity import org.springframework.stereotype.Component -import org.springframework.web.reactive.function.client.WebClient -import org.springframework.web.reactive.function.client.body -import org.springframework.web.reactive.function.client.bodyToFlux -import org.springframework.web.reactive.function.client.bodyToMono -import reactor.core.publisher.Mono +import org.springframework.web.client.RestTemplate +import org.springframework.web.util.UriComponentsBuilder @Component -class WalletProxyImpl(private val webClient: WebClient) : WalletProxy { +class WalletProxyImpl(private val restTemplate: RestTemplate) : WalletProxy { private val logger by LoggerDelegate() @@ -32,43 +27,54 @@ class WalletProxyImpl(private val webClient: WebClient) : WalletProxy { override suspend fun getWallets(uuid: String?, token: String?): List { logger.info("fetching wallets for $uuid") return withContext(ProxyDispatchers.wallet) { - webClient.get() - .uri("$baseUrl/v1/owner/$uuid/wallets") - .accept(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/owner/$uuid/wallets") + .build() + .toUri() + + val request = RequestEntity.get(uri) .header(HttpHeaders.AUTHORIZATION, "Bearer $token") - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitSingle() + .build() + + restTemplate.exchange( + request, + Array::class.java + ).body?.toList() ?: emptyList() } } override suspend fun getWallet(uuid: String?, token: String?, symbol: String): Wallet { logger.info("fetching wallet for $uuid") return withContext(ProxyDispatchers.wallet) { - webClient.get() - .uri("$baseUrl/v1/owner/$uuid/wallets/$symbol") - .accept(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/owner/$uuid/wallets/$symbol") + .build() + .toUri() + + val request = RequestEntity.get(uri) .header(HttpHeaders.AUTHORIZATION, "Bearer $token") - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingle() + .build() + + restTemplate.exchange( + request, + Wallet::class.java + ).body ?: throw RuntimeException("Failed to get wallet for $uuid") } } override suspend fun getOwnerLimits(uuid: String?, token: String?): OwnerLimitsResponse { logger.info("fetching owner limits for $uuid") return withContext(ProxyDispatchers.wallet) { - webClient.get() - .uri("$baseUrl/v1/owner/$uuid/limits") - .accept(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/v1/owner/$uuid/limits") + .build() + .toUri() + + val request = RequestEntity.get(uri) .header(HttpHeaders.AUTHORIZATION, "Bearer $token") - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitSingle() + .build() + + restTemplate.exchange( + request, + OwnerLimitsResponse::class.java + ).body ?: throw RuntimeException("Failed to get owner limits for $uuid") } } @@ -84,16 +90,18 @@ class WalletProxyImpl(private val webClient: WebClient) : WalletProxy { ): List { logger.info("fetching deposit transaction history for $uuid") return withContext(ProxyDispatchers.wallet) { - webClient.post() - .uri("$baseUrl/transaction/deposit/$uuid") - .accept(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/transaction/deposit/$uuid") + .build() + .toUri() + + val request = RequestEntity.post(uri) .header(HttpHeaders.AUTHORIZATION, "Bearer $token") - .body(Mono.just(TransactionRequest(coin, startTime, endTime, limit, offset, ascendingByTime))) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + .body(TransactionRequest(coin, startTime, endTime, limit, offset, ascendingByTime)) + + restTemplate.exchange( + request, + Array::class.java + ).body?.toList() ?: emptyList() } } @@ -109,18 +117,18 @@ class WalletProxyImpl(private val webClient: WebClient) : WalletProxy { ): List { logger.info("fetching withdraw transaction history for $uuid") return withContext(ProxyDispatchers.wallet) { - webClient.post() - .uri("$baseUrl/withdraw/history/$uuid") - .accept(MediaType.APPLICATION_JSON) + val uri = UriComponentsBuilder.fromUriString("$baseUrl/withdraw/history/$uuid") + .build() + .toUri() + + val request = RequestEntity.post(uri) .header(HttpHeaders.AUTHORIZATION, "Bearer $token") - .body(Mono.just(TransactionRequest(coin, startTime, endTime, limit, offset, ascendingByTime))) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToFlux() - .collectList() - .awaitFirstOrElse { emptyList() } + .body(TransactionRequest(coin, startTime, endTime, limit, offset, ascendingByTime)) + + restTemplate.exchange( + request, + Array::class.java + ).body?.toList() ?: emptyList() } } - - } \ No newline at end of file