From 35151a172fefb3a947d2a649af38025d5d64434d Mon Sep 17 00:00:00 2001 From: ialbert Date: Wed, 17 Jan 2024 13:44:39 +0300 Subject: [PATCH] SDF06/SDF10 timeouts This reverts commit fd900e27 --- .../clearing/service/EventsReceiver.java | 9 + .../service/executors/Sdf06Executor.java | 274 +++++++----------- .../service/executors/Sdf10Executor.java | 252 +++++++--------- 3 files changed, 221 insertions(+), 314 deletions(-) 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 0414ed83b..4527e3f19 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,6 +15,7 @@ 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; @@ -137,6 +138,14 @@ 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 d8f3ae3b2..9d1ba048a 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,7 +17,9 @@ 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; @@ -27,14 +29,16 @@ 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.clearing.session.stage.impl.GatewayBatchRequester; +import ru.spcex.platform.classes.base.SpcexObjectBase; 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; @@ -74,19 +78,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, - GatewayBatchRequester gateway) { + TradingTimeService tradingTimeService, KafkaSender kafkaSender, FilenameObtainer filenameObtainer, DmiService dmiService, AssetTBFProcessing assets) { this.imdgProvider = imdgProvider; this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class); this.tradingTimeService = tradingTimeService; @@ -102,8 +106,6 @@ public class Sdf06Executor { this.filenameObtainer = filenameObtainer; this.dmiService = dmiService; this.assets = assets; - this.gateway = gateway; - this.gateway.setName("Sdf06"); } public void execute(BaseRequest systemRequest) { @@ -116,13 +118,12 @@ 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); @@ -156,7 +157,6 @@ 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,80 +171,19 @@ public class Sdf06Executor { } } if (requests.size() > 0) { - 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 { - SDf07 sdf07 = createSdf07(sdf06, now, errorResult3); - sdf07.setGenerationId(sdf07GroupId); - sdf07Imdg.insert(sdf07); - log.trace("statement.id={}, sdf07.id={}, sdf06.id={} sdf06.getGenerationId={} rejected (by gateway answer)", - statementId, - sdf07.getId(), - sdf06.getId(), sdf06.getGenerationId()); - stmt.setOperationStatus(OperationStatus.Rejected.getKey()); - stmt.setUpdated(updatedTime); - stmt.setOutSDfId(sdf07.getId()); - statementImdg.update(stmt); - } - sdf07WasCreated = true; - } - } - } - if (sdf07WasCreated) { + this.sdf06GroupId = groupId; + this.sdf07GroupId = sdf07GroupId; + AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest(); + assetOperationListRequest.setAssetOperationRequests(requests); + kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest); + } else 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", @@ -288,87 +227,92 @@ 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); -// } -// } + 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; + } private void sendToExporter(Long generationId, String fileName) { ExportToFileRequest exportRequest = new ExportToFileRequest(); @@ -445,6 +389,4 @@ 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 ff005ad98..a3d24aaea 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,7 +19,9 @@ 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; @@ -27,12 +29,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; @@ -42,7 +44,10 @@ import ru.spcex.platform.utils.validation.IValidator; import java.math.BigDecimal; import java.time.Instant; import java.time.LocalDate; -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Map; +import java.util.Optional; import java.util.function.Function; import static ru.spcex.platform.utils.number.BigDecimalUtil.safeBD; @@ -65,12 +70,10 @@ 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 = "Не найден ТКР"; @@ -80,11 +83,14 @@ 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, GatewayBatchRequester gateway) { + TradingTimeService tradingTimeService, KafkaSender kafkaSender, AssetTBFProcessing assets) { this.imdgProvider = imdgProvider; this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class); this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); @@ -99,8 +105,6 @@ 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) { @@ -112,17 +116,16 @@ 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); @@ -162,7 +165,6 @@ 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()) ? @@ -176,65 +178,12 @@ public class Sdf10Executor { } } if (requests.size() > 0) { - 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) { + this.sdf10GroupId = groupId; + this.sdf11GroupId = sdf11GroupId; + AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest(); + assetOperationListRequest.setAssetOperationRequests(requests); + kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest); + } else if (sdf11WasCreated) { sendToExporter(sdf11GroupId); } } @@ -280,6 +229,10 @@ 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(), @@ -294,75 +247,80 @@ 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"); -// } -// } + 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; + } private void sendToExporter(Long generationId) { SwtExporterRequest swtReq = new SwtExporterRequest(); @@ -448,6 +406,4 @@ public class Sdf10Executor { } return GENERAL_ERROR; } - - private static record Trio(SDf10 sDf10, Statement stmt, Security security) {} }