clearing-service http://jira.mfd.msk:8088/browse/CLS-188 доделки по новому ТЗ

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));
+            }
+        });
+    }
+
 }

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();
     }
 }

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,7 +97,6 @@ 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);
         ImdgPredicateBuilder pb = sTradeImdg.predicateBuilder();
@@ -112,12 +112,13 @@ public class ExecutionDepositComponent {
         // Выявление новых сделок необходимо выполнить следующие контрольные проверки:
 
         // Проверить все инструменты.
+        Set<String> secCodesOfSecurity;
         {
             Set<String> 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<Security> foundSecurities = securityImdg.getCollectionObjectsByPredicate(allIn);
-            Set<String> 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();
+        LocalDate today = LocalDate.now();
-        log.info("generationId = {}", generationId);
         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());
             Collection<ExecutionDeposit> 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);
+                    newED = createExecutionDeposit(trade, null, null);
-                    verification(newED);
                     executionDepositImdg.insert(newED);
                     sendNotification(newED);
                 } catch (ClearingException ce) {
                     auditMessage(ce);
                 } catch (Exception e) {
-                    if (newED != null) {
+                    log.error("When create new ExecutionDeposit by STrade[{}] error: {}", trade.getId(), ExceptionUtils.getStackTrace(e));
-                        newED.setCoverageStatus(Allowed.DENIED.getKey());
-                    }
-                    log.error("When create new ExecutionDeposit by STrade[{}]", trade.getId());
                 }
 
             } else {
-                long[] idToLong = existsEDeposit.stream().mapToLong(ed -> ed.getId()).toArray();
+                long[] idToLong = existsEDeposit.stream().mapToLong(SpcexObjectBase::getId).toArray();
-                log.warn("S_TRADE[{}] already has executionDeposit: {}", trade.getId(), Arrays.toString(idToLong));
+                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<String> newSymbolRequest) {
-        final String destination = Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW;
-        List<String> symbolRequests = new ArrayList<>(newSymbolRequest); // чтобы в случае ошибки отобразить номер в логе
-        List<Future<RecordMetadata>> 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<Object> 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<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, request));
-                sendAll.add(send);
-            }
-        }
-        int i = 0;
-        for (Future<RecordMetadata> 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();

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 * ?

clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/object/ExecutionDepositMapStore.java

@@ -27,6 +27,10 @@ public class ExecutionDepositMapStore extends TemplateMapStore<ExecutionDeposit>
         return IMDGDistributedNames.Map_ExecutionDeposit;
     }
 
+    public String[] getIndexingField() {
+        return new String[]{"exchangeExecutionId"};
+    }
+
     @Override
     public String[] getFields() {
         return new String[]{"ID", "CREATED_AT", "UPDATED_AT",