clearing-service Sdf04 ожидание полностью готовой группы в SessionMonitor

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

@@ -15,12 +15,15 @@ import ru.spcex.clearing.platform.messaging.domain.cud.account.sdf01.AccountSdf0
 import ru.spcex.clearing.platform.messaging.domain.cud.account.sdf01.AccountSdfRequestPart;
 import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.clearing.SessionContinueEvent;
+import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationNewRequest;
 import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
 import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
 import ru.spcex.clearing.service.executors.AbstractExecutor;
 import ru.spcex.clearing.service.executors.Reviser;
 import ru.spcex.clearing.service.model.Result;
 import ru.spcex.platform.classes.base.SpcexObjectBase;
+import ru.spcex.platform.enumeration.ObjectType;
+import ru.spcex.platform.enumeration.Priority;
 import ru.spcex.platform.enumeration.SdfTable;
 import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgProvider;
@@ -236,7 +239,15 @@ public class StatementService extends QueueConsumer implements InitializingBean
         if (service != null) {
             Result res = service.execute(sdfGroup, statementRequest);
             SessionContinueEvent continueSessionBn = new SessionContinueEvent(SdfTable.SDF_04);
-            kafkaSender.sendRequestToQueue(Consts.CONTINUE_SESSION_BN_FIRST_PART, continueSessionBn);
+            if (res.isAnyHasError()) {
+                NotificationNewRequest nRequest = new NotificationNewRequest();
+                nRequest.setObjectType(ObjectType.rgst.getKey());
+                nRequest.setPriority(Priority.HIGH.getKey());
+                nRequest.setComment(String.format("Ошибка : некоторые из новых SDF_04 имеют неправильный статус"));
+                kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, nRequest);
+            } else { // отправить в SessionMonitor
+                kafkaSender.sendRequestToQueue(Consts.CONTINUE_SESSION_BN_FIRST_PART, continueSessionBn);
+            }
         } else {
             log.warn("Executor for SDF_04 not set");
         }
@@ -272,6 +283,15 @@ public class StatementService extends QueueConsumer implements InitializingBean
         AbstractExecutor service = executorsMap.get(SdfTable.SDF_13);
         Result res = service.execute(sdfGroup, statementRequest);
         SessionContinueEvent continueSessionBn = new SessionContinueEvent(SdfTable.SDF_13);
+//        if (res.isAnyHasError()) { todo
+//            NotificationNewRequest nRequest = new NotificationNewRequest();
+//            nRequest.setObjectType(ObjectType.rgst.getKey());
+//            nRequest.setPriority(Priority.HIGH.getKey());
+//            nRequest.setComment(String.format("Ошибка : некоторые из новых SDF_13 имеют неправильный статус"));
+//            kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, nRequest);
+//        } else {
+//            kafkaSender.sendRequestToQueue(Consts.CONTINUE_SESSION_BN_FIRST_PART, continueSessionBn);
+//        }
         kafkaSender.sendRequestToQueue(Consts.CONTINUE_SESSION_BN_FIRST_PART, continueSessionBn);
 //        finishSendCommand(res, service, statementRequest);
     }

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf04Executor.java

@@ -71,7 +71,12 @@ public class Sdf04Executor extends AbstractExecutor<SDf04> {
         Result result = new Result();
         Long generationIdForGroup = imdgProvider.getImdgIdGenerator().nextId();
         result.setChildGenerationId(generationIdForGroup);
+        result.setAnyHasError(false);
         for (SDf04 sdf04 : sdf) {
+            if (!("OK!".equals(sdf04.getImp_result()))) {
+                log.trace("sdf04[{}].Imp_result=\"{}\" was not ok.", sdf04.getId(), sdf04.getImp_result());
+                result.setAnyHasError(true);
+            }
             //обычно мы ищем по группу sdf04, здесь как будто всегда только одна запись, todo нужно прочекать этот момент
             log.debug("Process sdf04 record; sdf04.id: {}", sdf04.getId());
             Collection<Registry> registries;

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/model/Result.java

@@ -9,6 +9,7 @@ public class Result {
     private List<AccountSdfRequestPart> accountRequests = new ArrayList<>();
     private Long childGenerationId;
     private String fileName;
+    private boolean anyHasError;
 
     public List<AccountSdfRequestPart> getAccountRequests() {
         return accountRequests;
@@ -33,4 +34,12 @@ public class Result {
     public void setFileName(String fileName) {
         this.fileName = fileName;
     }
+
+    public boolean isAnyHasError() {
+        return anyHasError;
+    }
+
+    public void setAnyHasError(boolean anyHasError) {
+        this.anyHasError = anyHasError;
+    }
 }

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

@@ -3,6 +3,9 @@ package ru.spcex.clearing.session.stage.monitor;
 import ru.spcex.platform.enumeration.SdfTable;
 import ru.spcex.platform.enumeration.Section;
 
+/**
+ * @see ru.spcex.clearing.platform.messaging.domain.cud.clearing.SessionContinueEvent
+ */
 public class SessionMonitorFactory {
     public static SessionMonitor paymentsWereCreated(Section section) {
         switch (section) {