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

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

@@ -0,0 +1,101 @@
+package ru.spcex.clearing.session.stage;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Service;
+import ru.clearing.classes.statics.data.misc.Session;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
+import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
+import ru.spcex.clearing.session.stage.impl.*;
+import ru.spcex.platform.enumeration.Section;
+import ru.spcex.platform.enumeration.SessionStatus;
+import ru.spcex.platform.enumeration.SessionType;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.utils.enumeration.IMessageResolver;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@Service
+public class PrimaryAuctionBnSession extends QueueConsumer {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final BalanceRevise balanceRevise;
+    private final DealsPrepare dealsPrepare;
+    private final RequirementsAndObligationCreation requirementsAndObligationCreation;
+    private final ObligationAdmission obligationsAdmission;
+
+    private final InclusionObligations inclusionObligations;
+    private final FormingRegistersOnOS formingRegistersOnOS;
+    private final FormingPaymentInstruction formingPaymentInstruction;
+    private final UnlockResources unlockResources;
+    private final EndStageNotification endStageNotification;
+
+    //
+    private final AtomicBoolean running;
+    private Long sessionId;
+    private Imdg<Session> sessionImdg;
+    private final IMessageResolver messageResolver;
+
+    public PrimaryAuctionBnSession(
+            @Qualifier("createConsumer") Consumer<String, Object> kafkaQueue,
+            ImdgProvider imdgProvider,
+            BalanceRevise balanceRevise,
+            DealsPrepare dealsPrepare,
+            RequirementsAndObligationCreation requirementsAndObligationCreation,
+            ObligationAdmission obligationsAdmission,
+            InclusionObligations inclusionObligations,
+            FormingRegistersOnOS formingRegistersOnOS,
+            FormingPaymentInstruction formingPaymentInstruction,
+            UnlockResources unlockResources,
+            EndStageNotification endStageNotification, IMessageResolver messageResolver) {
+        super(kafkaQueue);
+        this.balanceRevise = balanceRevise;
+        this.dealsPrepare = dealsPrepare;
+        this.requirementsAndObligationCreation = requirementsAndObligationCreation;
+        this.obligationsAdmission = obligationsAdmission;
+        this.inclusionObligations = inclusionObligations;
+        this.formingRegistersOnOS = formingRegistersOnOS;
+        this.formingPaymentInstruction = formingPaymentInstruction;
+        this.unlockResources = unlockResources;
+        this.endStageNotification = endStageNotification;
+        //
+        this.sessionImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Session, Session.class);
+        this.messageResolver = messageResolver;
+        this.running = new AtomicBoolean(false);
+    }
+
+    public void runSession() {
+        synchronized (this.running) {
+            if (this.running.get()) {
+                log.info("already running session.id={}", this.sessionId);
+                return;
+            } else {
+                this.running.set(true);
+            }
+        }
+        Session newSession = new Session();
+        newSession.setSection(Section.FOND.getKey());
+        newSession.setSessionType(SessionType.IPOB.getKey());
+        newSession.setSessionStatus(SessionStatus.CLRN.getKey());
+        //todo  companyId/securityId/userId передается из сообщения очереди
+        sessionImdg.insert(newSession);
+        sessionId = newSession.getId();
+        //java.util.function.Consumer<StageResult<?>> logError = stageResult -> {
+        //    if (!stageResult.success) {
+        //        log.error("stageResult.error={}", messageResolver.resolve(stageResult.error));
+        //        newSession.setSessionStatus(SessionStatus.CLOS.getKey());
+        //    }
+        //};
+        //{
+        //    StageResult<?> reviseResult = balanceRevise.submit(task(TaskType.StartRevise));
+        //    logError.accept(reviseResult);
+        //}
+
+    }
+
+    private Task<?> task(TaskType taskType) {
+        return new Task<>(taskType, null);
+    }
+}

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

@@ -62,8 +62,8 @@ public class BalanceRevise implements ISessionStage {
                 cashFlow(); //((SdfClearingRequest) task.getData()).getGroupId() if needed
                 return revise();
             }
+            default -> throw new IllegalStateException("unknown task " + task.getTaskType());
         }
-        return null;
     }
 
 

platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SessionType.java

@@ -0,0 +1,24 @@
+package ru.spcex.platform.enumeration;
+
+import ru.spcex.platform.utils.enumeration.IEnumKey;
+
+public enum SessionType implements IEnumKey {
+    IPOB("IPOB"),
+    ;
+
+    SessionType(String key) {
+        this.key = key;
+    }
+
+    private String key;
+
+    @Override
+    public String getKey() {
+        return this.key;
+    }
+
+    @Override
+    public boolean equalsByKey(String key) {
+        return IEnumKey.super.equalsByKey(key);
+    }
+}