diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ClearingService.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ClearingService.java index 7213ef7f6..18079d147 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ClearingService.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ClearingService.java @@ -20,13 +20,16 @@ public class ClearingService implements DisposableBean { private final PaymentUpdateBySdf04 paymentUpdater; private final VerificationResultComponent verificationResultComponent; private final Clearing clearing; + private final ExecutionDepositComponent executionDepositComponent; @Autowired public ClearingService(SdfCreatorBySTLDPayment sdfCreator, PaymentUpdateBySdf04 paymentUpdater, - VerificationResultComponent verificationResultComponent, Clearing clearing) { + VerificationResultComponent verificationResultComponent, Clearing clearing, + ExecutionDepositComponent executionDepositComponent) { this.sdfCreator = sdfCreator; this.paymentUpdater = paymentUpdater; this.verificationResultComponent = verificationResultComponent; + this.executionDepositComponent = executionDepositComponent; this.clearing = clearing; this.executor = Executors.newSingleThreadExecutor(); } @@ -81,4 +84,16 @@ public class ClearingService implements DisposableBean { } }); } + + public void executeSTrade() { + log.info("start STrade check for ExecutionDeposit"); + executor.execute(() -> { + try { + executionDepositComponent.processNewTS(); + } catch (Throwable e) { + log.error("{}", ExceptionUtils.getStackTrace(e)); + } + }); + } + } 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 76e0aba39..66b6c8f47 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 @@ -34,6 +34,9 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean { callback(Object.class) //todo check Object suitable .setConsumer(event -> clearingService.executeClearing()) .forDestination(Task.startOfClearing.topic(), callbacks::put); + callback(Object.class) + .setConsumer(event -> clearingService.executeSTrade()) + .forDestination(Task.getOfTrades.topic(), callbacks::put); init(); } } 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 761141c85..f45bca061 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 @@ -23,6 +23,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.registry.DealRegisterNewRequest; import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityNewRequest; +import ru.spcex.platform.classes.base.SpcexObjectBase; import ru.spcex.platform.enumeration.Allowed; import ru.spcex.platform.enumeration.Market; import ru.spcex.platform.imdg.api.Imdg; @@ -96,11 +97,10 @@ public class ExecutionDepositComponent { log.info("Reset trading day for search STrade: tradeNum={}, tradeDat={}", tradeNum, tradingDay); } - @Scheduled(cron = "${clearing-service.scheduler.check-s-trade}") public void processNewTS() { - log.debug("Start check new S_TRADE. Start tradeNum={}", tradeNum); + log.debug("Start check new S_TRADE after {}", tradingDay); ImdgPredicateBuilder pb = sTradeImdg.predicateBuilder(); - ImdgPredicate sql = pb.and(pb.greater("tradeNum", tradeNum), pb.greatEqual("tradeDateTime", tradingDay)); + ImdgPredicate sql = pb.greatEqual("tradeDateTime", tradingDay); Collection sTrades = sTradeImdg.getCollectionObjectsByPredicate(sql); log.info("Found {} new s_trade with trade_num>{}", sTrades.size(), tradeNum); @@ -112,12 +112,13 @@ public class ExecutionDepositComponent { // Выявление новых сделок необходимо выполнить следующие контрольные проверки: // Проверить все инструменты. + Set secCodesOfSecurity; { Set secCodesOfSTrade = sTrades.stream().map(STrade::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); - Set secCodesOfSecurity = foundSecurities.stream().map(Security::getSecuritySymbol).filter(Objects::nonNull).collect(Collectors.toSet()); + 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()); @@ -126,43 +127,41 @@ public class ExecutionDepositComponent { notFoundSymbol.removeAll(secCodesOfSecurity); log.info("Found only {} Security by {} secCodes from STrade. Not found: {}", secCodesOfSecurity.size(), secCodesOfSTrade.size(), notFoundSymbol); - createNewSecurities(notFoundSymbol); - log.info("Stop till they all will be created"); - - auditMessage("В security нет записей с securitySymbol", notFoundSymbol); - return; } } - Long generationId = idGenerator.nextId(); - log.info("generationId = {}", generationId); + LocalDate today = LocalDate.now(); for (STrade trade : sTrades) { - log.trace("Check s_trade[{}].tradeNum={}", trade.getId(), trade.getTradeNum()); + log.trace("Check s_trade[{}].tradeNum={} on date {}", trade.getId(), trade.getTradeNum(), today); Collection existsEDeposit = executionDepositImdg.getCollectionObjectsByFieldValues(Map.of( "exchangeExecutionId", trade.getTradeNum(), - "exchangeExecutionTime", trade.getTradeDateTime() + "clearingDate", today )); if (existsEDeposit.isEmpty()) { log.trace("S_TRADE[{}] new", trade.getId()); - ExecutionDeposit newED = null; + + // Проверка 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, Allowed.ALLOWED/*todo уточнить момент заполнения*/, generationId); - verification(newED); + newED = createExecutionDeposit(trade, null, null); executionDepositImdg.insert(newED); sendNotification(newED); } catch (ClearingException ce) { auditMessage(ce); } catch (Exception e) { - if (newED != null) { - newED.setCoverageStatus(Allowed.DENIED.getKey()); - } - log.error("When create new ExecutionDeposit by STrade[{}]", trade.getId()); + log.error("When create new ExecutionDeposit by STrade[{}] error: {}", trade.getId(), ExceptionUtils.getStackTrace(e)); } } else { - long[] idToLong = existsEDeposit.stream().mapToLong(ed -> ed.getId()).toArray(); - log.warn("S_TRADE[{}] already has executionDeposit: {}", trade.getId(), Arrays.toString(idToLong)); + long[] idToLong = existsEDeposit.stream().mapToLong(SpcexObjectBase::getId).toArray(); + log.trace("S_TRADE[{}] already has executionDeposit: {}", trade.getId(), Arrays.toString(idToLong)); } } @@ -171,14 +170,6 @@ public class ExecutionDepositComponent { log.info("Process completed. Next tradeNum is {}", newMaxTradeNum); } - protected void verification(ExecutionDeposit forED) throws ClearingException { - /*todo Рассчитанные в КС контрольные суммы (общее количество сделок и суммарный объем заключенных сделок в денежном выражении) - должны совпадать со значениями, рассчитанными Торговой системой: - count(execution[tradingDay]) = count (trade_arqua) - */ - // использовать ли VerificationResultComponent для сверки или здесь код добавить. - - } protected void auditMessage(ClearingException ce) { log.error("AUDIT error code {}: {}", ce.getEnumMsg(), ce.getMessage()); @@ -192,51 +183,6 @@ public class ExecutionDepositComponent { log.error("audit \"clearing-service\", errorText: {}", txt); } - /** - * в очередь kafka для модуля securities-service сообщение о добавлении инструмента с параметром securitySymbol=s_trade.sec_code - * - * @param newSymbolRequest - */ - protected void createNewSecurities(Collection newSymbolRequest) { - final String destination = Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW; - List symbolRequests = new ArrayList<>(newSymbolRequest); // чтобы в случае ошибки отобразить номер в логе - List> sendAll = new ArrayList<>(symbolRequests.size()); - for (String newSymbol : symbolRequests) { - if (newSymbol == null || newSymbol.isEmpty()) { - log.warn("Empty SecuritySumbol"); - } else { - MoneyMarketSecurityNewRequest requestPayload = new MoneyMarketSecurityNewRequest(); - requestPayload.setSecuritySymbol(newSymbol); - - BaseRequest request = new BaseRequest<>(); - request.setId(idGenerator.nextId()); - request.setActionType(ActionType.NEW); - request.setRequestPayload(requestPayload); - -// saveRequestToStorage(destination, request); //сохраняет данные о запросе в хранилище - log.trace("Send to {} new symbol \"{}\" ", destination, newSymbol); - Future send = kafka.send(new ProducerRecord<>(destination, request)); - sendAll.add(send); - } - } - int i = 0; - for (Future future : sendAll) { - try { - future.get(); // get exception - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - String about = i < symbolRequests.size() ? symbolRequests.get(i) : "(out of range i=" + i + ")"; - log.warn("Thread interrupted! On send symbol \"{}\"", about); - throw new RuntimeException(e); - } catch (ExecutionException e) { - String about = i < symbolRequests.size() ? symbolRequests.get(i) : "(out of range i=" + i + ")"; - log.error("Error send message for symbol \"{}\" to {}: {}", destination, - about, ExceptionUtils.getStackTrace(e.getCause() == null ? e : e.getCause())); - } - i++; - } - } - protected void sendNotification(ExecutionDeposit forED) { final String destination = Consts.REGISTRY_DEAL_REGISTER_NEW; DealRegisterNewRequest requestPayload = new DealRegisterNewRequest(); diff --git a/clearing-parent/clearing-service/src/main/resources/application.properties b/clearing-parent/clearing-service/src/main/resources/application.properties index bb4e61e6b..f80985c62 100644 --- a/clearing-parent/clearing-service/src/main/resources/application.properties +++ b/clearing-parent/clearing-service/src/main/resources/application.properties @@ -20,4 +20,3 @@ clearing-service.kafka-producer.linger-ms=1 clearing-service.kafka-producer.buffer-memory=33554432 clearing-service.scheduler.check-payment-instruction=*/5 * * * * * -clearing-service.scheduler.check-s-trade=0 * 0 1 * ? diff --git a/clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/object/ExecutionDepositMapStore.java b/clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/object/ExecutionDepositMapStore.java index c8a5185dd..ffd05dfb0 100644 --- a/clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/object/ExecutionDepositMapStore.java +++ b/clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/object/ExecutionDepositMapStore.java @@ -27,6 +27,10 @@ public class ExecutionDepositMapStore extends TemplateMapStore return IMDGDistributedNames.Map_ExecutionDeposit; } + public String[] getIndexingField() { + return new String[]{"exchangeExecutionId"}; + } + @Override public String[] getFields() { return new String[]{"ID", "CREATED_AT", "UPDATED_AT",