clearing-service http://jira.mfd.msk:8088/browse/CLS-290 stage 11

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

@@ -54,7 +54,7 @@ public class ExecutionDepositComponent {
     //fixme ждать ТЗ
     Long tradeNum;
     Instant tradingDay;
-    private final IMessageResolver msgResolver = new SimpleMessageResolver();
+    private final IMessageResolver msgResolver;
     private final Function<STrades, IValidator> stradesValidator;
     private final KafkaSender kafkaSender;
 
@@ -62,12 +62,14 @@ public class ExecutionDepositComponent {
     @Autowired
     public ExecutionDepositComponent(ImdgProvider imdgProvider, Producer<String, Object> kafka,
                                      @Qualifier("sTradesValidator") Function<STrades, IValidator> stradesValidator,
-                                     @Qualifier("kafkaSenderWithoutRequestInfo") KafkaSender kafkaSender) {
+                                     @Qualifier("kafkaSenderWithoutRequestInfo") KafkaSender kafkaSender,
+                                     IMessageResolver msgResolver) {
         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.stradesValidator = stradesValidator;
         this.kafkaSender = kafkaSender;
+        this.msgResolver = msgResolver;
 
         resetTradingDay();
     }
@@ -127,7 +129,7 @@ public class ExecutionDepositComponent {
             }
             ExecutionDeposit newED;
             try {
-                newED = createExecutionDeposit(sTrd,  validator);
+                newED = createExecutionDeposit(sTrd, validator);
                 executionDepositImdg.insert(newED);
                 sendNotification(newED);
                 log.debug("New executionDeposit.id={} was created.", newED.getId());

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/TaskType.java

@@ -34,4 +34,11 @@ public enum TaskType {
      * step 8
      */
     UnlockResources,
+
+
+
+    /**
+     * Step 11
+     */
+    EndStageNotification;
 }

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/BalanceRevise.java

@@ -43,12 +43,13 @@ public class BalanceRevise implements ISessionStage {
     private KafkaSender kafkaSender;
 
     @Autowired
-    public BalanceRevise(ImdgProvider imdgProvider) {
+    public BalanceRevise(ImdgProvider imdgProvider, KafkaSender kafkaSender) {
         this.imdgProvider = imdgProvider;
         this.idGenerator = imdgProvider.getImdgIdGenerator();
         this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class);
         this.currencyImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Currency, Currency.class);
         this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class);
+        this.kafkaSender = kafkaSender;
     }
 
     @Override

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/EndStageNotification.java

@@ -0,0 +1,112 @@
+package ru.spcex.clearing.session.stage.impl;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import ru.clearing.classes.statics.data.registry.Registry;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
+import ru.spcex.clearing.platform.messaging.domain.Consts;
+import ru.spcex.clearing.platform.messaging.domain.cud.clearing.SdfClearingRequest;
+import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
+import ru.spcex.clearing.session.stage.ISessionStage;
+import ru.spcex.clearing.session.stage.StageResult;
+import ru.spcex.clearing.session.stage.Task;
+import ru.spcex.clearing.session.stage.task.EndStageNotificationPayload;
+import ru.spcex.platform.enumeration.*;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.utils.enumeration.EnumMessage;
+import ru.spcex.platform.utils.enumeration.IMessageResolver;
+
+import java.util.Collection;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import static ru.spcex.clearing.error.ClearingErrorInternal.SessionGeneralError;
+
+@Service
+public class EndStageNotification implements ISessionStage {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private final Imdg<Registry> registryImdg;
+    private final KafkaSender kafkaSender;
+    private final IMessageResolver msgResolver;
+
+    @Autowired
+    public EndStageNotification(ImdgProvider imdgProvider, KafkaSender kafkaSender, IMessageResolver msgResolver) {
+        this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class);
+        this.kafkaSender = kafkaSender;
+        this.msgResolver = msgResolver;
+    }
+
+    @Override
+    public StageResult<?> submit(Task<?> task) {
+        EndStageNotificationPayload payload = (EndStageNotificationPayload) task.getData();
+        switch (task.getTaskType()) {
+            case EndStageNotification -> {
+                return endStageNotification(payload.getSection());
+            }
+            default -> {
+                throw new IllegalStateException("Unknown task type: " + task.getTaskType());
+            }
+        }
+    }
+
+    protected Collection<Registry> selectRegistry() {
+        String registrySQL = "registryStatus=" + RegistryStatus.OK.getKey() + ")";
+        Collection<Registry> result = registryImdg.getCollectionObjectsBySQL(registrySQL);
+        log.trace("Selected {} registry's by sql: {}", result.size(), registrySQL);
+        return result;
+    }
+
+    protected StageResult<?> endStageNotification(String section) {
+        Collection<Registry> forRegistries = selectRegistry();
+        Collection<Long> groups = forRegistries.stream()
+                .map(Registry::getGroupId)
+                .filter(Objects::nonNull)
+                .distinct().collect(Collectors.toList());
+        log.debug("Sending notifications for {} groups ({} registers) on section {}",
+                groups, forRegistries.size(), section);
+        for (Long groupId : groups) {
+            log.trace("For registry group {} send notification",
+                    groupId);
+            StageResult sResult;
+            if (Section.FOND.equalsByKey(section)) {
+                sResult = notificationDF14(groupId);
+            } else if (Section.MKR.equalsByKey(section)) {
+                sResult = notificationDF05(groupId);
+            } else {
+                throw new IllegalArgumentException("Unsupported section " + section);
+            }
+            if (sResult.getError() != null) {
+                log.warn("When sending groupId={} has error: {}", groupId, msgResolver.resolve(sResult.getError()));
+            }
+        }
+        StageResult<Collection<Registry>> res = new StageResult<>(null, true);
+        return res;
+    }
+
+    protected StageResult notificationDF14(Long groupId) {
+        SdfClearingRequest sdf14Request = new SdfClearingRequest();
+        sdf14Request.setGroupId(groupId);
+        Long msgKey = kafkaSender.sendRequestToQueue(Consts.SDF14_PROCESS, sdf14Request);
+        if (msgKey == null) {
+            log.error("failed to put SDF14 request to kafka queue");
+            return new StageResult<>(new EnumMessage(SessionGeneralError), false);
+        }
+        return new StageResult<>(null, true);
+    }
+
+    protected StageResult notificationDF05(Long groupId) {
+        SdfClearingRequest sdf05Request = new SdfClearingRequest();
+        sdf05Request.setGroupId(groupId);
+        Long msgKey = kafkaSender.sendRequestToQueue(Consts.SDF05_PROCESS, sdf05Request);
+        if (msgKey == null) {
+            log.error("failed to put SDF05 request to kafka queue");
+            return new StageResult<>(new EnumMessage(SessionGeneralError), false);
+        }
+        return new StageResult<>(null, true);
+    }
+
+}

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/task/EndStageNotificationPayload.java

@@ -0,0 +1,13 @@
+package ru.spcex.clearing.session.stage.task;
+
+public class EndStageNotificationPayload {
+    private String section;
+
+    public String getSection() {
+        return section;
+    }
+
+    public void setSection(String section) {
+        this.section = section;
+    }
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java

@@ -111,6 +111,7 @@ public interface Consts {
     String SDF05_PROCESS = "sdf05-process";
     String SDF07_PROCESS = "sdf07-process";
     String SDF11_PROCESS = "sdf11-process";
+    String SDF14_PROCESS = "sdf14-process";
     String SDF51_PROCESS = "sdf51-process";
     String SDF53_PROCESS = "sdf53-process";
     String SDF54_PROCESS = "sdf54-process";