From fd900e27b6872eb1aaef19d953afb3c0da0ebb3f Mon Sep 17 00:00:00 2001 From: ialbert Date: Wed, 27 Dec 2023 11:32:10 +0300 Subject: [PATCH] http://jira.mfd.msk:8088/browse/CLS-606 --- .../clearing/service/EventsReceiver.java | 9 - .../service/executors/Sdf06Executor.java | 274 +++++++++++------- .../service/executors/Sdf10Executor.java | 252 +++++++++------- .../stage/impl/GatewayBatchRequester.java | 146 ++++++++++ .../gatewayapi/service/OperationService.java | 1 + .../AssetOperationApprovalRequest.java | 10 + 6 files changed, 471 insertions(+), 221 deletions(-) create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/GatewayBatchRequester.java diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java index 4527e3f19..0414ed83b 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java @@ -15,7 +15,6 @@ import ru.spcex.clearing.platform.messaging.domain.cud.clearing.CreateRegistryRe import ru.spcex.clearing.platform.messaging.domain.cud.clearing.Sdf04Request; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.SessionContinueEvent; import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonIdRequest; -import ru.spcex.clearing.platform.messaging.domain.cud.gateway.AssetOperationApprovalRequest; import ru.spcex.clearing.platform.messaging.domain.cud.payment.PIClearingOutbondActionNewRequest; import ru.spcex.clearing.platform.messaging.domain.cud.registry.*; import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; @@ -138,14 +137,6 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean { callback(StatementRequest.class) .setConsumer(sdf06Executor::execute) .forDestination(Consts.STATEMENT_PROCESS_SDF06, callbacks::put); - - callback(AssetOperationApprovalRequest.class) - .setConsumer(req -> { - sdf06Executor.processGatewayResponse(req); - sdf10Executor.processGatewayResponse(req); - }) - .forDestination(Consts.ASSET_OPERATION_APPROVAL, callbacks::put); - callback(STradesImportedRequest.class) .setConsumer(event -> clearingService.executeSTrade()) .forDestination(S_TRADES_IMPORTED, callbacks::put); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java index 9d1ba048a..6bc5afbef 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java @@ -17,9 +17,7 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import ru.spcex.clearing.platform.messaging.domain.Consts; import ru.spcex.clearing.platform.messaging.domain.cud.balance.ExportToFileRequest; import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest; -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.SingleAssetResponse; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.clearing.service.AssetTrio; @@ -29,16 +27,14 @@ import ru.spcex.clearing.service.registry.DmiService; import ru.spcex.clearing.service.schedule.TradingTimeService; import ru.spcex.clearing.service.validation.Sdf06NewValidationRule; import ru.spcex.clearing.service.validation.ValidationStored; -import ru.spcex.platform.classes.base.SpcexObjectBase; +import ru.spcex.clearing.session.stage.impl.GatewayBatchRequester; import ru.spcex.platform.enumeration.*; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgProvider; -import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; import ru.spcex.platform.imdg.validation.ImdgValidationContext; import ru.spcex.platform.utils.enumeration.EnumMessage; import ru.spcex.platform.utils.enumeration.IMessageResolver; -import ru.spcex.platform.utils.text.TextUtil; import ru.spcex.platform.utils.validation.IValidator; import ru.spcex.platform.utils.validation.ValidatorImpl; @@ -78,19 +74,19 @@ public class Sdf06Executor { //отказ от gateway либо недостаточно средств у нас private final static BigDecimal errorResult3 = new BigDecimal("3"); private final static BigDecimal errorResult4 = new BigDecimal("4"); + private final GatewayBatchRequester gateway; //not thread safe: used in EventReceiver single thread executor //переменная контроля за тем что от гейтвея пришел ответ именно на наш запрос LinkedList> cachedRequests = new LinkedList<>(); - private Long sdf06GroupId; - private Long sdf07GroupId; @Autowired public Sdf06Executor(ImdgProvider imdgProvider, IMessageResolver messageResolver, @Qualifier("sdf06ValidatorNew") Function sDf06Validator, - TradingTimeService tradingTimeService, KafkaSender kafkaSender, FilenameObtainer filenameObtainer, DmiService dmiService, AssetTBFProcessing assets) { + TradingTimeService tradingTimeService, KafkaSender kafkaSender, FilenameObtainer filenameObtainer, DmiService dmiService, AssetTBFProcessing assets, + GatewayBatchRequester gateway) { this.imdgProvider = imdgProvider; this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class); this.tradingTimeService = tradingTimeService; @@ -106,6 +102,8 @@ public class Sdf06Executor { this.filenameObtainer = filenameObtainer; this.dmiService = dmiService; this.assets = assets; + this.gateway = gateway; + this.gateway.setName("Sdf06"); } public void execute(BaseRequest systemRequest) { @@ -118,12 +116,13 @@ public class Sdf06Executor { log.info("start processing SDF06: groupId={} file={}", groupId, fileName); Instant now = Instant.now(); Long sdf07GroupId = idGenerator.nextId(); - if (sdf06GroupId != null) { - log.warn("currently awaiting gateway response for groupId: {}. adding to cache groupId {}", sdf06GroupId, groupId); - cachedRequests.add(systemRequest); - return; - } + //if (sdf06GroupId != null) { + // log.warn("currently awaiting gateway response for groupId: {}. adding to cache groupId {}", sdf06GroupId, groupId); + // cachedRequests.add(systemRequest); + // return; + //} Collection requests = new ArrayList<>(); + Collection records = new ArrayList<>(); boolean sdf07WasCreated = false; for (SDf06 sDf06 : sdfs) { IValidator validator = sDf06Validator.apply(sDf06); @@ -157,6 +156,7 @@ public class Sdf06Executor { log.debug("Statement created: {}", stmt.getId()); if (InOutDirection.out.getKey().equals(stmt.getInOutDirection()) && tradingTimeService.isTradingTime()) { requests.add(GatewayRequestCreator.from(stmt, company.getTradingCode(), tcr.getCode())); + records.add(new Trio(sDf06, stmt, tcr)); } else { processedApproved(stmt, sDf06, now, sdf07GroupId); if (tcr != null) { @@ -171,19 +171,80 @@ public class Sdf06Executor { } } if (requests.size() > 0) { - this.sdf06GroupId = groupId; - this.sdf07GroupId = sdf07GroupId; - AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest(); - assetOperationListRequest.setAssetOperationRequests(requests); - kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest); - } else if (sdf07WasCreated) { + Instant updatedTime = Instant.now(); + Optional> gatewayResponse = gateway.gatewayRequestAndWaitBatch(requests); + if (gatewayResponse.isEmpty()) { + log.error("couldn't retrieve SDF06 gateway response for groupId={}/sdf07.groupId={}.", groupId, sdf07GroupId); + sdf07WasCreated = true; + records.forEach(trio -> { + SDf06 sDf06 = trio.sdf06(); + Statement stmt = trio.stmt(); + SDf07 sdf07 = createSdf07(sDf06, now, errorResult3); + sdf07.setGenerationId(sdf07GroupId); + sdf07Imdg.insert(sdf07); + stmt.setOperationStatus(OperationStatus.Rejected.getKey()); + stmt.setUpdated(updatedTime); + stmt.setOutSDfId(sdf07.getId()); + statementImdg.update(stmt); + }); + } else { + List gtwRsp = gatewayResponse.get(); + for (SingleAssetResponse gatewayMsg : gtwRsp) { + //получаем запрос для текущей группы sdf06 + //находим группу + Long statementId = gatewayMsg.getEntityId(); + Trio trio = records.stream() + .filter(s -> Objects.equals(s.stmt().getId(), statementId)) + .findFirst() + .orElse(null); + if (trio == null) { + log.error("SDF06.groupId={}, SDF07.groupId={} gateway response statement.id {} with type {} not found", + groupId, sdf07GroupId, statementId, InOutSDfType.type6.getKey()); + continue; + } + SDf06 sdf06 = trio.sdf06(); + Statement stmt = trio.stmt(); + TradingClearingRegistry tcr = trio.tcr(); + if (gatewayMsg.isApproved()) { + processedApproved(stmt, sdf06, updatedTime, sdf07GroupId); + dmiService.setProcContract(tcr != null ? tcr.getId() : null, + CurrencyCode.RUB.getKey(), + safeBD(sdf06.getSum()), + sdf06.getNumber().toString() + ); + if (tcr != null) { + Optional asts = assets.searchMoneyByAccAndCompany(tcr.getCompanyId(), tcr.getMoneyAccountId()); + asts.ifPresent(a -> assets.process(a.a__b(), a.a__t(), a.a__f(), BigDecimal.ZERO)); + } + } else { + log.trace("statement.id={}, sdf07.id={}, sdf06.id={} rejected (by gateway answer)", + statementId, + sdf06.getGenerationId(), + sdf06.getId()); + SDf07 sdf07 = createSdf07(sdf06, now, errorResult3); + sdf07.setGenerationId(sdf07GroupId); + sdf07Imdg.insert(sdf07); + stmt.setOperationStatus(OperationStatus.Rejected.getKey()); + stmt.setUpdated(updatedTime); + stmt.setOutSDfId(sdf07.getId()); + statementImdg.update(stmt); + } + sdf07WasCreated = true; + } + } + } + if (sdf07WasCreated) { sendToExporter(sdf07GroupId, fileName); } + if (cachedRequests.size() > 0) { + BaseRequest statementRequest = cachedRequests.removeFirst(); + execute(statementRequest); + } } - private void processedApproved(Statement stmt, SDf06 sDf06, Instant time) { - processedApproved(stmt, sDf06, time, sdf07GroupId); - } + //private void processedApproved(Statement stmt, SDf06 sDf06, Instant time) { + // processedApproved(stmt, sDf06, time, sdf07GroupId); + //} private void processedApproved(Statement stmt, SDf06 sDf06, Instant time, Long sdf07GenerationId) { log.trace("statement.id={}, sdf07.id={}, sdf06.id={} executed", @@ -227,92 +288,87 @@ public class Sdf06Executor { return result; } - public void processGatewayResponse(BaseRequest req) { - //может прийти неограниченно позже 06го, после того как прошел клиринг например... - Instant now = Instant.now(); - String fileName = null; - boolean requestIsIntendedForSdf06 = false; - for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) { - //получаем запрос для текущей группы sdf06 - //находим группу - Long statementId = gatewayMsg.getEntityId(); - ImdgPredicateBuilder pb = statementImdg.predicateBuilder(); - Statement stmt = statementImdg.getFirstObjectByPredicate( - pb.and( - pb.equals("id", statementId), - pb.equals("inOutSDfType", InOutSDfType.type6.getKey()) - ) - ); - if (stmt == null) { - log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type6.getKey()); - return; - } - requestIsIntendedForSdf06 = true; - - Long sdf06Id = stmt.getInSDfId(); - SDf06 sdf06 = sdf06Imdg.getSingleObjectByID(sdf06Id); - if (fileName == null && !TextUtil.isEmpty(sdf06.getFileName())) { - fileName = sdf06.getFileName(); - } - if (sdf06 == null) { - log.error("Sdf06.id {} not found by statement.id {}", sdf06Id, statementId); - return; - } - - Long groupId = sdf06.getGenerationId(); - //сверяем группу SDF06 пришедшего запроса с ожидаемой - if (sdf06GroupId == null || !sdf06GroupId.equals(groupId)) { - log.error("do not currently waiting for gateway response for statement.id {} sdf06 groupId {}; waiting for {}", - statementId, - groupId, - sdf06Id); - return; - } - - Instant updatedTime = Instant.now(); - if (gatewayMsg.isApproved()) { - processedApproved(stmt, sdf06, updatedTime); - Optional tcr = searchTcrOnGatewayResponse(sdf06); - dmiService.setProcContract(tcr.map(SpcexObjectBase::getId).orElse(null), - CurrencyCode.RUB.getKey(), - safeBD(sdf06.getSum()), - sdf06.getNumber().toString() - ); - if (tcr.isPresent()) { - Optional asts = assets.searchMoneyByAccAndCompany(tcr.get().getCompanyId(), tcr.get().getMoneyAccountId()); - asts.ifPresent(a -> assets.process(a.a__b(), a.a__t(), a.a__f(), BigDecimal.ZERO)); - } - } else { - log.trace("statement.id={}, sdf07.id={}, sdf06.id={} rejected (by gateway answer)", - statementId, - sdf06.getGenerationId(), - sdf06.getId()); - SDf07 sdf07 = createSdf07(sdf06, now, errorResult3); - sdf07.setGenerationId(sdf07GroupId); - sdf07Imdg.insert(sdf07); - stmt.setOperationStatus(OperationStatus.Rejected.getKey()); - stmt.setUpdated(updatedTime); - stmt.setOutSDfId(sdf07.getId()); - statementImdg.update(stmt); - } - } - if (requestIsIntendedForSdf06) { - sendToExporter(sdf07GroupId, fileName); - log.debug("All gateway responses received for SDF06 groupId {}. SDF07 groupId {}", sdf06GroupId, sdf07GroupId); - clearContext(); - } else { - log.debug("gateway response was not for SDF06 executor"); - } - if (cachedRequests.size() > 0) { - BaseRequest statementRequest = cachedRequests.removeFirst(); - execute(statementRequest); - } - } - - private void clearContext() { - this.sdf06GroupId = null; - this.sdf07GroupId = null; - } +// public void processGatewayResponse(BaseRequest req) { +// //может прийти неограниченно позже 06го, после того как прошел клиринг например... +// Instant now = Instant.now(); +// String fileName = null; +// boolean requestIsIntendedForSdf06 = false; +// for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) { +// //получаем запрос для текущей группы sdf06 +// //находим группу +// Long statementId = gatewayMsg.getEntityId(); +// ImdgPredicateBuilder pb = statementImdg.predicateBuilder(); +// Statement stmt = statementImdg.getFirstObjectByPredicate( +// pb.and( +// pb.equals("id", statementId), +// pb.equals("inOutSDfType", InOutSDfType.type6.getKey()) +// ) +// ); +// if (stmt == null) { +// log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type6.getKey()); +// return; +// } +// requestIsIntendedForSdf06 = true; +// +// Long sdf06Id = stmt.getInSDfId(); +// SDf06 sdf06 = sdf06Imdg.getSingleObjectByID(sdf06Id); +// if (fileName == null && !TextUtil.isEmpty(sdf06.getFileName())) { +// fileName = sdf06.getFileName(); +// } +// if (sdf06 == null) { +// log.error("Sdf06.id {} not found by statement.id {}", sdf06Id, statementId); +// return; +// } +// +// Long groupId = sdf06.getGenerationId(); +// //сверяем группу SDF06 пришедшего запроса с ожидаемой +// if (sdf06GroupId == null || !sdf06GroupId.equals(groupId)) { +// log.error("do not currently waiting for gateway response for statement.id {} sdf06 groupId {}; waiting for {}", +// statementId, +// groupId, +// sdf06Id); +// return; +// } +// +// Instant updatedTime = Instant.now(); +// if (gatewayMsg.isApproved()) { +// processedApproved(stmt, sdf06, updatedTime); +// Optional tcr = searchTcrOnGatewayResponse(sdf06); +// dmiService.setProcContract(tcr.map(SpcexObjectBase::getId).orElse(null), +// CurrencyCode.RUB.getKey(), +// safeBD(sdf06.getSum()), +// sdf06.getNumber().toString() +// ); +// if (tcr.isPresent()) { +// Optional asts = assets.searchMoneyByAccAndCompany(tcr.get().getCompanyId(), tcr.get().getMoneyAccountId()); +// asts.ifPresent(a -> assets.process(a.a__b(), a.a__t(), a.a__f(), BigDecimal.ZERO)); +// } +// } else { +// log.trace("statement.id={}, sdf07.id={}, sdf06.id={} rejected (by gateway answer)", +// statementId, +// sdf06.getGenerationId(), +// sdf06.getId()); +// SDf07 sdf07 = createSdf07(sdf06, now, errorResult3); +// sdf07.setGenerationId(sdf07GroupId); +// sdf07Imdg.insert(sdf07); +// stmt.setOperationStatus(OperationStatus.Rejected.getKey()); +// stmt.setUpdated(updatedTime); +// stmt.setOutSDfId(sdf07.getId()); +// statementImdg.update(stmt); +// } +// } +// if (requestIsIntendedForSdf06) { +// sendToExporter(sdf07GroupId, fileName); +// log.debug("All gateway responses received for SDF06 groupId {}. SDF07 groupId {}", sdf06GroupId, sdf07GroupId); +// clearContext(); +// } else { +// log.debug("gateway response was not for SDF06 executor"); +// } +// if (cachedRequests.size() > 0) { +// BaseRequest statementRequest = cachedRequests.removeFirst(); +// execute(statementRequest); +// } +// } private void sendToExporter(Long generationId, String fileName) { ExportToFileRequest exportRequest = new ExportToFileRequest(); @@ -389,4 +445,6 @@ public class Sdf06Executor { } return Optional.empty(); } + + private static record Trio (SDf06 sdf06, Statement stmt, TradingClearingRegistry tcr) {}; } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf10Executor.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf10Executor.java index a3d24aaea..ff005ad98 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf10Executor.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf10Executor.java @@ -19,9 +19,7 @@ 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.balance.StatementRequest; -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.SingleAssetResponse; import ru.spcex.clearing.platform.messaging.domain.cud.importexport.SwtExporterRequest; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; @@ -29,12 +27,12 @@ import ru.spcex.clearing.service.registry.AssetTBFProcessing; import ru.spcex.clearing.service.registry.RegistryManager; import ru.spcex.clearing.service.schedule.TradingTimeService; import ru.spcex.clearing.service.validation.ValidationStored; +import ru.spcex.clearing.session.stage.impl.GatewayBatchRequester; import ru.spcex.clearing.session.stage.util.RegistryUtil; import ru.spcex.platform.enumeration.*; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgProvider; -import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; import ru.spcex.platform.imdg.api.predicate.specific.SecuritySelector; import ru.spcex.platform.utils.enumeration.EnumMessage; import ru.spcex.platform.utils.enumeration.IEnumId; @@ -44,10 +42,7 @@ import ru.spcex.platform.utils.validation.IValidator; import java.math.BigDecimal; import java.time.Instant; import java.time.LocalDate; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Map; -import java.util.Optional; +import java.util.*; import java.util.function.Function; import static ru.spcex.platform.utils.number.BigDecimalUtil.safeBD; @@ -70,10 +65,12 @@ public class Sdf10Executor { private final KafkaSender kafkaSender; private final SecuritySelector scrtSlct; private final AssetTBFProcessing assets; + private final GatewayBatchRequester gateway; private final static String OK = "OK"; private final static String SYNTAX_ERROR = "Синтаксическая ошибка (файл сформирован неверно)"; private final static String GATEWAY_REJECTED = "Отрицательный ответ от Торговой Системы"; + private final static String GATEWAY_NO_ANSWER = "Не удалось получить ответ от Торговой Системы"; private final static String NOT_FOUND_ACCOUNT = "Не найден код раздела субсчета/счета депо"; private final static String NOT_FOUND_COMPANY = "Не найдена компания"; private final static String NOT_FOUND_TCR = "Не найден ТКР"; @@ -83,14 +80,11 @@ public class Sdf10Executor { private final static String SESSION_IS_ACTIVE = "Существует активная сессия"; private final static String GENERAL_ERROR = "Общая ошибка обработки SDF10"; - private Long sdf10GroupId; - private Long sdf11GroupId; - @Autowired public Sdf10Executor(ImdgProvider imdgProvider, RegistryManager rgsMng, IMessageResolver messageResolver, @Qualifier("sdf10Validator") Function sDf10Validator, - TradingTimeService tradingTimeService, KafkaSender kafkaSender, AssetTBFProcessing assets) { + TradingTimeService tradingTimeService, KafkaSender kafkaSender, AssetTBFProcessing assets, GatewayBatchRequester gateway) { this.imdgProvider = imdgProvider; this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class); this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); @@ -105,6 +99,8 @@ public class Sdf10Executor { this.plannerAllTodayImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_PlannerAllToday, PlannerAllToday.class); this.scrtSlct = new SecuritySelector<>(imdgProvider, Security.class); this.assets = assets; + this.gateway = gateway; + this.gateway.setName("sdf10"); } public void execute(BaseRequest systemRequest) { @@ -116,16 +112,17 @@ public class Sdf10Executor { Instant now = Instant.now(); Long sdf11GroupId = idGenerator.nextId(); log.info("processing SDF10 groupId: {} sdf10s: {}", groupId, sdfs.size()); - if (sdf10GroupId != null) { - log.warn("currently awaiting gateway response for groupId: {}. skipping groupId {}", sdf10GroupId, groupId); - sdfs.forEach(sdf10 -> { - SDf11 errorSdf11 = createSdf11(sdf10, sdf11GroupId, now, PREVIOUS_REQUEST_NOT_PROCESSED); - this.sdf11Imdg.insert(errorSdf11); - }); - sendToExporter(sdf11GroupId); - return; - } + //if (sdf10GroupId != null) { + // log.warn("currently awaiting gateway response for groupId: {}. skipping groupId {}", sdf10GroupId, groupId); + // sdfs.forEach(sdf10 -> { + // SDf11 errorSdf11 = createSdf11(sdf10, sdf11GroupId, now, PREVIOUS_REQUEST_NOT_PROCESSED); + // this.sdf11Imdg.insert(errorSdf11); + // }); + // sendToExporter(sdf11GroupId); + // return; + //} Collection requests = new ArrayList<>(); + List records = new ArrayList<>(); boolean sdf11WasCreated = false; for (SDf10 sDf10 : sdfs) { IValidator validator = sDf10Validator.apply(sDf10); @@ -165,6 +162,7 @@ public class Sdf10Executor { log.debug("Statement created: {}", stmt.getId()); if (InOutDirection.out.getKey().equals(stmt.getInOutDirection()) && tradingTimeService.isTradingTime()) { requests.add(requestFromStatement(stmt, company.getTradingCode(), tcr.getCode(), sDf10.getSecurityCode())); + records.add(new Trio(sDf10, stmt, security)); } else { SDf11 sDf11 = processedApproved(stmt, sDf10, now, sdf11GroupId); BigDecimal amount = InOutDirection.in.equalsByKey(stmt.getInOutDirection()) ? @@ -178,12 +176,65 @@ public class Sdf10Executor { } } if (requests.size() > 0) { - this.sdf10GroupId = groupId; - this.sdf11GroupId = sdf11GroupId; - AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest(); - assetOperationListRequest.setAssetOperationRequests(requests); - kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest); - } else if (sdf11WasCreated) { + Instant updatedTime = Instant.now(); + Optional> gatewayResponse = gateway.gatewayRequestAndWaitBatch(requests); + if (gatewayResponse.isEmpty()) { + log.error("couldn't retrieve SDF10 gateway response for groupId={}/sdf11.groupId={}", groupId, sdf11GroupId); + sdf11WasCreated = true; + records.forEach(trio -> { + SDf10 sDf10 = trio.sDf10(); + Statement stmt = trio.stmt(); + SDf11 errorSdf11 = createSdf11(sDf10, sdf11GroupId, now, GATEWAY_NO_ANSWER); + errorSdf11.setGenerationId(sdf11GroupId); + sdf11Imdg.insert(errorSdf11); + stmt.setOperationStatus(OperationStatus.Rejected.getKey()); + stmt.setUpdated(updatedTime); + stmt.setOutSDfId(errorSdf11.getId()); + statementImdg.update(stmt); + }); + } else { + List gtwRsp = gatewayResponse.get(); + Instant gtwReceived = Instant.now(); + for (SingleAssetResponse gatewayMsg : gtwRsp) { + Long statementId = gatewayMsg.getEntityId(); + Trio trio = records.stream() + .filter(r -> Objects.equals(r.stmt().getId(), statementId)) + .findFirst() + .orElse(null); + if (trio == null) { + log.error("SDF10.groupId={}, SDF07.groupId={}" + + " gateway response statement.id {} with type {} not found", + groupId, sdf11GroupId, statementId, InOutSDfType.type10.getKey()); + continue; + } + SDf10 sdf10 = trio.sDf10(); + Statement stmt = trio.stmt(); + Security security = trio.security(); + if (gatewayMsg.isApproved()) { + createDs_iResponseFromGateway( + stmt.getAccountId(), + security != null ? security.getSecuritySymbol() : null, + stmt.getAmount().negate(), + sdf10.getOutDocument()); + //tcrId, companyId, securitySymbol + processedApproved(stmt, sdf10, gtwReceived, sdf11GroupId); + } else { + log.trace("statement.id={}, sdf11.id={}, sdf10.id={} rejected (by gateway answer)", + statementId, + sdf10.getGenerationId(), + sdf10.getId()); + SDf11 sdf11 = createSdf11(sdf10, sdf11GroupId, now, GATEWAY_REJECTED); + sdf11Imdg.insert(sdf11); + stmt.setOperationStatus(OperationStatus.Rejected.getKey()); + stmt.setUpdated(gtwReceived); + stmt.setOutSDfId(sdf11.getId()); + statementImdg.update(stmt); + } + sdf11WasCreated = true; + } + } + } + if (sdf11WasCreated) { sendToExporter(sdf11GroupId); } } @@ -229,10 +280,6 @@ public class Sdf10Executor { log.debug("created DS*I.id={}", rgsD.getId()); } - private void processedApproved(Statement stmt, SDf10 sDf10, Instant time) { - processedApproved(stmt, sDf10, time, sdf11GroupId); - } - private SDf11 processedApproved(Statement stmt, SDf10 sDf10, Instant time, Long sdf11GenerationId) { log.trace("statement.id={}, sdf11.id={}, sdf10.id={} executed", stmt.getId(), @@ -247,80 +294,75 @@ public class Sdf10Executor { return sdf11; } - public void processGatewayResponse(BaseRequest req) { - Instant now = Instant.now(); - boolean requestIsIntendedForSdf10 = false; - for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) { - //получаем запрос для текущей группы sdf10 - //находим группу - Long statementId = gatewayMsg.getEntityId(); - ImdgPredicateBuilder pb = statementImdg.predicateBuilder(); - Statement stmt = statementImdg.getFirstObjectByPredicate( - pb.and( - pb.equals("id", statementId), - pb.equals("inOutSDfType", InOutSDfType.type10.getKey()) - ) - ); - if (stmt == null) { - log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type10.getKey()); - return; - } - requestIsIntendedForSdf10 = true; - - Long sdf10Id = stmt.getInSDfId(); - SDf10 sdf10 = sdf10Imdg.getSingleObjectByID(sdf10Id); - if (sdf10 == null) { - log.error("Sdf10.id {} not found by statement.id {}", sdf10Id, statementId); - return; - } - - Long groupId = sdf10.getGenerationId(); - //сверяем группу SDF10 пришедшего запроса с ожидаемой - if (sdf10GroupId == null || !sdf10GroupId.equals(groupId)) { - log.error("do not currently waiting for gateway response for statement.id {} sdf10 groupId {}; waiting for {}", - statementId, - groupId, - sdf10Id); - return; - } - - Instant updatedTime = Instant.now(); - if (gatewayMsg.isApproved()) { - Security security = scrtSlct.selectSecurityById(stmt.getSecurityId()); - createDs_iResponseFromGateway( - stmt.getAccountId(), -// stmt.getAddresseeId(), - security != null ? security.getSecuritySymbol() : null, - stmt.getAmount().negate(), - sdf10.getOutDocument()); - //tcrId, companyId, securitySymbol - processedApproved(stmt, sdf10, updatedTime); - } else { - log.trace("statement.id={}, sdf11.id={}, sdf10.id={} rejected (by gateway answer)", - statementId, - sdf10.getGenerationId(), - sdf10.getId()); - SDf11 sdf11 = createSdf11(sdf10, sdf11GroupId, now, GATEWAY_REJECTED); - sdf11Imdg.insert(sdf11); - stmt.setOperationStatus(OperationStatus.Rejected.getKey()); - stmt.setUpdated(updatedTime); - stmt.setOutSDfId(sdf11.getId()); - statementImdg.update(stmt); - } - } - if (requestIsIntendedForSdf10) { - sendToExporter(sdf11GroupId); - log.debug("All gateway responses received for SDF10 groupId {}. SDF11 groupId {}", sdf10GroupId, sdf11GroupId); - clearContext(); - } else { - log.debug("gateway response was not for SDF10 executor"); - } - } - - private void clearContext() { - this.sdf10GroupId = null; - this.sdf11GroupId = null; - } +// public void processGatewayResponse(BaseRequest req) { +// Instant now = Instant.now(); +// boolean requestIsIntendedForSdf10 = false; +// for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) { +// //получаем запрос для текущей группы sdf10 +// //находим группу +// Long statementId = gatewayMsg.getEntityId(); +// ImdgPredicateBuilder pb = statementImdg.predicateBuilder(); +// Statement stmt = statementImdg.getFirstObjectByPredicate( +// pb.and( +// pb.equals("id", statementId), +// pb.equals("inOutSDfType", InOutSDfType.type10.getKey()) +// ) +// ); +// if (stmt == null) { +// log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type10.getKey()); +// return; +// } +// requestIsIntendedForSdf10 = true; +// +// Long sdf10Id = stmt.getInSDfId(); +// SDf10 sdf10 = sdf10Imdg.getSingleObjectByID(sdf10Id); +// if (sdf10 == null) { +// log.error("Sdf10.id {} not found by statement.id {}", sdf10Id, statementId); +// return; +// } +// +// Long groupId = sdf10.getGenerationId(); +// //сверяем группу SDF10 пришедшего запроса с ожидаемой +// if (sdf10GroupId == null || !sdf10GroupId.equals(groupId)) { +// log.error("do not currently waiting for gateway response for statement.id {} sdf10 groupId {}; waiting for {}", +// statementId, +// groupId, +// sdf10Id); +// return; +// } +// +// Instant updatedTime = Instant.now(); +// if (gatewayMsg.isApproved()) { +// Security security = scrtSlct.selectSecurityById(stmt.getSecurityId()); +// createDs_iResponseFromGateway( +// stmt.getAccountId(), +//// stmt.getAddresseeId(), +// security != null ? security.getSecuritySymbol() : null, +// stmt.getAmount().negate(), +// sdf10.getOutDocument()); +// //tcrId, companyId, securitySymbol +// processedApproved(stmt, sdf10, updatedTime); +// } else { +// log.trace("statement.id={}, sdf11.id={}, sdf10.id={} rejected (by gateway answer)", +// statementId, +// sdf10.getGenerationId(), +// sdf10.getId()); +// SDf11 sdf11 = createSdf11(sdf10, sdf11GroupId, now, GATEWAY_REJECTED); +// sdf11Imdg.insert(sdf11); +// stmt.setOperationStatus(OperationStatus.Rejected.getKey()); +// stmt.setUpdated(updatedTime); +// stmt.setOutSDfId(sdf11.getId()); +// statementImdg.update(stmt); +// } +// } +// if (requestIsIntendedForSdf10) { +// sendToExporter(sdf11GroupId); +// log.debug("All gateway responses received for SDF10 groupId {}. SDF11 groupId {}", sdf10GroupId, sdf11GroupId); +// clearContext(); +// } else { +// log.debug("gateway response was not for SDF10 executor"); +// } +// } private void sendToExporter(Long generationId) { SwtExporterRequest swtReq = new SwtExporterRequest(); @@ -406,4 +448,6 @@ public class Sdf10Executor { } return GENERAL_ERROR; } + + private static record Trio(SDf10 sDf10, Statement stmt, Security security) {} } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/GatewayBatchRequester.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/GatewayBatchRequester.java new file mode 100644 index 000000000..4ac6a5f71 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/GatewayBatchRequester.java @@ -0,0 +1,146 @@ + +package ru.spcex.clearing.session.stage.impl; + +import org.apache.kafka.clients.consumer.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.context.annotation.Scope; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.config.element.ClearingServiceSettings; +import ru.spcex.clearing.config.element.SessionStageSettings; +import ru.spcex.clearing.error.ClearingError; +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.SingleAssetResponse; +import ru.spcex.clearing.platform.messaging.service.QueueConsumer; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +import ru.spcex.platform.utils.enumeration.EnumMessage; +import ru.spcex.platform.utils.log.ExceptionUtils; + +import java.util.*; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; + +@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE) +@Component +public class GatewayBatchRequester extends QueueConsumer implements InitializingBean { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final KafkaSender kafkaSender; + private final Integer GATEWAY_TIMEOUT; + private String name; + + private final Lock gatewayLock = new ReentrantLock(); + private final Condition gatewayCondition = gatewayLock.newCondition(); + + private Long kafkaReqId = null; + private List batchResponses = null; + + @Autowired + public GatewayBatchRequester(@Qualifier("kafkaConsumerGateway") Supplier> consumer, + ClearingServiceSettings settings, + KafkaSender kafkaSender) { + super(consumer.get()); + this.kafkaSender = kafkaSender; + SessionStageSettings sessionSettings = settings.getSessionStage(); + this.GATEWAY_TIMEOUT = sessionSettings != null ? sessionSettings.getInspectionGatewayTimeout() : 10; + } + + @Override + public void afterPropertiesSet() throws Exception { + callback(AssetOperationApprovalRequest.class) + .setConsumer(this::receiveGatewayAnswer) + .forDestination(Consts.ASSET_OPERATION_APPROVAL, callbacks::put); + init(); + } + + public Optional> gatewayRequestAndWaitBatch(Collection requests) { + if (requests == null || requests.size() == 0) { + return Optional.of(Collections.emptyList()); + } + List responseReceived = null; + Long kafkaReqLogging = null; + gatewayLock.lock(); + try { + if (this.kafkaReqId != null) { + log.error("{}.kafkaReqId={} already sent to gateway", getName(), this.kafkaReqId); + throw new IllegalStateException(" forbid reusing GatewayRequester instance in multiple threads"); + } + //send Request + AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest(); + assetOperationListRequest.setAssetOperationRequests(requests); + this.kafkaReqId = kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest); + log.debug("sent {}.id={} to gateway", getName(), this.kafkaReqId); + boolean gatewayReceived = gatewayCondition.await(GATEWAY_TIMEOUT, TimeUnit.SECONDS); + kafkaReqLogging = this.kafkaReqId; + if (gatewayReceived) { + responseReceived = batchResponses; + batchResponses = null; + } else { + kafkaReqId = null; + } + } catch (InterruptedException e) { + log.error(ExceptionUtils.getStackTrace(e)); + } finally { + gatewayLock.unlock(); + } + if (responseReceived != null) { + log.debug("{}.id={} answer from gateway received. size: {}", getName(), kafkaReqLogging, responseReceived.size()); + return Optional.of(responseReceived); + } else { + log.error("{}.id={} answer from gateway not received.", getName(), kafkaReqLogging); + return Optional.empty(); + } + } + + private void receiveGatewayAnswer(BaseRequest sysReq) { + AssetOperationApprovalRequest requestPayload = sysReq.getRequestPayload(); + Long initialReqId = requestPayload.getInitialRequestId(); + log.trace("gateway answer received, BaseRequest.id={} initialRequestId={}", sysReq.getId(), initialReqId); + List approvals = requestPayload.getApprovals(); + if (approvals == null) { + log.trace("gateway BaseRequest.id={} initialReqId={} approvals null", sysReq.getId(), initialReqId); + return; + } + gatewayLock.lock(); + try { + if (kafkaReqId == null) { + log.trace("gateway BaseRequest.id={} initialReqId={} currently not waiting for {} answer", sysReq.getId(), initialReqId, getName()); + return; + } + if (Objects.equals(this.kafkaReqId, initialReqId)) { + log.trace("gateway BaseRequest.id={} answer found for {}.id={}", sysReq.getId(), getName(), initialReqId); + this.kafkaReqId = null; + this.batchResponses = requestPayload.getApprovals() != null ? requestPayload.getApprovals() : Collections.emptyList(); + gatewayCondition.signal(); + } + } finally { + gatewayLock.unlock(); + } + } + + public String getName() { + return name != null ? name : "entity"; + } + + public void setName(String name) { + this.name = name; + } + + public static EnumMessage mapError(Optional gtwRsp) { + if (gtwRsp.isEmpty()) { + return new EnumMessage(ClearingError.GatewayTimeout); + } else if (!gtwRsp.get()) { + return new EnumMessage(ClearingError.GatewayNotApproved); + } else throw new IllegalArgumentException("gateway response was OK. cannot create error message"); + } +} 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 index 731974af5..3929dcb2d 100644 --- 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 @@ -53,6 +53,7 @@ public class OperationService { } AssetOperationApprovalRequest assetOperationApprovalRequest = new AssetOperationApprovalRequest(); assetOperationApprovalRequest.setApprovals(responsesToClearing); + assetOperationApprovalRequest.setInitialRequestId(requestId); kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION_APPROVAL, assetOperationApprovalRequest); sentAssets.remove(requestId); } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java index a20b6e9a1..6ff735d34 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java @@ -5,6 +5,8 @@ import com.fasterxml.jackson.annotation.JsonProperty; import java.util.List; public class AssetOperationApprovalRequest { + @JsonProperty + private Long initialRequestId; @JsonProperty private List approvals; @@ -15,4 +17,12 @@ public class AssetOperationApprovalRequest { public void setApprovals(List approvals) { this.approvals = approvals; } + + public Long getInitialRequestId() { + return initialRequestId; + } + + public void setInitialRequestId(Long initialRequestId) { + this.initialRequestId = initialRequestId; + } }