From ed47b5291b65076204fc4aa924716987d9b364fb Mon Sep 17 00:00:00 2001 From: ialbert Date: Wed, 3 May 2023 17:29:23 +0300 Subject: [PATCH] http://jira.mfd.msk:8088/browse/CLS-277 --- .../ru/spcex/clearing/config/KafkaConfig.java | 12 + .../clearing/config/ValidationConfig.java | 22 ++ .../spcex/clearing/error/ClearingError.java | 8 +- .../service/ExecutionDepositComponent.java | 278 +++++++----------- .../validation/STradesValidationRule.java | 111 +++++++ .../service/validation/ValidationStored.java | 5 + .../spcex/platform/enumeration/Section.java | 24 ++ .../messaging/service/sender/KafkaSender.java | 11 + .../service/sender/KafkaSenderBuilder.java | 2 + .../sender/KafkaSenderBuilderImpl.java | 6 + .../sender/KafkaSyncRequestReplySender.java | 78 +++++ 11 files changed, 380 insertions(+), 177 deletions(-) create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/STradesValidationRule.java create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java create mode 100644 platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Section.java create mode 100644 platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSyncRequestReplySender.java diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/KafkaConfig.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/KafkaConfig.java index 0a281b5e1..5a7e0add1 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/KafkaConfig.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/KafkaConfig.java @@ -46,4 +46,16 @@ public class KafkaConfig { }) .build(); } + + @Autowired + @Bean("kafkaSenderWithoutRequestInfo") + public KafkaSender kafkaSenderWithoutRequestInfo(Producer kafkaProducer, ImdgProvider imdgProvider) { + ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator(); + return KafkaSender + .setup() + .producer(kafkaProducer) + .idGenerator(imdgIdGenerator::nextId) + .saveRequestInfo(false) + .build(); + } } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java index 18854bb55..90024d854 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java @@ -7,9 +7,13 @@ import ru.clearing.classes.statics.data.account.AccountBalance; import ru.clearing.classes.statics.data.company.Company; import ru.clearing.classes.statics.data.company.relation.Relation; import ru.clearing.classes.statics.data.execution.ExecutionDeposit; +import ru.clearing.classes.statics.data.misc.STrades; +import ru.clearing.classes.statics.data.registry.TradingClearingRegistry; +import ru.clearing.classes.statics.data.security.Security; import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.service.validation.ClrngValidationStored; import ru.spcex.clearing.service.validation.ExecutionDepositValidationRule; +import ru.spcex.clearing.service.validation.STradesValidationRule; import ru.spcex.platform.classes.base.SpcexObjectBase; import ru.spcex.platform.enumeration.ClearingCategory; import ru.spcex.platform.imdg.api.Imdg; @@ -22,6 +26,7 @@ import java.util.HashMap; import java.util.Map; import java.util.function.BiConsumer; import java.util.function.BiFunction; +import java.util.function.Function; @Configuration public class ValidationConfig { @@ -36,6 +41,9 @@ public class ValidationConfig { addImdg.accept(IMDGDistributedNames.Map_Company, Company.class); addImdg.accept(IMDGDistributedNames.Map_Account, Account.class); addImdg.accept(IMDGDistributedNames.Map_AccountBalance, AccountBalance.class); + addImdg.accept(IMDGDistributedNames.Map_Security, Security.class); + addImdg.accept(IMDGDistributedNames.Map_Company, Company.class); + addImdg.accept(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class); } /** @@ -63,4 +71,18 @@ public class ValidationConfig { }; } + @Bean("sTradesValidator") + public Function sTradesValidator() { + return sTrades -> { + ImdgValidationContext context = new ImdgValidationContext<>(); + context.setValidatedObject(sTrades); + addImdg.accept(context, IMDGDistributedNames.Map_Security); + addImdg.accept(context, IMDGDistributedNames.Map_Company); + addImdg.accept(context, IMDGDistributedNames.Map_TradingClearingRegistry); + return new ValidatorImpl<>(context, + STradesValidationRule.SecurityPresent, + STradesValidationRule.CompanyPresent, + STradesValidationRule.TradingClearingRegistryPresent); + }; + } } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java index 2e9d0341c..91ec6aa4d 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java @@ -1,13 +1,17 @@ package ru.spcex.clearing.error; -import ru.spcex.platform.utils.enumeration.IEnumId; +import ru.spcex.platform.utils.enumeration.IErrorEnumId; -public enum ClearingError implements IEnumId { +public enum ClearingError implements IErrorEnumId { GeneralError(5400L), RecordNotFound(5406L), CompanyCreditCheck(5412L), CompanyDebitCheck(5413L), CompanyNotFound(5410L), + SecurityNotFound(5416L), + TradingClearingRegistryNotFound(5418L), + TradingClearingRegistryNotActive(5419L), + NewDealsNotFound(5423L), ; private final Long id; diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ExecutionDepositComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ExecutionDepositComponent.java index 7cb432dcd..8ad8ec77d 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ExecutionDepositComponent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ExecutionDepositComponent.java @@ -1,11 +1,10 @@ package ru.spcex.clearing.service; import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.clients.producer.RecordMetadata; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @@ -14,33 +13,31 @@ import ru.clearing.classes.statics.data.company.Company; import ru.clearing.classes.statics.data.execution.ExecutionDeposit; import ru.clearing.classes.statics.data.misc.Listing; import ru.clearing.classes.statics.data.misc.STrades; +import ru.clearing.classes.statics.data.registry.TradingClearingRegistry; import ru.clearing.classes.statics.data.security.Security; import ru.spcex.clearing.error.ClearingError; import ru.spcex.clearing.error.ClearingException; import ru.spcex.clearing.imdg.IMDGDistributedNames; -import ru.spcex.clearing.platform.messaging.domain.ActionType; -import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import ru.spcex.clearing.platform.messaging.domain.Consts; import ru.spcex.clearing.platform.messaging.domain.cud.registry.DealRegisterNewRequest; -import ru.spcex.platform.enumeration.Allowed; -import ru.spcex.platform.enumeration.Market; -import ru.spcex.platform.enumeration.MoneyFlowSide; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +import ru.spcex.clearing.service.validation.ValidationStored; +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.ImdgPredicate; -import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; -import ru.spcex.platform.utils.enumeration.EnumMessage; +import ru.spcex.platform.utils.enumeration.*; import ru.spcex.platform.utils.log.ExceptionUtils; import ru.spcex.platform.utils.time.TimeUtil; +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.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.stream.Collectors; +import java.util.Collection; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; /** * 1.35. executionDeposit - Сделки @@ -64,10 +61,15 @@ public class ExecutionDepositComponent { Long tradeNum; Instant tradingDay; + private final IMessageResolver msgResolver = new SimpleMessageResolver(); + private final Function stradesValidator; + private final KafkaSender kafkaSender; @Autowired - public ExecutionDepositComponent(ImdgProvider imdgProvider, Producer kafka) { + public ExecutionDepositComponent(ImdgProvider imdgProvider, Producer kafka, + @Qualifier("sTradesValidator") Function stradesValidator, + @Qualifier("kafkaSenderWithoutRequestInfo") KafkaSender kafkaSender) { this.imdgProvider = imdgProvider; this.sTradeImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class); this.securityImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Security, Security.class); @@ -78,6 +80,8 @@ public class ExecutionDepositComponent { this.idGenerator = imdgProvider.getImdgIdGenerator(); this.kafka = kafka; + this.stradesValidator = stradesValidator; + this.kafkaSender = kafkaSender; resetTradingDay(); } @@ -98,102 +102,53 @@ public class ExecutionDepositComponent { public void processNewTS() { log.debug("Start check new S_TRADE after {}", tradingDay); - Collection sTrades; - { - ImdgPredicateBuilder pb = sTradeImdg.predicateBuilder(); - ImdgPredicate sql = pb.greatEqual("tradeDateTime", tradingDay); - sTrades = sTradeImdg.getCollectionObjectsByPredicate(sql); - } - log.info("Found {} new s_trade with trade_num>{}", sTrades.size(), tradeNum); + //выбираем STrades на сегодня с правильным section + Collection sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( + "tradeDate", LocalDate.now(), + "section", Section.MKR.getKey() + )); + log.info("Found {} s_trade for today", sTrades.size()); if (sTrades.isEmpty()) { - log.info("No new sTrades."); + logError(ClearingError.NewDealsNotFound); return; } - // Выявление новых сделок необходимо выполнить следующие контрольные проверки: - - // Проверить все инструменты. - Set secCodesOfSecurity; - { - Set secCodesOfSTrade = sTrades.stream().map(STrades::getSecCode).filter(Objects::nonNull).collect(Collectors.toSet()); - log.debug("Verify {} instruments for {} STrade's.", secCodesOfSTrade.size(), sTrades.size()); - ImdgPredicate allIn = securityImdg.predicateBuilder().in("securitySymbol", secCodesOfSTrade.toArray(new String[0])); - Collection foundSecurities = securityImdg.getCollectionObjectsByPredicate(allIn); - secCodesOfSecurity = foundSecurities.stream().map(Security::getSecuritySymbol).filter(Objects::nonNull).collect(Collectors.toSet()); - if (secCodesOfSecurity.containsAll(secCodesOfSTrade)) { - log.debug("All {} Security found by {} secCodes from STrade", - secCodesOfSecurity.size(), secCodesOfSTrade.size()); - } else { - HashSet notFoundSymbol = new HashSet<>(secCodesOfSTrade); - notFoundSymbol.removeAll(secCodesOfSecurity); - log.info("Found only {} Security by {} secCodes from STrade. Not found: {}", - secCodesOfSecurity.size(), secCodesOfSTrade.size(), notFoundSymbol); + //убираем уже добавленные в ExecutionDeposit + sTrades.removeIf(sTrd -> { + Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); + if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { + log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); + return true; } - } + MoneyFlowSide excDepSide = sTrdSide.equals(Side.BUY) ? MoneyFlowSide.BUY : MoneyFlowSide.SELL; + return executionDepositImdg.getSingleObjectByFieldValues( + Map.of("tradingDate", sTrd.getTradeDate(), + "exchangeExecutionId", sTrd.getTradeNum(), + "side", excDepSide.getKey())) != null; + }); + log.info("{} strades left after already-added filtering", sTrades.size()); - // Необходимо проверять отсутствие ExecutionDeposit с exchangeExecutionId и clearingDate и side. - // Но при этом, во время создания новых ExecutionDeposit по STrade, TradeNum могут повторяться. - LocalDate today = LocalDate.now(); + for (STrades sTrd : sTrades) { + log.trace("S_TRADE[{}] new", sTrd.getId()); - for (STrades trade : sTrades) { - log.trace("Check s_trade[{}].tradeNum={} operation={} on date {}", trade.getId(), trade.getTradeNum(), trade.getOperation(), today); - boolean existEDeposit; - { - ImdgPredicateBuilder pb = executionDepositImdg.predicateBuilder(); - - ImdgPredicate predicateOperation = pb.equals("side", trade.getOperation()); - if ("B".equalsIgnoreCase(trade.getOperation())) { - predicateOperation = pb.or(predicateOperation, pb.equals("side", MoneyFlowSide.BUY.getKey())); - } -// if ("BUY".equalsIgnoreCase(trade.getOperation())) { -// predicateOperation = pb.or(predicateOperation, pb.equals("side", "B")); -// } - if ("S".equalsIgnoreCase(trade.getOperation())) { - predicateOperation = pb.or(predicateOperation, pb.equals("side", MoneyFlowSide.SELL.getKey())); - } -// if ("SELL".equalsIgnoreCase(trade.getOperation())) { -// predicateOperation = pb.or(predicateOperation, pb.equals("side", "S")); -// } - - ImdgPredicate predicate = pb.and( - pb.equals("exchangeExecutionId", trade.getTradeNum()), - pb.equals("clearingDate", today), - predicateOperation - ); - Collection existsEDeposit = executionDepositImdg.getCollectionObjectsByPredicate(predicate); -// Collection existsEDeposit=executionDepositImdg.getCollectionObjectsByFieldValues(Map.of( -// "exchangeExecutionId", trade.getTradeNum(), -// "clearingDate", today -// )); - existEDeposit = !existsEDeposit.isEmpty(); - log.trace("Check exist ExecutionDeposit, SQL \"{}\", found {}, exist {}", predicate, existsEDeposit.size(), existEDeposit); + //проверки + IValidator validator = stradesValidator.apply(sTrd); + Optional error = validator.tillFirstError(); + if (error.isPresent()) { + logError(sTrd.getId(), error.get()); + continue; } - - if (!existEDeposit) { - log.trace("S_TRADE[{}] new", trade.getId()); - - // Проверка secCode - if (!secCodesOfSecurity.contains(trade.getSecCode())) { - log.error("Error {}: STrade[{}].secCode={} not found", - ClearingError.RecordNotFound.getId(), trade.getId(), trade.getSecCode()); - continue; - } - - ExecutionDeposit newED; - try { - newED = createExecutionDeposit(trade, null, null); - executionDepositImdg.insert(newED); - sendNotification(newED); - log.debug("New executionDeposit.id={} was created.", newED.getId()); - } catch (ClearingException ce) { - auditMessage(ce); - } catch (Exception e) { - log.error("When create new ExecutionDeposit by STrade[{}] error: {}", trade.getId(), ExceptionUtils.getStackTrace(e)); - } - - } else { - log.trace("S_TRADE[{}] already has executionDeposit: exchangeExecutionId={} on date {}", trade.getId(), trade.getTradeNum(), today); + ExecutionDeposit newED; + try { + newED = createExecutionDeposit(sTrd, validator); + executionDepositImdg.insert(newED); + sendNotification(newED); + log.debug("New executionDeposit.id={} was created.", newED.getId()); + } catch (ClearingException ce) { + auditMessage(ce); + } catch (Exception e) { + log.error("When create new ExecutionDeposit by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e)); } } @@ -207,106 +162,79 @@ public class ExecutionDepositComponent { log.error("AUDIT error code {}: {}", ce.getEnumMsg(), ce.getMessage()); } - protected void auditMessage(String message, Object... ids) { - String txt = message; - if (ids != null && ids.length > 0) { - txt += " object:" + Arrays.toString(ids); - } - log.error("audit \"clearing-service\", errorText: {}", txt); - } - - protected void sendNotification(ExecutionDeposit forED) { + protected void sendNotification(ExecutionDeposit forED) throws ClearingException { final String destination = Consts.REGISTRY_DEAL_REGISTER_NEW; DealRegisterNewRequest requestPayload = new DealRegisterNewRequest(); requestPayload.setExecutionId(forED.getId()); requestPayload.setExchangeExecutionId(forED.getExchangeExecutionId()); -// requestPayload.setId(idGenerator.nextId()); - - BaseRequest request = new BaseRequest<>(); - request.setId(idGenerator.nextId()); - request.setActionType(ActionType.NEW); - request.setRequestPayload(requestPayload); - log.trace("Send to {} new ExecutionDeposit[{}]", destination, forED.getId()); - Future send = kafka.send(new ProducerRecord<>(destination, request)); - try { - send.get(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RuntimeException(e); - } catch (ExecutionException e) { - throw new RuntimeException(e); + Long generatedRequestId = kafkaSender.sendRequestToQueue(destination, requestPayload); + if (generatedRequestId == null) { + throw new RuntimeException("failed to send request to " + destination); } } protected ExecutionDeposit createExecutionDeposit(STrades sTrades, - Allowed coverageStatus, Long sessionId) throws ClearingException { - Account account = null; // todo CLS-275 accountImdg.getSingleObjectByFieldValues(Map.of("account", sTrades.getMoneyAccount())); - Security security = securityImdg.getSingleObjectByFieldValues(Map.of("securitySymbol", sTrades.getSecCode())); - if (security == null) { - log.warn("security securitySymbol=\"{}\" not found", sTrades.getSecCode()); - throw new ClearingException(new EnumMessage(ClearingError.RecordNotFound, sTrades.getSecCode())); - } - Listing listing = null; - if (security != null) { - listing = listingImdg.getSingleObjectByFieldValues(Map.of("securityId", security.getId())); - } - Company company = companyImdg.getSingleObjectByFieldValues(Map.of("tradingCode", sTrades.getFirmId())); - if (company == null) { - throw new ClearingException(new EnumMessage(ClearingError.CompanyNotFound, sTrades.getFirmId())); - } + IValidator validator) throws ClearingException { + Security security = validator.getStored(ValidationStored.STradesSecurity); + Company company = validator.getStored(ValidationStored.STradesCompany); + Company counterCompany = validator.getStored(ValidationStored.STradesCounterCompany); + TradingClearingRegistry rgstr = validator.getStored(ValidationStored.STradesTradingClearingRegistry); + Listing listing = listingImdg.getSingleObjectByFieldValues(Map.of("securityId", security.getId())); - return createExecutionDeposit(sTrades, account, listing, company, security, coverageStatus, sessionId); - } - - private ExecutionDeposit createExecutionDeposit(STrades sTrades, Account account, Listing listing, - Company company, Security security, - Allowed coverageStatus, Long sessionId) { ExecutionDeposit eDeposit = new ExecutionDeposit(); - eDeposit.setId(idGenerator.nextId()); final Instant now = Instant.now(); - final LocalDate nowDay = TimeUtil.toLocalDate(now); eDeposit.setCreated(now); - eDeposit.setTradingDate(nowDay); - eDeposit.setClearingDate(nowDay); - + eDeposit.setTradingDate(sTrades.getTradeDate()); + eDeposit.setClearingDate(TimeUtil.toLocalDate(now)); eDeposit.setExchangeExecutionId(sTrades.getTradeNum()); eDeposit.setExchangeExecutionTime(sTrades.getTradeDateTime()); - if (account != null) { - // todo CLS-275 eDeposit.setAccountId(account.getId()); - } + eDeposit.setTradingClearingRegistryId(rgstr.getId()); eDeposit.setMarket(Market.mkrs.getKey()); eDeposit.setPrice(sTrades.getPrice()); eDeposit.setLots(sTrades.getQty()); if (listing != null && listing.getLotSize() != null && eDeposit.getLots() != null) { - BigDecimal quantity = eDeposit.getLots().multiply(listing.getLotSize()); - eDeposit.setQuantity(quantity); + BigDecimal lots = eDeposit.getLots(); + BigDecimal listingLotSize = listing.getLotSize(); + if (listingLotSize != null && lots != null) { + eDeposit.setQuantity(lots.multiply(listingLotSize)); + } } eDeposit.setFirstLegAmount(sTrades.getValue()); eDeposit.setSecondLegAmount(sTrades.getValue()); - //eDeposit.setInterestAmount(null); - String operation = sTrades.getOperation(); //Символьный код по справочнику moneyFlowSide), соответствующий значению из s_trade.operation (sTrade.getOperation()) - if ("B".equalsIgnoreCase(operation)) { - operation = MoneyFlowSide.BUY.getKey(); + { + Side sTradeSide = IEnumKey.getEnumByKey(Side.class, sTrades.getOperation()); + if (sTradeSide != null) { + switch (sTradeSide) { + case BUY -> eDeposit.setSide(MoneyFlowSide.BUY.getKey()); + case SELL -> eDeposit.setSide(MoneyFlowSide.SELL.getKey()); + } + } } - if ("S".equalsIgnoreCase(operation)) { - operation = MoneyFlowSide.SELL.getKey(); - } - eDeposit.setSide(operation); - eDeposit.setSettlementCurrency("RUB"); // (справочник currencyCode) + eDeposit.setSettlementCurrency(CurrencyCode.RUB.getKey()); eDeposit.setCompanyId(company.getId()); - // todo CLS-275 eDeposit.setDuration(sTrades.getDaysToMatDate()); - eDeposit.setFirstLegSettlementDate(nowDay); - eDeposit.setSecondLegSettlementDate(sTrades.getSettleDate()); - //eDeposit.setFirstLegSettlementCode(null); - //eDeposit.setSecondLegSettlementCode(null); + eDeposit.setDuration(sTrades.getRepoTerm()); + eDeposit.setFirstLegSettlementDate(sTrades.getTradeDate()); + { + LocalDate sTrdSettleDate = sTrades.getSettleDate(); + if (sTrdSettleDate != null) { + eDeposit.setSecondLegSettlementDate(sTrdSettleDate.plusDays(sTrades.getRepoTerm())); + } + } + eDeposit.setFirstLegSettlementCode(sTrades.getSettleCode()); + eDeposit.setSecondLegSettlementCode(sTrades.getSettleCode()); eDeposit.setSecurityFullName(security.getFullName()); eDeposit.setSecuritySymbol(security.getSecuritySymbol()); eDeposit.setSecurityId(security.getId()); - //eDeposit.setCounterPartyId(null); - eDeposit.setCoverageStatus(coverageStatus == null ? null : coverageStatus.getKey()); // Заполняется по справочнику allowed в результате расчета требований и обязательств. TODO - eDeposit.setSessionId(sessionId); - + eDeposit.setContract(sTrades.getClassCode()); + eDeposit.setCounterPartyId(counterCompany.getId()); return eDeposit; } + private void logError(IEnumId subject, Object... args) { + log.warn("{}", msgResolver.resolve(new EnumMessage(subject, args))); + } + + private void logError(Long sTradeId, EnumMessage msg) { + log.warn("sTrade id={} {}", sTradeId, msgResolver.resolve(msg)); + } } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/STradesValidationRule.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/STradesValidationRule.java new file mode 100644 index 000000000..7673e340e --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/STradesValidationRule.java @@ -0,0 +1,111 @@ +package ru.spcex.clearing.service.validation; + +import ru.clearing.classes.statics.data.company.Company; +import ru.clearing.classes.statics.data.misc.STrades; +import ru.clearing.classes.statics.data.registry.TradingClearingRegistry; +import ru.clearing.classes.statics.data.security.Security; +import ru.spcex.clearing.error.ClearingError; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.platform.enumeration.ServiceStatus; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.validation.ImdgValidationContext; +import ru.spcex.platform.utils.enumeration.EnumMessage; +import ru.spcex.platform.utils.enumeration.IEnumKey; +import ru.spcex.platform.utils.text.TextUtil; +import ru.spcex.platform.utils.validation.IValidationRule; + +import java.util.Map; +import java.util.Optional; + +public enum STradesValidationRule implements IValidationRule> { + + // 2.1. Найти в security запись, у которой security.securitySymbol=sTrades.secCode. + // Если такой записи нет, записать в лог ошибку (5416) "Инструмент %s не найден". + SecurityPresent() { + @Override + public Optional validate(ImdgValidationContext context) { + STrades validatedObject = context.getValidatedObject(); + if (TextUtil.isEmpty(validatedObject.getSecCode())) { + return of(ClearingError.SecurityNotFound, validatedObject.getSecCode()); + } + Imdg securityImdg = context.obtainMap(IMDGDistributedNames.Map_Security, Security.class); + Security security = securityImdg.getSingleObjectByFieldValues(Map.of("securitySymbol", validatedObject.getSecCode())); + if (security == null) { + return of(ClearingError.SecurityNotFound, validatedObject.getSecCode()); + } + context.storeObject(ValidationStored.STradesSecurity, security); + return empty(); + } + }, + // 2.2. Найти в company запись, у которой company.tradingCode=sTrades.firmId. + // Если такой записи нет, записать в лог ошибку (5410) "Компания %s не найдена". + CompanyPresent() { + @Override + public Optional validate(ImdgValidationContext context) { + STrades validatedObject = context.getValidatedObject(); + if (TextUtil.isEmpty(validatedObject.getFirmId())) { + return of(ClearingError.CompanyNotFound, validatedObject.getFirmId()); + } + Imdg companyImdg = context.obtainMap(IMDGDistributedNames.Map_Company, Company.class); + Company company = companyImdg.getSingleObjectByFieldValues(Map.of("tradingCode", validatedObject.getFirmId())); + if (company == null) { + return of(ClearingError.CompanyNotFound, validatedObject.getFirmId()); + } + context.storeObject(ValidationStored.STradesCompany, company); + return empty(); + } + }, + CounterCompanyPresent() { + @Override + public Optional validate(ImdgValidationContext context) { + STrades validatedObject = context.getValidatedObject(); + if (TextUtil.isEmpty(validatedObject.getCpFirmId())) { + return of(ClearingError.CompanyNotFound, validatedObject.getCpFirmId()); + } + Imdg companyImdg = context.obtainMap(IMDGDistributedNames.Map_Company, Company.class); + Company company = companyImdg.getSingleObjectByFieldValues(Map.of("tradingCode", validatedObject.getCpFirmId())); + if (company == null) { + return of(ClearingError.CompanyNotFound, validatedObject.getCpFirmId()); + } + context.storeObject(ValidationStored.STradesCounterCompany, company); + return empty(); + } + }, + // 2.3. Найти в tradingClearingRegistry запись, у которой tradingClearingRegistry.code=sTrades.account + // и tradingClearingRegistry.companyId=companyId, найденому в предыдущем пункте. + // Если такой записи нет, записать в лог ошибку (5418) "Торгово-клиринговый регистр %s не найден для компании %s". + TradingClearingRegistryPresent() { + @Override + public Optional validate(ImdgValidationContext context) { + STrades validatedObject = context.getValidatedObject(); + if (TextUtil.isEmpty(validatedObject.getAccount())) { + return of(ClearingError.TradingClearingRegistryNotFound, validatedObject.getAccount()); + } + if (context.getStoredObject(ValidationStored.STradesCompany) == null) { //maybe unnecessary + return of(ClearingError.CompanyNotFound, validatedObject.getSecCode()); + } + Imdg registryImdg = context.obtainMap(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class); + TradingClearingRegistry tcRegister = registryImdg.getSingleObjectByFieldValues( + Map.of("code", validatedObject.getAccount(), + "companyId", ((Company) context.getStoredObject(ValidationStored.STradesCompany)).getId() + )); + if (tcRegister == null) { + return of(ClearingError.TradingClearingRegistryNotFound, validatedObject.getAccount()); + } + // 2.4. Проверить, что найденный торгово-клиринговый регистр в активном состоянии: + // tradingClearingRegistry.status≠BLKD/SSPD/CLOS (см. справочник serviceStatus). + // Иначе записать в лог ошибку (5419) "Торгово-клиринговый регистр %s неактивен". + ServiceStatus status = IEnumKey.getEnumByKey(ServiceStatus.class, tcRegister.getStatus()); + if (IEnumKey.contains(status, ServiceStatus.Blocked, ServiceStatus.Suspended, ServiceStatus.Closed)) { + return of(ClearingError.TradingClearingRegistryNotActive, validatedObject.getAccount()); + } + context.storeObject(ValidationStored.STradesTradingClearingRegistry, tcRegister); + return empty(); + } + }; + + @Override + public String ruleName() { + return "STradesValidationRule." + name(); + } +} \ No newline at end of file diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java new file mode 100644 index 000000000..e1c671c24 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java @@ -0,0 +1,5 @@ +package ru.spcex.clearing.service.validation; + +public enum ValidationStored { + STradesCompany, STradesCounterCompany, STradesSecurity, STradesTradingClearingRegistry +} diff --git a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Section.java b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Section.java new file mode 100644 index 000000000..ec74d7252 --- /dev/null +++ b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Section.java @@ -0,0 +1,24 @@ +package ru.spcex.platform.enumeration; + +import ru.spcex.platform.utils.enumeration.IEnumKey; + +public enum Section implements IEnumKey { + MKR("MKR"), + FOND("FOND"); + + Section(String key) { + this.key = key; + } + + private String key; + + @Override + public String getKey() { + return this.key; + } + + @Override + public boolean equalsByKey(String key) { + return IEnumKey.super.equalsByKey(key); + } +} diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSender.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSender.java index 6b321bc90..613966cc8 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSender.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSender.java @@ -24,6 +24,7 @@ public class KafkaSender { private Producer kafka; private Supplier idGenerator; private Function imdgGenerator; + private boolean saveRequestInfo = true; KafkaSender() { this.allImdgMaps = new ConcurrentHashMap<>(); @@ -55,6 +56,12 @@ public class KafkaSender { } private void saveRequestToStorage(String destination, BaseRequest request) { + if (!saveRequestInfo) { + return; + } + if (imdgGenerator == null) { + throw new IllegalStateException("imdgGenerator is null"); + } KafkaImdgInsert imdgInsert = getImdg(destination); RequestInfo requestInfo = RequestInfo.create(request.getId()); imdgInsert.insert(requestInfo); @@ -76,4 +83,8 @@ public class KafkaSender { void setImdgProvider(Function imdgGenerator) { this.imdgGenerator = imdgGenerator; } + + void setSaveRequestInfo(boolean saveRequestInfo) { + this.saveRequestInfo = saveRequestInfo; + } } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSenderBuilder.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSenderBuilder.java index f07f23463..0a6d2f6af 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSenderBuilder.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSenderBuilder.java @@ -12,5 +12,7 @@ public interface KafkaSenderBuilder { KafkaSenderBuilder imdgProvider(Function imdgGenerator); + KafkaSenderBuilder saveRequestInfo(boolean saveRequestInfo); + KafkaSender build(); } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSenderBuilderImpl.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSenderBuilderImpl.java index d20b848ec..0e89ad5cb 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSenderBuilderImpl.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSenderBuilderImpl.java @@ -31,6 +31,12 @@ public class KafkaSenderBuilderImpl implements KafkaSenderBuilder{ return this; } + @Override + public KafkaSenderBuilder saveRequestInfo(boolean saveRequestInfo) { + kafkaSender.setSaveRequestInfo(saveRequestInfo); + return this; + } + @Override public KafkaSender build() { return kafkaSender; diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSyncRequestReplySender.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSyncRequestReplySender.java new file mode 100644 index 000000000..2d176d8e6 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSyncRequestReplySender.java @@ -0,0 +1,78 @@ +package ru.spcex.clearing.platform.messaging.service.sender; + +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.kafka.requestreply.ReplyingKafkaTemplate; +import ru.spcex.clearing.platform.messaging.domain.ActionType; +import ru.spcex.clearing.platform.messaging.domain.BaseRequest; +import ru.spcex.clearing.platform.messaging.service.RequestInfo; +import ru.spcex.platform.classes.base.SpcexObjectBase; +import ru.spcex.platform.utils.log.ExceptionUtils; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.function.Function; +import java.util.function.Supplier; + +public class KafkaSyncRequestReplySender { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final Map allImdgMaps; + private Producer kafka; + private Supplier idGenerator; + private Function imdgGenerator; + private ReplyingKafkaTemplate kafkaTemplate; + + KafkaSyncRequestReplySender() { + this.allImdgMaps = new ConcurrentHashMap<>(); + } + + public static KafkaSenderBuilderImpl setup() { + return new KafkaSenderBuilderImpl(); + } + + + public Long sendRequestToQueue(String destination, Object requestPayload) { + BaseRequest request = new BaseRequest<>(); + request.setId(idGenerator.get()); + request.setActionType(ActionType.SYSTEM); + request.setRequestPayload(requestPayload); + //сохраняет данные о запросе в хранилище + saveRequestToStorage(destination, request); + Future send = kafka.send(new ProducerRecord<>(destination, request)); + try { + send.get(); + } catch (InterruptedException | ExecutionException e) { + log.error(ExceptionUtils.getStackTrace(e)); + return null; + } + return request.getId(); + } + + private void saveRequestToStorage(String destination, BaseRequest request) { + KafkaImdgInsert imdgInsert = getImdg(destination); + RequestInfo requestInfo = RequestInfo.create(request.getId()); + imdgInsert.insert(requestInfo); + } + + @SuppressWarnings("unchecked") + private KafkaImdgInsert getImdg(String mapName) { + return allImdgMaps.computeIfAbsent(mapName, (mapName1) -> imdgGenerator.apply(mapName)); + } + + void setProducer(Producer kafka) { + this.kafka = kafka; + } + + void setIdGenerator(Supplier idGenerator) { + this.idGenerator = idGenerator; + } + + void setImdgProvider(Function imdgGenerator) { + this.imdgGenerator = imdgGenerator; + } +}