From 21976322c2c161ccd55384cb416ac98514b96ddc Mon Sep 17 00:00:00 2001 From: psemenkov Date: Fri, 26 Apr 2024 16:22:14 +0300 Subject: [PATCH] =?UTF-8?q?http://jira.mfd.msk:8088/browse/CLS-637=20?= =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D0=BB=20=D1=84=D0=BE=D1=80?= =?UTF-8?q?=D0=BC=D0=B8=D1=80=D0=BE=D0=B2=D0=B0=D0=BD=D0=B8=D0=B5=20=D1=81?= =?UTF-8?q?=D0=BE=D0=BE=D0=B1=D1=89=D0=B5=D0=BD=D0=B8=D1=8F=20=D0=BF=D1=80?= =?UTF-8?q?=D0=B8=20=D0=BE=D1=82=D0=BF=D1=80=D0=B0=D0=B2=D0=BA=D0=B5=20?= =?UTF-8?q?=D0=B7=D0=B0=D0=BF=D1=80=D0=BE=D1=81=D0=B0=20=D0=B2=20=D0=A2?= =?UTF-8?q?=D0=A1.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/GatewayController.java | 7 +- .../gatewayapi/service/GatewayService.java | 15 ++- .../service/NotificationService.java | 79 ++++++++++++++- .../builders/GatewayResultBuilder.java | 95 +++++++++++++++++++ .../service/GatewayServiceTest.java | 38 +++++++- 5 files changed, 225 insertions(+), 9 deletions(-) create mode 100644 clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/builders/GatewayResultBuilder.java 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 3ee258360..47bfb56d9 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 @@ -158,7 +158,7 @@ public class GatewayController { @ResponseBody public CommonResponse limits(@RequestBody LimitRequest request) { log.debug("Received message: {}", request); - notificationService.sendLimitNotification(request); + executor.submit(() -> notificationService.sendLimitNotification(request)); return createCommonResponse(request.getId()); } @@ -175,7 +175,10 @@ public class GatewayController { @ResponseBody public CommonResponse operations(@RequestBody OperationsRequest request) { log.debug("Received message: {}", request); - executor.submit(() -> operationService.checkAssetsAndSendMessage(request)); + executor.submit(() -> { + operationService.checkAssetsAndSendMessage(request); + notificationService.updateGatewayResultInMap(request); + }); return createCommonResponse(request.getId()); } 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 0b7f95d91..87389b7e5 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 @@ -10,13 +10,16 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.http.*; import org.springframework.stereotype.Service; import org.springframework.web.client.RestTemplate; +import ru.clearing.classes.statics.data.register.GatewayResult; import ru.spcex.clearing.gatewayapi.config.GatewayApiSettings; import ru.spcex.clearing.gatewayapi.config.InboundServerSettings; import ru.spcex.clearing.gatewayapi.controller.inbound.request.operations.SentAsset; import ru.spcex.clearing.gatewayapi.controller.outbound.request.OutboundRequest; import ru.spcex.clearing.gatewayapi.controller.outbound.response.SuccessResponse; import ru.spcex.clearing.gatewayapi.enums.OutboundRequestType; +import ru.spcex.clearing.gatewayapi.service.builders.GatewayResultBuilder; import ru.spcex.clearing.gatewayapi.service.builders.OutboundRequestBuilder; +import ru.spcex.clearing.imdg.IMDGDistributedNames; 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; @@ -32,6 +35,8 @@ import ru.spcex.platform.enumeration.ReportKeys; import ru.spcex.platform.enumeration.ReportType; import ru.spcex.platform.enumeration.Section; import ru.spcex.platform.enumeration.Task; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.utils.enumeration.IEnumKey; import java.math.BigDecimal; @@ -47,6 +52,7 @@ public class GatewayService extends QueueConsumer implements InitializingBean { private final InboundServerSettings inboundServerSettings; private final Map> sentAssets; private final Map outboundRequestByUuid; + protected final Imdg gatewayResultImdg; private final static Pattern REPORT_FILE_NAME_PATTERN = Pattern.compile("(\\S*)_(\\d{15}).csv"); public GatewayService(Consumer kafkaQueue, @@ -55,13 +61,15 @@ public class GatewayService extends QueueConsumer implements InitializingBean { UserRoleVerification userRoleVerification, GatewayApiSettings gatewayApiSettings, @Qualifier("sentAssets") Map> sentAssets, - Map outboundRequestByUuid) { + Map outboundRequestByUuid, + ImdgProvider imdgProvider) { super(kafkaQueue, kafkaProducer); this.restTemplate = restTemplate; this.userRoleVerification = userRoleVerification; this.inboundServerSettings = gatewayApiSettings.getInboundServer(); this.sentAssets = sentAssets; this.outboundRequestByUuid = outboundRequestByUuid; + this.gatewayResultImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_GatewayResult, GatewayResult.class); } @Override @@ -121,6 +129,7 @@ public class GatewayService extends QueueConsumer implements InitializingBean { HttpEntity request = makeDefaultRequest(outboundFondRequest); outboundRequestByUuid.put(outboundFondRequest.getId(), outboundFondRequest); + gatewayResultImdg.insert(GatewayResultBuilder.builder().outboundRequest(outboundFondRequest).build()); ResponseEntity response = restTemplate.exchange(url, HttpMethod.POST, request, SuccessResponse.class); SuccessResponse bodyResponse = response.getBody(); if (bodyResponse != null) { @@ -181,7 +190,9 @@ public class GatewayService extends QueueConsumer implements InitializingBean { Long requestId = request.getId(); Collection assetOperations = request.getRequestPayload().getAssetOperationRequests(); List assets = new ArrayList<>(); + GatewayResultBuilder gatewayBuilder = GatewayResultBuilder.builder(); for (AssetOperationRequest assetOperation : assetOperations) { + gatewayBuilder.assetOperation(assetOperation); Map content = new HashMap<>() {{ put("direction", assetOperation.getDirection()); put("code", assetOperation.getCode()); @@ -199,10 +210,12 @@ public class GatewayService extends QueueConsumer implements InitializingBean { .type(OutboundRequestType.ASSET_OPERATION.getKey()) .content(content).build(); + gatewayBuilder.outboundRequest(outboundRequest); String url = formingInboundUrl(inboundServerSettings.getPathLOCM()); HttpEntity r = makeDefaultRequest(outboundRequest); try { outboundRequestByUuid.put(outboundRequest.getId(), outboundRequest); + gatewayResultImdg.insert(gatewayBuilder.build()); ResponseEntity response = restTemplate.exchange(url, HttpMethod.POST, r, SuccessResponse.class); SuccessResponse bodyResponse = response.getBody(); if (bodyResponse != null) { diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/NotificationService.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/NotificationService.java index 1d1e38fa2..157e85c4b 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/NotificationService.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/NotificationService.java @@ -3,16 +3,24 @@ package ru.spcex.clearing.gatewayapi.service; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; +import ru.clearing.classes.statics.data.register.GatewayResult; import ru.spcex.clearing.gatewayapi.controller.inbound.request.limit.LimitRequest; +import ru.spcex.clearing.gatewayapi.controller.inbound.request.operations.OperationsRequest; import ru.spcex.clearing.gatewayapi.controller.outbound.request.OutboundRequest; import ru.spcex.clearing.gatewayapi.controller.outbound.response.ErrorResponse; import ru.spcex.clearing.gatewayapi.enums.OutboundRequestType; +import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.domain.Consts; import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationNewRequest; +import ru.spcex.clearing.platform.messaging.serialization.LogFormatter; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.platform.enumeration.ObjectType; import ru.spcex.platform.enumeration.Priority; +import ru.spcex.platform.enumeration.ResultStatuses; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; +import java.time.Instant; import java.util.List; import java.util.Map; import java.util.UUID; @@ -23,12 +31,15 @@ public class NotificationService { private final KafkaSender kafkaSender; private final Map outboundRequestByUuid; - private final List supportedTypes = List.of(OutboundRequestType.FILL_LIMITS.getKey()); + protected final Imdg gatewayResultImdg; + private final List supportedTypes = List.of(OutboundRequestType.FILL_LIMITS.getKey(), OutboundRequestType.ASSET_OPERATION.getKey()); public NotificationService(KafkaSender kafkaSender, - Map outboundRequestByUuid) { + Map outboundRequestByUuid, + ImdgProvider imdgProvider) { this.kafkaSender = kafkaSender; this.outboundRequestByUuid = outboundRequestByUuid; + this.gatewayResultImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_GatewayResult, GatewayResult.class); } public void sendLimitNotification(LimitRequest request) { @@ -40,6 +51,9 @@ public class NotificationService { } log.info("Receive request: uuid = {}, file = {} with result = {} and sending section = {}.", request.getId(), request.getLogFile(), request.getResult(), outboundReq.getSection()); + + updateGatewayResultInMap(request); + newRequest.setObjectType(ObjectType.rgst.getKey()); newRequest.setComment(request.getDescription()); if (request.getResult() != null && request.getResult()) { @@ -53,14 +67,69 @@ public class NotificationService { kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, newRequest); } + public void updateGatewayResultInMap(LimitRequest request) { + if (request.getId() == null || request.getResult() == null) { + log.warn("GatewayResult not found because LimitRequest not valid: {}", LogFormatter.toString(request)); + return; + } + GatewayResult gatewayResult = gatewayResultImdg.getFirstObjectByFieldValues(Map.of("requestId", request.getId().toString())); + if (gatewayResult == null) { + log.warn("GatewayResult by requestId = {} not found in map", request.getId().toString()); + return; + } + gatewayResult.setUpdated(Instant.now()); + gatewayResult.setGatewayResultStatus(request.getResult() ? ResultStatuses.success.getKey() : ResultStatuses.notSuccess.getKey()); + gatewayResult.setResult(LogFormatter.toString(request)); + gatewayResultImdg.update(gatewayResult); + log.info("Updated gatewayResult: {} in map", LogFormatter.toString(gatewayResult)); + } + + public void updateGatewayResultInMap(OperationsRequest request) { + if (request.getId() == null || request.getResult() == null) { + log.warn("GatewayResult not found because OperationsRequest not valid: {}", LogFormatter.toString(request)); + return; + } + GatewayResult gatewayResult = gatewayResultImdg.getFirstObjectByFieldValues(Map.of("requestId", request.getId().toString())); + if (gatewayResult == null) { + log.warn("GatewayResult by requestId = {} not found in map", request.getId().toString()); + return; + } + gatewayResult.setUpdated(Instant.now()); + gatewayResult.setGatewayResultStatus(request.getResult() ? ResultStatuses.success.getKey() : ResultStatuses.notSuccess.getKey()); + gatewayResult.setResult(LogFormatter.toString(request)); + gatewayResultImdg.update(gatewayResult); + log.info("Updated gatewayResult: {} in map", LogFormatter.toString(gatewayResult)); + } + + public void updateGatewayResultInMap(ErrorResponse response) { + if (response.getId() == null) { + log.warn("GatewayResult not found because OperationsRequest not valid: {}", LogFormatter.toString(response)); + return; + } + GatewayResult gatewayResult = gatewayResultImdg.getFirstObjectByFieldValues(Map.of("requestId", response.getId().toString())); + if (gatewayResult == null) { + log.warn("GatewayResult by requestId = {} not found in map", response.getId().toString()); + return; + } + gatewayResult.setUpdated(Instant.now()); + gatewayResult.setGatewayResultStatus(ResultStatuses.notSuccess.getKey()); + gatewayResult.setResult(LogFormatter.toString(response)); + gatewayResultImdg.update(gatewayResult); + log.info("Updated gatewayResult: {} in map", LogFormatter.toString(gatewayResult)); + } + public void sendErrorNotification(ErrorResponse response, OutboundRequest request) { NotificationNewRequest newRequest = new NotificationNewRequest(); - log.info("Receive ErrorResponse: uuid = {}, file = {} and sending section = {}.", - response.getId(), request.getContent().get("file"), request.getSection()); + updateGatewayResultInMap(response); + if (OutboundRequestType.FILL_LIMITS.equalsByKey(request.getType()) && request.getContent() != null && request.getContent().size() > 0) + log.info("Receive ErrorResponse: uuid = {}, file = {}, sending type = {} and section = {}.", + response.getId(), request.getContent().get("file"), request.getType(), request.getSection()); + else + log.info("Receive ErrorResponse: uuid = {}, sending type = {} and section = {}.", response.getId(), request.getType(), request.getSection()); newRequest.setObjectType(ObjectType.rgst.getKey()); newRequest.setComment(response.getMessage()); newRequest.setPriority(Priority.HIGH.getKey()); - log.info("Sending notification to kafka: {}", newRequest); + log.info("Sending notification to kafka: {}", LogFormatter.toString(newRequest)); kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, newRequest); } diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/builders/GatewayResultBuilder.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/builders/GatewayResultBuilder.java new file mode 100644 index 000000000..4f7a77464 --- /dev/null +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/builders/GatewayResultBuilder.java @@ -0,0 +1,95 @@ +package ru.spcex.clearing.gatewayapi.service.builders; + +import ru.clearing.classes.statics.data.register.GatewayResult; +import ru.spcex.clearing.gatewayapi.controller.outbound.request.OutboundRequest; +import ru.spcex.clearing.gatewayapi.enums.OutboundRequestType; +import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationRequest; +import ru.spcex.clearing.platform.messaging.serialization.LogFormatter; +import ru.spcex.platform.enumeration.InOutDirection; +import ru.spcex.platform.enumeration.Section; + +import java.time.Instant; + +import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey; + +public class GatewayResultBuilder { + + private String requestId; + private String nameRequest; + private String request; + private InOutDirection direction; + + private GatewayResultBuilder() { + } + + public static GatewayResultBuilder builder() { + return new GatewayResultBuilder(); + } + + public GatewayResultBuilder assetOperation(AssetOperationRequest assetOperation) { + this.direction = getEnumByKey(InOutDirection.class, assetOperation.getDirection()); + return this; + } + + public GatewayResultBuilder outboundRequest(OutboundRequest outboundRequest) { + this.request = LogFormatter.toString(outboundRequest); + this.requestId = outboundRequest.getId().toString(); + Section section = getEnumByKey(Section.class, outboundRequest.getSection()); + if (direction != null && OutboundRequestType.ASSET_OPERATION.equalsByKey(outboundRequest.getType())) + this.nameRequest = getNameReqByAsset(section, direction); + else if (OutboundRequestType.FILL_LIMITS.equalsByKey(outboundRequest.getType())) + this.nameRequest = getNameReqByLim(section); + return this; + } + + public GatewayResult build() { + GatewayResult outboundRequest = new GatewayResult(); + Instant now = Instant.now(); + outboundRequest.setCreated(now); + outboundRequest.setUpdated(now); + outboundRequest.setRequestId(requestId); + outboundRequest.setNameRequest(nameRequest); + outboundRequest.setRequest(request); + return outboundRequest; + } + + private String getNameReqByAsset(Section section, InOutDirection direction) { + String nameRequest; + if (InOutDirection.in.equals(direction)) { + if (Section.MKR.equals(section)) { + nameRequest = "Зачисление денежных средств"; + } else if (Section.FOND.equals(section)) { + nameRequest = "Зачисление ценных бумаг"; + } else if (Section.CURR.equals(section)) { + nameRequest = "Зачисление валютных средств"; + } else { + nameRequest = "Зачисление " + section.getKey(); + } + } else { + if (Section.MKR.equals(section)) { + nameRequest = "Списание денежных средств"; + } else if (Section.FOND.equals(section)) { + nameRequest = "Списание ценных бумаг"; + } else if (Section.CURR.equals(section)) { + nameRequest = "Списание валютных средств"; + } else { + nameRequest = "Списание " + section.getKey(); + } + } + return nameRequest; + } + + private String getNameReqByLim(Section section) { + String nameRequest; + if (Section.MKR.equals(section)) { + nameRequest = "Лимиты по денежным средствам"; + } else if (Section.FOND.equals(section)) { + nameRequest = "Лимиты по ценным бумагам"; + } else if (Section.CURR.equals(section)) { + nameRequest = "Лимиты по валютным средствам"; + } else { + nameRequest = "Лимиты по " + section.getKey(); + } + return nameRequest; + } +} diff --git a/clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/service/GatewayServiceTest.java b/clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/service/GatewayServiceTest.java index 5cb275a18..70390545d 100644 --- a/clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/service/GatewayServiceTest.java +++ b/clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/service/GatewayServiceTest.java @@ -32,18 +32,23 @@ import ru.spcex.clearing.gatewayapi.controller.outbound.request.OutboundRequest; import ru.spcex.clearing.gatewayapi.controller.outbound.response.ErrorResponse; import ru.spcex.clearing.gatewayapi.controller.outbound.response.SuccessResponse; import ru.spcex.clearing.platform.messaging.domain.BaseRequest; +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.utilities.LimExportedRequest; import ru.spcex.clearing.test.config.ImdgTestConfig; import ru.spcex.clearing.test.config.KafkaTestConfig; +import ru.spcex.platform.enumeration.InOutDirection; import ru.spcex.platform.enumeration.Section; import ru.spcex.platform.imdg.api.ImdgProvider; import javax.annotation.PostConstruct; import java.io.ByteArrayOutputStream; import java.io.File; +import java.math.BigDecimal; import java.nio.charset.Charset; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.List; import java.util.UUID; import java.util.concurrent.atomic.AtomicReference; @@ -61,6 +66,7 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers. public class GatewayServiceTest { private static final CharacterEncodingFilter CHARACTER_ENCODING_FILTER = new CharacterEncodingFilter(); public static final String LIMITS_URL = "/limits/"; + public static final String OPERATIONS_URL = "/operations/"; private MockRestServiceServer mockServer; @Autowired @@ -137,6 +143,36 @@ public class GatewayServiceTest { service.requestOnLimit(req); } + @Disabled//Для отладки + @Test + void requestOnAsset() throws Exception { + String message = "test response"; + BaseRequest req = new BaseRequest<>(); + AssetOperationListRequest reqP = new AssetOperationListRequest(); + AssetOperationRequest reqI = new AssetOperationRequest(); + req.setId(1L); + reqP.setAssetOperationRequests(List.of(reqI)); + req.setRequestPayload(reqP); + reqI.setAmount(BigDecimal.ONE); + reqI.setDirection(InOutDirection.in.getKey()); + AtomicReference uuid = prepareMockServerResponse(HttpStatus.OK, message); + service.requestOnAsset(req); + LimitRequest request = new LimitRequest(); + request.setId(uuid.get()); + request.setResult(true); + request.setDescription(message); + MvcResult mvcResult = perform(MockMvcRequestBuilders.post(OPERATIONS_URL) + .contentType(MediaType.APPLICATION_JSON) + .content(mapper.writeValueAsString(request))) + .andDo(print())//output to the log request and response + .andExpect(status().isOk()) + .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON)) + .andReturn(); + + uuid = prepareMockServerResponse(HttpStatus.BAD_REQUEST, message); + service.requestOnAsset(req); + } + public AtomicReference prepareMockServerResponse(HttpStatus responseStatus, String message) { mockServer.reset(); @@ -145,7 +181,7 @@ public class GatewayServiceTest { ResponseCreator responseCreator = r -> { String reqBody = StreamUtils.copyToString((ByteArrayOutputStream) r.getBody(), Charset.defaultCharset()); OutboundRequest request = mapper.readValue(reqBody, OutboundRequest.class); - if (HttpStatus.BAD_REQUEST.equals(responseStatus)){ + if (HttpStatus.BAD_REQUEST.equals(responseStatus)) { ErrorResponse errorResponse = new ErrorResponse(); errorResponse.setId(request.getId()); errorResponse.setMessage(message);