From aa9081ccf32e1978bac65a8a3e4038fa6e45a0b3 Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Thu, 22 Jun 2023 12:38:20 +0300 Subject: [PATCH] =?UTF-8?q?=D0=B2=D0=B2=D0=BE=D0=B4=D1=8B/=D0=B2=D1=8B?= =?UTF-8?q?=D0=B2=D0=BE=D0=B4=D1=8B=20-=20gateway?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../gatewayapi/config/BeanConfiguration.java | 9 +++ .../controller/GatewayController.java | 28 ++++++++- .../request/operations/OperationsRequest.java | 38 ++++++++++++ .../request/operations/SentAsset.java | 42 +++++++++++++ .../gatewayapi/service/GatewayService.java | 39 ++++++------ .../gatewayapi/service/OperationService.java | 61 +++++++++++++++++++ 6 files changed, 197 insertions(+), 20 deletions(-) create mode 100644 clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/OperationsRequest.java create mode 100644 clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/SentAsset.java create mode 100644 clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/OperationService.java diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/BeanConfiguration.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/BeanConfiguration.java index de1aebb14..edd2cde73 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/BeanConfiguration.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/BeanConfiguration.java @@ -4,6 +4,7 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Scope; +import ru.spcex.clearing.gatewayapi.controller.request.operations.SentAsset; import ru.spcex.clearing.gatewayapi.errors.GatewayError; import ru.spcex.clearing.util.security.UserRoleVerification; import ru.spcex.clearing.util.services.IMDGMessageResolver; @@ -11,6 +12,9 @@ import ru.spcex.platform.enumeration.UserRole; import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.utils.enumeration.IMessageResolver; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -32,4 +36,9 @@ public class BeanConfiguration { public ExecutorService executor() { return Executors.newFixedThreadPool(10); } + + @Bean("sentAssets") + public Map> sentAssets() { + return new ConcurrentHashMap<>(); + } } diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/GatewayController.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/GatewayController.java index 67e934dac..866476cc9 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/GatewayController.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/GatewayController.java @@ -15,8 +15,10 @@ import ru.spcex.clearing.gatewayapi.controller.request.company.CompaniesRequest; import ru.spcex.clearing.gatewayapi.controller.request.limit.LimitRequest; import ru.spcex.clearing.gatewayapi.controller.request.listing.fond.FondListingsRequest; import ru.spcex.clearing.gatewayapi.controller.request.listing.mkr.MMListingsRequest; +import ru.spcex.clearing.gatewayapi.controller.request.operations.OperationsRequest; import ru.spcex.clearing.gatewayapi.controller.response.CommonResponse; import ru.spcex.clearing.gatewayapi.exception.GatewayException; +import ru.spcex.clearing.gatewayapi.service.OperationService; import ru.spcex.clearing.gatewayapi.service.processor.CompanyProcessor; import ru.spcex.clearing.gatewayapi.service.processor.ListingFondProcessor; import ru.spcex.clearing.gatewayapi.service.processor.ListingMMProcessor; @@ -39,6 +41,7 @@ public class GatewayController { private final ListingFondProcessor listingFondProcessor; private final ListingMMProcessor listingMMProcessor; private final ExecutorService executor; + private final OperationService operationService; private List validTypes = Arrays.asList("DAY_START", "ON_DEMAND"); @@ -46,12 +49,14 @@ public class GatewayController { CompanyProcessor companiesProcessor, ListingFondProcessor listingFondProcessor, ListingMMProcessor listingMMProcessor, - @Qualifier("gatewayExecutor") ExecutorService executor) { + @Qualifier("gatewayExecutor") ExecutorService executor, + OperationService operationService) { this.messageResolver = messageResolver; this.companiesProcessor = companiesProcessor; this.listingFondProcessor = listingFondProcessor; this.listingMMProcessor = listingMMProcessor; this.executor = executor; + this.operationService = operationService; } @@ -125,7 +130,24 @@ public class GatewayController { @ResponseBody public CommonResponse limits(@RequestBody LimitRequest request) { log.debug("Received message: {}", request); - return createResponseByLimit(request.getId()); + return createCommonResponse(request.getId()); + } + + @ApiResponses(value = { + @ApiResponse(code = 200, message = "OK", response = CommonResponse.class), + @ApiResponse(code = 400, message = "Ошибка валидации", response = CommonResponse.class) + }) + @RequestMapping( + path = "/operations", + method = RequestMethod.POST, + consumes = MediaType.APPLICATION_JSON_VALUE, + produces = MediaType.APPLICATION_JSON_VALUE + ) + @ResponseBody + public CommonResponse operations(@RequestBody OperationsRequest request) { + log.debug("Received message: {}", request); + executor.submit(() -> operationService.checkAssetsAndSendMessage(request)); + return createCommonResponse(request.getId()); } @ResponseStatus(value = HttpStatus.BAD_REQUEST) @@ -151,7 +173,7 @@ public class GatewayController { return commonResponse; } - private CommonResponse createResponseByLimit(UUID uuid) { + private CommonResponse createCommonResponse(UUID uuid) { CommonResponse commonResponse = new CommonResponse(); commonResponse.setId(uuid); // commonResponse.setType(request.getType()); diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/OperationsRequest.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/OperationsRequest.java new file mode 100644 index 000000000..01d412d9e --- /dev/null +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/OperationsRequest.java @@ -0,0 +1,38 @@ +package ru.spcex.clearing.gatewayapi.controller.request.operations; + +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.UUID; + +public class OperationsRequest { + @JsonProperty("id") + private UUID id; + @JsonProperty("result") + private Boolean result; + @JsonProperty("description") + private String description; + + public UUID getId() { + return id; + } + + public void setId(UUID id) { + this.id = id; + } + + public Boolean getResult() { + return result; + } + + public void setResult(Boolean result) { + this.result = result; + } + + public String getDescription() { + return description; + } + + public void setDescription(String description) { + this.description = description; + } +} diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/SentAsset.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/SentAsset.java new file mode 100644 index 000000000..d98d7c74c --- /dev/null +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/SentAsset.java @@ -0,0 +1,42 @@ +package ru.spcex.clearing.gatewayapi.controller.request.operations; + +import java.util.UUID; + +public class SentAsset { + private boolean isReceivedResponse = false; + private UUID id; + private Long statementId; + private OperationsRequest operationsRequest; + + public boolean isReceivedResponse() { + return isReceivedResponse; + } + + public void setReceivedResponse(boolean receivedResponse) { + isReceivedResponse = receivedResponse; + } + + public UUID getId() { + return id; + } + + public void setId(UUID id) { + this.id = id; + } + + public OperationsRequest getOperationsRequest() { + return operationsRequest; + } + + public void setOperationsRequest(OperationsRequest operationsRequest) { + this.operationsRequest = operationsRequest; + } + + public Long getStatementId() { + return statementId; + } + + public void setStatementId(Long statementId) { + this.statementId = statementId; + } +} diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/GatewayService.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/GatewayService.java index 3366425fa..ba86cc481 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/GatewayService.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/GatewayService.java @@ -1,10 +1,12 @@ package ru.spcex.clearing.gatewayapi.service; +import org.apache.commons.lang3.exception.ExceptionUtils; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.http.*; import org.springframework.stereotype.Service; import org.springframework.web.client.HttpClientErrorException; @@ -12,15 +14,14 @@ import org.springframework.web.client.RestTemplate; import ru.spcex.clearing.gatewayapi.config.GatewayApiSettings; import ru.spcex.clearing.gatewayapi.config.InboundServerSettings; import ru.spcex.clearing.gatewayapi.controller.request.OutboundRequest; +import ru.spcex.clearing.gatewayapi.controller.request.operations.SentAsset; import ru.spcex.clearing.gatewayapi.enums.OutboundRequestType; import ru.spcex.clearing.gatewayapi.service.builders.OutboundRequestBuilder; import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import ru.spcex.clearing.platform.messaging.domain.Consts; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationListRequest; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationRequest; -import ru.spcex.clearing.platform.messaging.domain.cud.gateway.AssetOperationApprovalRequest; import ru.spcex.clearing.platform.messaging.domain.cud.gateway.GatewayTaskRequest; -import ru.spcex.clearing.platform.messaging.domain.cud.gateway.SingleAssetResponse; import ru.spcex.clearing.platform.messaging.domain.cud.utilities.LimExportedRequest; import ru.spcex.clearing.platform.messaging.service.QueueConsumer; import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate; @@ -38,19 +39,21 @@ public class GatewayService extends QueueConsumer implements InitializingBean { private final UserRoleVerification userRoleVerification; private final RestTemplate restTemplate; private final InboundServerSettings inboundServerSettings; - private final KafkaSender kafkaSender; + + private final Map> sentAssets; public GatewayService(Consumer kafkaQueue, Producer kafkaProducer, RestTemplate restTemplate, UserRoleVerification userRoleVerification, GatewayApiSettings gatewayApiSettings, - KafkaSender kafkaSender) { + KafkaSender kafkaSender, + @Qualifier("sentAssets") Map> sentAssets) { super(kafkaQueue, kafkaProducer); this.restTemplate = restTemplate; this.userRoleVerification = userRoleVerification; this.inboundServerSettings = gatewayApiSettings.getInboundServer(); - this.kafkaSender = kafkaSender; + this.sentAssets = sentAssets; } @Override @@ -145,9 +148,9 @@ public class GatewayService extends QueueConsumer implements InitializingBean { } public void requestOnAsset(BaseRequest request) { + Long requestId = request.getId(); Collection assetOperations = request.getRequestPayload().getAssetOperationRequests(); - List responses = new ArrayList<>(); - AssetOperationApprovalRequest assetOperationApprovalRequest = new AssetOperationApprovalRequest(); + List assets = new ArrayList<>(); for (AssetOperationRequest assetOperation : assetOperations) { Map content = Map.of("direction", assetOperation.getDirection(), "amount", assetOperation.getAmount(), @@ -166,19 +169,21 @@ public class GatewayService extends QueueConsumer implements InitializingBean { .content(content).build(); String url = formingInboundUrl(inboundServerSettings.getPathLOCM()); - HttpEntity r = makeDefaultRequest(outboundRequest); - ResponseEntity response = restTemplate.exchange(url, HttpMethod.POST, r, Map.class); - Map bodyResponse = response.getBody(); - SingleAssetResponse singleAssetResponse = new SingleAssetResponse(); - singleAssetResponse.setApproved(Boolean.parseBoolean((String) bodyResponse.get("result"))); - singleAssetResponse.setStatementId(assetOperation.getStatementId()); - responses.add(singleAssetResponse); - log.debug("Response from service: {}", bodyResponse); + try { + ResponseEntity response = restTemplate.exchange(url, HttpMethod.POST, r, Map.class); + Map bodyResponse = response.getBody(); + log.debug("Response from service: {}", bodyResponse != null ? bodyResponse : "empty"); + } catch (Throwable err){ + log.warn("Exchange error: {}", ExceptionUtils.getStackTrace(err)); + SentAsset sentAsset = new SentAsset(); + sentAsset.setId(outboundRequest.getId()); + sentAsset.setStatementId(assetOperation.getStatementId()); + assets.add(sentAsset); + } } - assetOperationApprovalRequest.setApprovals(responses); - kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION_APPROVAL, responses); + sentAssets.put(requestId, assets); } private String formingInboundUrl(String path) { diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/OperationService.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/OperationService.java new file mode 100644 index 000000000..5de2e4f3e --- /dev/null +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/OperationService.java @@ -0,0 +1,61 @@ +package ru.spcex.clearing.gatewayapi.service; + +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Service; +import ru.spcex.clearing.gatewayapi.controller.request.operations.OperationsRequest; +import ru.spcex.clearing.gatewayapi.controller.request.operations.SentAsset; +import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.gateway.AssetOperationApprovalRequest; +import ru.spcex.clearing.platform.messaging.domain.cud.gateway.SingleAssetResponse; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +@Service +public class OperationService { + private final Map> sentAssets; + private final KafkaSender kafkaSender; + + public OperationService(@Qualifier("sentAssets") Map> sentAssets, + KafkaSender kafkaSender) { + this.sentAssets = sentAssets; + this.kafkaSender = kafkaSender; + } + + public void checkAssetsAndSendMessage(OperationsRequest operationsRequest) { + Long requestId = null; + for (Map.Entry> entry : sentAssets.entrySet()) { + List value = entry.getValue(); + Optional assetById = value.stream() + .filter(sentAsset -> sentAsset.getId().equals(operationsRequest.getId())) + .findFirst(); + if (assetById.isPresent()) { + requestId = entry.getKey(); + SentAsset sentAsset = assetById.get(); + sentAsset.setReceivedResponse(true); + sentAsset.setOperationsRequest(operationsRequest); + break; + } + } + if (requestId != null) { + List assets = sentAssets.get(requestId); + boolean isAllReceivedInBatch = assets.stream().allMatch(SentAsset::isReceivedResponse); + if (isAllReceivedInBatch) { + List responsesToClearing = new ArrayList<>(); + for (SentAsset sentAsset : assets) { + SingleAssetResponse singleAssetResponse = new SingleAssetResponse(); + singleAssetResponse.setApproved(sentAsset.getOperationsRequest().getResult()); + singleAssetResponse.setStatementId(sentAsset.getStatementId()); + responsesToClearing.add(singleAssetResponse); + } + AssetOperationApprovalRequest assetOperationApprovalRequest = new AssetOperationApprovalRequest(); + assetOperationApprovalRequest.setApprovals(responsesToClearing); + kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION_APPROVAL, assetOperationApprovalRequest); + sentAssets.remove(requestId); + } + } + } +}