execution uploading optimization

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java

@@ -31,6 +31,9 @@ import ru.spcex.clearing.platform.messaging.domain.cud.utilities.STradesImported
 import ru.spcex.clearing.platform.messaging.service.QueueConsumerV2;
 import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
 import ru.spcex.clearing.platform.messaging.service.Status;
+import ru.spcex.clearing.service.execution.ExecutionCurrencyComponent;
+import ru.spcex.clearing.service.execution.ExecutionDepositComponent;
+import ru.spcex.clearing.service.execution.ExecutionFondComponent;
 import ru.spcex.clearing.service.executors.Sdf06Executor;
 import ru.spcex.clearing.service.executors.Sdf10Executor;
 import ru.spcex.clearing.service.payment.PaymentInstructionOutboundService;
@@ -76,6 +79,9 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean
     private final StatementServiceV2 statementService;
     private final SessionTerminator sessionTerminator;
     private final PaymentInstructionOutboundService pmtOutboundService;
+    private final ExecutionCurrencyComponent execCurrUpload;
+    private final ExecutionDepositComponent execDepUpload;
+    private final ExecutionFondComponent execFondUpload;
 
     @Autowired
     public EventsReceiver(@Qualifier("kafkaConsumer") Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaResponseQueue,
@@ -87,7 +93,7 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean
                           CurrencySession currencySession,
                           PrimaryAuctionB0Session primaryAuctionB0Session, PrimaryAuctionT0Session primaryAuctionT0Session, IntermediateMkrSession intermediateMkrSession, FinalMkrSession finalMkrSession, UnitedSession unitedSession, ReturnDepositSession returnDepositSession, SessionManager sessionManager,
                           Sdf06Executor sdf06Executor,
-                          Sdf10Executor sdf10Executor, BalanceRevise balanceRevise, Sdf05Sender sdf05Sender, StatementServiceV2 statementService, SessionTerminator sessionTerminator, PaymentInstructionOutboundService pmtOutboundService) {
+                          Sdf10Executor sdf10Executor, BalanceRevise balanceRevise, Sdf05Sender sdf05Sender, StatementServiceV2 statementService, SessionTerminator sessionTerminator, PaymentInstructionOutboundService pmtOutboundService, ExecutionCurrencyComponent execCurrUpload, ExecutionDepositComponent execDepUpload, ExecutionFondComponent execFondUpload) {
         super(kafkaQueue, kafkaResponseQueue);
         this.environment = environment;
         this.errorResolver = errorResolver;
@@ -110,10 +116,16 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean
         this.statementService = statementService;
         this.sessionTerminator = sessionTerminator;
         this.pmtOutboundService = pmtOutboundService;
+        this.execCurrUpload = execCurrUpload;
+        this.execDepUpload = execDepUpload;
+        this.execFondUpload = execFondUpload;
     }
 
     @Override
     public void afterPropertiesSet() {
+        execCurrUpload.initExecCash();
+        execDepUpload.initExecCash();
+        execFondUpload.initExecCash();
         callback(Sdf04Request.class)
                 .setConsumer(event -> {
                     Sdf04Request requestPayload = event.getRequestPayload();

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadKey.java

@@ -0,0 +1,37 @@
+package ru.spcex.clearing.service.execution;
+
+import java.time.LocalDate;
+import java.util.Objects;
+import ru.clearing.classes.statics.data.execution.ExecutionCommon;
+
+public record ExecUploadKey(LocalDate tradeDate, Long exchangeExecutionId, String side) {
+
+    public static ExecUploadKey cash(LocalDate tradingDate, Long exchangeExecutionId, String side) {
+        return new ExecUploadKey(tradingDate, exchangeExecutionId, side);
+    }
+
+    public static ExecUploadKey cash(ExecutionCommon exec) {
+        return new ExecUploadKey(exec.getTradingDate(), exec.getExchangeExecutionId(), exec.getSide());
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        ExecUploadKey that = (ExecUploadKey) o;
+
+        if (!Objects.equals(tradeDate, that.tradeDate)) return false;
+        if (!Objects.equals(exchangeExecutionId, that.exchangeExecutionId))
+            return false;
+        return Objects.equals(side, that.side);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = tradeDate != null ? tradeDate.hashCode() : 0;
+        result = 31 * result + (exchangeExecutionId != null ? exchangeExecutionId.hashCode() : 0);
+        result = 31 * result + (side != null ? side.hashCode() : 0);
+        return result;
+    }
+}

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java

@@ -5,8 +5,11 @@ import java.time.Instant;
 import java.time.LocalDate;
 import java.time.format.DateTimeFormatter;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
@@ -32,6 +35,7 @@ import ru.spcex.clearing.service.validation.ValidationStored;
 import ru.spcex.platform.enumeration.Section;
 import ru.spcex.platform.enumeration.Side;
 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;
@@ -48,11 +52,12 @@ import ru.spcex.platform.utils.validation.IValidator;
  */
 @Component
 @EnableScheduling
-public class ExecutionCurrencyComponent {
+public class ExecutionCurrencyComponent implements IExecutionUploadComponent {
     private final Logger log = LoggerFactory.getLogger(getClass());
 
     private final Imdg<STrades> sTradeImdg;
     private final Imdg<ExecutionCurrency> executionCurrencyImdg;
+    private final ImdgId idGen;
     private final Imdg<Listing> listingImdg;
 
 //    protected transient Long tradeNum; //todo used?
@@ -61,6 +66,7 @@ public class ExecutionCurrencyComponent {
     private final Function<STrades, IValidator> stradesValidator;
     private final KafkaSender kafkaSender;
     private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy");
+    private final Set<ExecUploadKey> cash = new HashSet<>();
 
 
 
@@ -77,6 +83,12 @@ public class ExecutionCurrencyComponent {
         this.msgResolver = msgResolver;
 
 //        resetTradingDay();
+        this.idGen = imdgProvider.getImdgIdGenerator();
+    }
+
+    @Override
+    public void initExecCash() {
+        ExecutionUploadCashUtil.loadExecutions(executionCurrencyImdg, cash);
     }
 
 //    /**
@@ -127,13 +139,19 @@ public class ExecutionCurrencyComponent {
                 return true;
             }
 //            MoneyFlowSide excDepSide = sTrdSide == Side.BUY ? MoneyFlowSide.BUY : MoneyFlowSide.SELL;
-            return executionCurrencyImdg.getFirstObjectByFieldValues(
+            ExecUploadKey execCash = ExecUploadKey.cash(sTrd.getTradeDate(),
-                    Map.of("tradingDate", sTrd.getTradeDate(),
+                sTrd.getTradeNum(),
-                            "exchangeExecutionId", sTrd.getTradeNum(),
+                sTrdSide.getKey());
-                            "side", sTrdSide.getKey())) != null;
+            return cash.contains(execCash);
+//            return executionCurrencyImdg.getFirstObjectByFieldValues(
+//                    Map.of("tradingDate", sTrd.getTradeDate(),
+//                            "exchangeExecutionId", sTrd.getTradeNum(),
+//                            "side", sTrdSide.getKey())) != null;
+
         });
         log.info("{} strades left after already-added filtering", sTrades.size());
 
+        Map<Long, ExecutionCurrency> execsToInsert = new HashMap<>();
         for (STrades sTrd : sTrades) {
             log.trace("S_TRADE[{}] new", sTrd.getId());
 
@@ -147,7 +165,10 @@ public class ExecutionCurrencyComponent {
             ExecutionCurrency newEC;
             try {
                 newEC = createExecutionCurrency(sTrd, validator);
-                executionCurrencyImdg.insert(newEC);
+                newEC.setId(idGen.nextId());
+                execsToInsert.put(newEC.getId(), newEC);
+                cash.add(ExecUploadKey.cash(newEC));
+                //executionCurrencyImdg.insert(newEC);
                 sendNotification(newEC);
                 log.debug("New executionCurrency.id={} was created.", newEC.getId());
             } catch (ClearingException ce) {
@@ -156,6 +177,7 @@ public class ExecutionCurrencyComponent {
                 log.error("When create new ExecutionCurrency by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e));
             }
         }
+        executionCurrencyImdg.putAll(execsToInsert, 200);
 
         Long newMaxTradeNum = sTrades.stream().mapToLong(STrades::getTradeNum).max().orElse(0); // orElseGet(() -> tradeNum)
         log.info("Process completed. Next tradeNum is {}", newMaxTradeNum);

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java

@@ -5,8 +5,11 @@ import java.time.Instant;
 import java.time.LocalDate;
 import java.time.format.DateTimeFormatter;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
@@ -34,6 +37,7 @@ import ru.spcex.platform.enumeration.MoneyFlowSide;
 import ru.spcex.platform.enumeration.Section;
 import ru.spcex.platform.enumeration.Side;
 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;
@@ -51,12 +55,13 @@ import ru.spcex.platform.utils.validation.IValidator;
  */
 @Component
 @EnableScheduling
-public class ExecutionDepositComponent {
+public class ExecutionDepositComponent implements IExecutionUploadComponent {
     private final Logger log = LoggerFactory.getLogger(getClass());
 
     private final Imdg<STrades> sTradeImdg;
     private final Imdg<ExecutionDeposit> executionDepositImdg;
     private final Imdg<Listing> listingImdg;
+    private final ImdgId idGen;
 
     //fixme ждать ТЗ
     Long tradeNum;
@@ -64,6 +69,7 @@ public class ExecutionDepositComponent {
     private final IMessageResolver msgResolver;
     private final Function<STrades, IValidator> stradesValidator;
     private final KafkaSender kafkaSender;
+    private final Set<ExecUploadKey> cash = new HashSet<>();
     private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy");
 
 
@@ -76,6 +82,7 @@ public class ExecutionDepositComponent {
         this.sTradeImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class);
         this.executionDepositImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ExecutionDeposit, ExecutionDeposit.class);
         this.listingImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Listing, Listing.class);
+        this.idGen = imdgProvider.getImdgIdGenerator();
         this.stradesValidator = stradesValidator;
         this.kafkaSender = kafkaSender;
         this.msgResolver = msgResolver;
@@ -130,13 +137,17 @@ public class ExecutionDepositComponent {
                 return true;
             }
             MoneyFlowSide excDepSide = sTrdSide.equals(Side.BUY) ? MoneyFlowSide.BUY : MoneyFlowSide.SELL;
-            return executionDepositImdg.getFirstObjectByFieldValues(
+            return cash.contains(
-                    Map.of("tradingDate", sTrd.getTradeDate(),
+                ExecUploadKey.cash(sTrd.getTradeDate(), sTrd.getTradeNum(), excDepSide.getKey())
-                            "exchangeExecutionId", sTrd.getTradeNum(),
+            );
-                            "side", excDepSide.getKey())) != null;
+            //return executionDepositImdg.getFirstObjectByFieldValues(
+            //        Map.of("tradingDate", sTrd.getTradeDate(),
+            //                "exchangeExecutionId", sTrd.getTradeNum(),
+            //                "side", excDepSide.getKey())) != null;
         });
         log.info("{} strades left after already-added filtering", sTrades.size());
 
+        Map<Long, ExecutionDeposit> execsToInsert = new HashMap<>();
         for (STrades sTrd : sTrades) {
             log.trace("S_TRADE[{}] new", sTrd.getId());
 
@@ -150,7 +161,10 @@ public class ExecutionDepositComponent {
             ExecutionDeposit newED;
             try {
                 newED = createExecutionDeposit(sTrd, validator);
-                executionDepositImdg.insert(newED);
+                newED.setId(idGen.nextId());
+                cash.add(ExecUploadKey.cash(newED));
+                execsToInsert.put(newED.getId(), newED);
+                //executionDepositImdg.insert(newED);
                 sendNotification(newED);
                 log.debug("New executionDeposit.id={} was created.", newED.getId());
             } catch (ClearingException ce) {
@@ -159,6 +173,7 @@ public class ExecutionDepositComponent {
                 log.error("When create new ExecutionDeposit by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e));
             }
         }
+        executionDepositImdg.putAll(execsToInsert, 200);
 
         Long newMaxTradeNum = sTrades.stream().mapToLong(STrades::getTradeNum).max().orElseGet(() -> tradeNum);
         log.info("Process completed. Next tradeNum is {}", newMaxTradeNum);
@@ -301,4 +316,9 @@ public class ExecutionDepositComponent {
         }
         return Optional.ofNullable(counterValidator.getStored(ValidationStored.STradesTradingClearingRegistry));
     }
+
+    @Override
+    public void initExecCash() {
+        ExecutionUploadCashUtil.loadExecutions(executionDepositImdg, cash);
+    }
 }

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java

@@ -5,8 +5,11 @@ import java.math.RoundingMode;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
@@ -44,6 +47,7 @@ import ru.spcex.platform.enumeration.Priority;
 import ru.spcex.platform.enumeration.Section;
 import ru.spcex.platform.enumeration.Side;
 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;
@@ -60,7 +64,7 @@ import ru.spcex.platform.utils.validation.IValidator;
 
 @Component
 @EnableScheduling
-public class ExecutionFondComponent {
+public class ExecutionFondComponent implements IExecutionUploadComponent {
     private final Logger log = LoggerFactory.getLogger(getClass());
 
     private final Imdg<STrades> sTradeImdg;
@@ -70,6 +74,7 @@ public class ExecutionFondComponent {
     private final Imdg<FixedIncomeCashFlow> fixedIncomeCashFlowImdg;
     private final Imdg<Market> marketImdg;
     private final Imdg<SCrossRate> crossRateImdg;
+    private final ImdgId idGen;
 
     //fixme ждать ТЗ
     Long tradeNum;
@@ -79,6 +84,7 @@ public class ExecutionFondComponent {
     private final KafkaSender kafkaSender;
     private final NotificationSender notifications;
     private final boolean valuation;
+    private final Set<ExecUploadKey> cash = new HashSet<>();
 
 
     @Autowired
@@ -94,6 +100,7 @@ public class ExecutionFondComponent {
         this.marketImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Market, Market.class);
         this.fixedIncomeCashFlowImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_FixedIncomeCashFlow, FixedIncomeCashFlow.class);
         this.crossRateImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SCrossRate, SCrossRate.class);
+        this.idGen = imdgProvider.getImdgIdGenerator();
         this.valuation = settings.getTrade().getValuation(); //fixme
         this.stradesValidator = stradesValidator;
         this.kafkaSender = kafkaSender;
@@ -149,13 +156,17 @@ public class ExecutionFondComponent {
                 return true;
             }
             Side excDepSide = sTrdSide.equals(Side.BUY) ? Side.BUY : Side.SELL;
-            return executionFondImdg.getFirstObjectByFieldValues(
+            return cash.contains(
-                    Map.of("tradingDate", sTrd.getTradeDate(),
+                ExecUploadKey.cash(sTrd.getTradeDate(), sTrd.getTradeNum(), excDepSide.getKey())
-                            "exchangeExecutionId", sTrd.getTradeNum(),
+            );
-                            "side", excDepSide.getKey())) != null;
+            //return executionFondImdg.getFirstObjectByFieldValues(
+            //        Map.of("tradingDate", sTrd.getTradeDate(),
+            //                "exchangeExecutionId", sTrd.getTradeNum(),
+            //                "side", excDepSide.getKey())) != null;
         });
         log.info("{} strades left after already-added filtering", sTrades.size());
 
+        Map<Long, ExecutionFond> execsToInsert = new HashMap<>();
         for (STrades sTrd : sTrades) {
             log.info("new S_TRADE[{}], valuation {}", sTrd.getId(), valuation);
 
@@ -169,7 +180,10 @@ public class ExecutionFondComponent {
             ExecutionFond newED;
             try {
                 newED = createExecutionFond(sTrd, validator);
-                executionFondImdg.insert(newED);
+                newED.setId(idGen.nextId());
+                cash.add(ExecUploadKey.cash(newED));
+                execsToInsert.put(newED.getId(), newED);
+                //executionFondImdg.insert(newED);
                 sendNotification(newED);
                 log.debug("New executionDeposit.id={} was created.", newED.getId());
             } catch (ClearingException ce) {
@@ -178,6 +192,7 @@ public class ExecutionFondComponent {
                 log.error("When create new ExecutionDeposit by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e));
             }
         }
+        executionFondImdg.putAll(execsToInsert, 200);
 
         Long newMaxTradeNum = sTrades.stream().mapToLong(STrades::getTradeNum).max().orElseGet(() -> tradeNum);
         log.info("Process completed. Next tradeNum is {}", newMaxTradeNum);
@@ -340,4 +355,9 @@ public class ExecutionFondComponent {
     private void logError(Long sTradeId, EnumMessage msg) {
         log.warn("sTrade id={} {}", sTradeId, msgResolver.resolve(msg));
     }
+
+    @Override
+    public void initExecCash() {
+        ExecutionUploadCashUtil.loadExecutions(executionFondImdg, cash);
+    }
 }

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionUploadCashUtil.java

@@ -0,0 +1,21 @@
+package ru.spcex.clearing.service.execution;
+
+import java.time.LocalDate;
+import java.util.Collection;
+import java.util.Set;
+import ru.clearing.classes.statics.data.execution.ExecutionCommon;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
+
+public class ExecutionUploadCashUtil {
+    public static <T extends ExecutionCommon>
+    void loadExecutions(Imdg<T> executionImdg, Set<ExecUploadKey> cash) {
+        LocalDate today = LocalDate.now();
+        ImdgPredicateBuilder pb = executionImdg.predicateBuilder();
+        Collection<T> execs = executionImdg.getCollectionObjectsByPredicate(
+            pb.equals("tradingDate", today)
+        );
+        execs.forEach(e -> cash.add(ExecUploadKey.cash(e)));
+    }
+
+}

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/IExecutionUploadComponent.java

@@ -0,0 +1,5 @@
+package ru.spcex.clearing.service.execution;
+
+public interface IExecutionUploadComponent {
+    void initExecCash();
+}