http://jira.mfd.msk:8088/browse/CLS-862

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

@@ -9,9 +9,11 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import static java.lang.String.format;
 import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.scheduling.annotation.EnableScheduling;
@@ -27,6 +29,7 @@ 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.notification.NotificationSender;
 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.registry.ExecutionType;
@@ -36,6 +39,8 @@ import ru.spcex.clearing.service.validation.strades.CompanyByTradingCodeCashingV
 import ru.spcex.clearing.service.validation.strades.STradesSecurityPresentDepositValidationRule;
 import ru.spcex.clearing.service.validation.strades.TcrByCodeAndCmpIsActiveCashingValidationRule;
 import ru.spcex.platform.enumeration.MoneyFlowSide;
+import ru.spcex.platform.enumeration.ObjectType;
+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;
@@ -73,6 +78,7 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
     private final Imdg<Company> imdgCompany;
     private final Imdg<TradingClearingRegistry> imdgTradingClearingRegistry;
     private final ImdgId idGen;
+    private final NotificationSender notifications;
 
     //fixme ждать ТЗ
     Long tradeNum;
@@ -86,7 +92,7 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
 
 
     @Autowired
-    public ExecutionDepositComponent(ImdgProvider imdgProvider, Producer<String, Object> kafka,
+    public ExecutionDepositComponent(ImdgProvider imdgProvider, Producer<String, Object> kafka, NotificationSender notifications,
                                      @Qualifier("kafkaSenderWithoutRequestInfo") KafkaSender kafkaSender,
                                      IMessageResolver msgResolver) {
         this.sTradeImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class);
@@ -96,6 +102,7 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
         this.imdgCompany = imdgProvider.getCashingImdg(IMDGDistributedNames.Map_Company, Company.class, null);
         this.imdgTradingClearingRegistry = imdgProvider.getCashingImdg(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class, null);
         this.idGen = imdgProvider.getImdgIdGenerator();
+        this.notifications = notifications;
         this.kafkaSender = kafkaSender;
         this.msgResolver = msgResolver;
 
@@ -154,7 +161,11 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
                 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());
+                        logNotifyW(format(
+                            "cannot parse strade.tradeNum=%d strade.operation %s",
+                            sTrd.getTradeNum(),
+                            sTrd.getOperation())
+                        );
                         return true;
                     }
                     MoneyFlowSide excDepSide = sTrdSide.equals(Side.BUY) ? MoneyFlowSide.BUY : MoneyFlowSide.SELL;
@@ -198,7 +209,10 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
                 } catch (ClearingException ce) {
                     auditMessage(ce);
                 } catch (Exception e) {
-                    log.error("When create new ExecutionDeposit by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e));
+                    String err = String.format("When create new ExecutionDeposit by STrade[%d] error: %s",
+                        sTrd.getId(),
+                        ExceptionUtils.getStackTrace(e));
+                    logNotifyE(err);
                 }
             }
             log.info("batch putAll {} ExecutionDeposit", execsToInsert.size());
@@ -211,7 +225,8 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
 
 
     protected void auditMessage(ClearingException ce) {
-        log.error("AUDIT error code {}: {}", ce.getEnumMsg(), ce.getMessage());
+        String err = String.format("AUDIT error code %s: %s", ce.getEnumMsg(), ce.getMessage());
+        logNotifyE(err);
     }
 
     protected void sendNotification(ExecutionDeposit forED) throws ClearingException {
@@ -330,11 +345,33 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
     }
 
     private void logError(IEnumId subject, Object... args) {
-        log.info("{}", msgResolver.resolve(new EnumMessage(subject, args)));
+        String err = msgResolver.resolve(new EnumMessage(subject, args));
+        log.info("{}", err);
+        notifications.sendNotification(ObjectType.vfrs, err, Priority.HIGH);
     }
 
+
     private void logError(Long sTradeId, EnumMessage msg) {
-        log.warn("sTrade id={} {}", sTradeId, msgResolver.resolve(msg));
+        String err = format("sTrade id=%d %s", sTradeId, msgResolver.resolve(msg));
+        log.warn(err);
+        notifications.sendNotification(ObjectType.vfrs, err, Priority.HIGH);
+    }
+
+    private void logNotifyW(String msg) {
+        logNotify(msg, Level.WARN);
+    }
+
+    private void logNotifyE(String msg) {
+        logNotify(msg, Level.ERROR);
+    }
+
+    private void logNotify(String msg, Level lvl) {
+        switch (lvl) {
+            case WARN -> log.warn(msg);
+            case ERROR -> log.error(msg);
+            default -> log.error("lvl not supported {}. {}", lvl, msg);
+        }
+        notifications.sendNotification(ObjectType.vfrs, msg, Priority.HIGH);
     }
 
     private Optional<TradingClearingRegistry> searchTcrByStrades(STrades sTrades) {

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

@@ -9,8 +9,10 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import static java.lang.String.format;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.scheduling.annotation.EnableScheduling;
@@ -183,7 +185,11 @@ public class ExecutionFondComponent implements IExecutionUploadComponent {
                 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());
+                        logNotifyW(format(
+                            "cannot parse strade.tradeNum=%d strade.operation %s",
+                            sTrd.getTradeNum(),
+                            sTrd.getOperation())
+                        );
                         return true;
                     }
                     Side excDepSide = sTrdSide.equals(Side.BUY) ? Side.BUY : Side.SELL;
@@ -227,7 +233,10 @@ public class ExecutionFondComponent implements IExecutionUploadComponent {
                 } catch (ClearingException ce) {
                     auditMessage(ce);
                 } catch (Exception e) {
-                    log.error("When create new ExecutionFond by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e));
+                    String err = String.format("When create new ExecutionFond by STrade[%d] error: %s",
+                        sTrd.getId(),
+                        ExceptionUtils.getStackTrace(e));
+                    logNotifyE(err);
                 }
             }
             log.info("batch putAll {} ExecutionFond", execsToInsert.size());
@@ -240,7 +249,8 @@ public class ExecutionFondComponent implements IExecutionUploadComponent {
 
 
     protected void auditMessage(ClearingException ce) {
-        log.error("AUDIT error code {}: {}", ce.getEnumMsg(), ce.getMessage());
+        String err = String.format("AUDIT error code %s: %s", ce.getEnumMsg(), ce.getMessage());
+        logNotifyE(err);
     }
 
     protected void sendNotification(ExecutionFond forEF) throws ClearingException {
@@ -389,11 +399,32 @@ public class ExecutionFondComponent implements IExecutionUploadComponent {
     }
 
     private void logError(IEnumId subject, Object... args) {
-        log.info("{}", msgResolver.resolve(new EnumMessage(subject, args)));
+        String err = msgResolver.resolve(new EnumMessage(subject, args));
+        log.info("{}", err);
+        notifications.sendNotification(ObjectType.vfrs, err, Priority.HIGH);
     }
 
     private void logError(Long sTradeId, EnumMessage msg) {
-        log.warn("sTrade id={} {}", sTradeId, msgResolver.resolve(msg));
+        String err = format("sTrade id=%d %s", sTradeId, msgResolver.resolve(msg));
+        log.warn(err);
+        notifications.sendNotification(ObjectType.vfrs, err, Priority.HIGH);
+    }
+
+    private void logNotifyW(String msg) {
+        logNotify(msg, Level.WARN);
+    }
+
+    private void logNotifyE(String msg) {
+        logNotify(msg, Level.ERROR);
+    }
+
+    private void logNotify(String msg, Level lvl) {
+        switch (lvl) {
+            case WARN -> log.warn(msg);
+            case ERROR -> log.error(msg);
+            default -> log.error("lvl not supported {}. {}", lvl, msg);
+        }
+        notifications.sendNotification(ObjectType.vfrs, msg, Priority.HIGH);
     }
 
     @Override