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

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

@@ -1,5 +1,6 @@
 package ru.spcex.clearing.service.execution;
 
+import static java.lang.String.format;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
 import java.time.Instant;
@@ -13,6 +14,7 @@ import java.util.Optional;
 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;
@@ -31,6 +33,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;
@@ -39,6 +42,8 @@ import ru.spcex.clearing.service.validation.ValidationStored;
 import ru.spcex.clearing.service.validation.strades.CompanyByTradingCodeCashingValidationRule;
 import ru.spcex.clearing.service.validation.strades.STradesSecurityPresentValidationRule;
 import ru.spcex.clearing.service.validation.strades.TcrByCodeAndCmpIsActiveCashingValidationRule;
+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;
@@ -88,13 +93,14 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent {
     private final KafkaSender kafkaSender;
     private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy");
     private final Map<ExecUploadKey, ExecUploadCashInfo> cash = new HashMap<>();
-
+    private final NotificationSender notifications;
 
 
     @Autowired
     public ExecutionCurrencyComponent(ImdgProvider imdgProvider, Producer<String, Object> kafka,
                                       @Qualifier("kafkaSenderWithoutRequestInfo") KafkaSender kafkaSender,
-                                      IMessageResolver msgResolver) {
+                                      IMessageResolver msgResolver,
+                                      NotificationSender notifications) {
         this.sTradeImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class);
         this.executionCurrencyImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ExecutionCurrency, ExecutionCurrency.class);
         this.listingImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Listing, Listing.class);
@@ -110,6 +116,7 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent {
 
 //        resetTradingDay();
         this.idGen = imdgProvider.getImdgIdGenerator();
+        this.notifications = notifications;
     }
 
     @Override
@@ -170,7 +177,11 @@ public class ExecutionCurrencyComponent 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 == Side.BUY ? MoneyFlowSide.BUY : MoneyFlowSide.SELL;
@@ -217,7 +228,10 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent {
                 } catch (ClearingException ce) {
                     auditMessage(ce);
                 } catch (Exception e) {
-                    log.error("When create new ExecutionCurrency by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e));
+                    String err = String.format("When create new ExecutionCurrency by STrade[%d] error: %s",
+                        sTrd.getId(),
+                        ExceptionUtils.getStackTrace(e));
+                    logNotifyE(err);
                 }
             }
             log.info("batch putAll {} ExecutionCurrency", execsToInsert.size());
@@ -230,7 +244,8 @@ public class ExecutionCurrencyComponent 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(ExecutionCurrency forED) throws ClearingException {
@@ -348,11 +363,32 @@ public class ExecutionCurrencyComponent 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) {