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

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/builder/sql/RegistryCodeSqlBuilder.java

@@ -1,6 +1,8 @@
 package ru.spcex.clearing.service.builder.sql;
 
 import ru.spcex.clearing.models.RegistryTradingParams;
+import ru.spcex.platform.imdg.api.predicate.ImdgPredicate;
+import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -55,4 +57,26 @@ public class RegistryCodeSqlBuilder {
         }
         return conditions.stream().collect(Collectors.joining(") or (", "(", ")"));
     }
+
+
+    public ImdgPredicate buildPredicate(ImdgPredicateBuilder pb) {
+        List<ImdgPredicate> conditions = new ArrayList<>();
+        for (RegistryTradingParams tradingParams : this.registryTradingParams) {
+            if (tradingParams.registryDesignation() != null) {
+                conditions.add(pb.equals("registryDesignation", tradingParams.registryDesignation().getKey()));
+            }
+            if (tradingParams.registryInstrumentType() != null) {
+                conditions.add(pb.equals("registryInstrumentType", tradingParams.registryInstrumentType().getKey()));
+            }
+            if (tradingParams.registryCapacity() != null) {
+                conditions.add(pb.equals("registryCapacity", tradingParams.registryCapacity().getKey()));
+            }
+            if (tradingParams.registryUnit() != null) {
+                conditions.add(pb.equals("registryUnit", tradingParams.registryUnit().getKey()));
+            }
+        }
+        if (conditions.isEmpty()) return null;
+        if (conditions.size() == 1) return conditions.get(0);
+        return pb.and(conditions.toArray(new ImdgPredicate[conditions.size()]));
+    }
 }

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

@@ -0,0 +1,142 @@
+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.misc.Session;
+import ru.clearing.classes.statics.data.registry.Registry;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
+import ru.spcex.clearing.models.RegistryTradingParams;
+import ru.spcex.clearing.platform.messaging.domain.Consts;
+import ru.spcex.clearing.platform.messaging.domain.cud.reports.ReportRequestWithRegistryId;
+import ru.spcex.clearing.platform.messaging.domain.cud.reports.ReportRequestWithSessionId;
+import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
+import ru.spcex.clearing.service.builder.sql.RegistryCodeSqlBuilder;
+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.FinishingSessionPayload;
+import ru.spcex.platform.enumeration.RegistryDesignation;
+import ru.spcex.platform.enumeration.SessionStatus;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.imdg.api.predicate.ImdgPredicate;
+import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
+import ru.spcex.platform.utils.enumeration.EnumMessage;
+import ru.spcex.platform.utils.enumeration.IMessageResolver;
+
+import java.time.Instant;
+import java.util.Collection;
+
+import static ru.spcex.clearing.error.ClearingErrorInternal.SessionGeneralError;
+
+@Service
+public class FinishingSession implements ISessionStage {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private final Imdg<Registry> registryImdg;
+    private final Imdg<Session> sessionImdg;
+    private final KafkaSender kafkaSender;
+    private final IMessageResolver msgResolver;
+
+    @Autowired
+    public FinishingSession(ImdgProvider imdgProvider, KafkaSender kafkaSender, IMessageResolver msgResolver) {
+        this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class);
+        this.sessionImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Session, Session.class);
+        this.kafkaSender = kafkaSender;
+        this.msgResolver = msgResolver;
+    }
+
+    @Override
+    public StageResult<?> submit(Task<?> task) {
+        FinishingSessionPayload payload = (FinishingSessionPayload) task.getData();
+        switch (task.getTaskType()) {
+            case FinishingSession -> {
+                return finishingSession(payload.getSessionId());
+            }
+            default -> {
+                throw new IllegalStateException("Unknown task type: " + task.getTaskType());
+            }
+        }
+    }
+
+    protected Collection<Registry> selectRegistry(Long sessionId) {
+        RegistryCodeSqlBuilder registryCodeSqlBuilder = RegistryCodeSqlBuilder.getInstance(
+                new RegistryTradingParams(RegistryDesignation.A, null, null, null)
+        );
+        ImdgPredicateBuilder pb = registryImdg.predicateBuilder();
+        ImdgPredicate condition = registryCodeSqlBuilder.buildPredicate(pb);
+        condition = pb.and(condition, pb.equals("sessionId", sessionId));
+        Collection<Registry> result = registryImdg.getCollectionObjectsByPredicate(condition);
+        log.trace("Selected {} registry's by sql: {}", result.size(), condition);
+        return result;
+    }
+
+    protected StageResult<?> finishingSession(Long sessionId) {
+        // 1. Изменить статус
+        Session theSession = sessionImdg.getSingleObjectByID(sessionId);
+        if (theSession == null) {
+            log.warn("Session {} not found", sessionId);
+        } else {
+            log.info("Finish status for session {}", sessionId);
+            theSession.setUpdated(Instant.now());
+            theSession.setSessionStatus(SessionStatus.CLOS.getKey());
+            sessionImdg.update(theSession);
+            log.trace("Session {} was updated", theSession.getId());
+        }
+
+        // 2. Отправка сообщений
+        Collection<Registry> forRegistries = selectRegistry(sessionId);
+        log.debug("Found {} registries for sessionId={}", forRegistries.size(), sessionId);
+
+        StageResult sResult = toReportSession(sessionId);
+        if (sResult.getError() != null) {
+            log.error("When sending sessionId={} has error: {}", sessionId, msgResolver.resolve(sResult.getError()));
+            return sResult;
+        }
+        for (Registry registryA : forRegistries) {
+            if (!RegistryDesignation.A.equalsByKey(registryA.getRegistryUnit())) {
+                log.warn("For registry {}.RegistryDesignation is not A.", registryA.getId());
+                continue;
+            }
+            sResult = toReportMoney(sessionId, registryA);
+            if (sResult.getError() != null) {
+                log.warn("When sending sessionId={}, registry.id={} has error: {}", sessionId, registryA.getId(), msgResolver.resolve(sResult.getError()));
+                return sResult;
+            }
+        }
+        StageResult<Collection<Registry>> res = new StageResult<>(null, true);
+        return res;
+    }
+
+    /**
+     * формирование операционного отчета об обязательствах
+     **/
+    protected StageResult toReportSession(Long sessionId) {
+        ReportRequestWithSessionId reportSessionRequest = new ReportRequestWithSessionId();
+        reportSessionRequest.setSessionId(sessionId);
+        Long msgKey = kafkaSender.sendRequestToQueue(Consts.CREATE_REPORT_FOR_SESSION_ID, reportSessionRequest);
+        if (msgKey == null) {
+            log.error("failed to put Report Session request to kafka queue");
+            return new StageResult<>(new EnumMessage(SessionGeneralError), false);
+        }
+        return new StageResult<>(null, true);
+    }
+
+    /**
+     * формирование операционного отчета о денежных средствах
+     **/
+    protected StageResult toReportMoney(Long sessionId, Registry registry) {
+        ReportRequestWithRegistryId reportMoneyRequest = new ReportRequestWithRegistryId();
+        reportMoneyRequest.setSessionId(sessionId);
+        reportMoneyRequest.setRegistryId(registry.getId());
+        Long msgKey = kafkaSender.sendRequestToQueue(Consts.CREATE_REPORT_FOR_REGISTRY, reportMoneyRequest);
+        if (msgKey == null) {
+            log.error("failed to put Report Registry request to kafka queue");
+            return new StageResult<>(new EnumMessage(SessionGeneralError), false);
+        }
+        return new StageResult<>(null, true);
+    }
+
+}

platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/predicate/ImdgPredicateBuilderHazelcast.java

@@ -58,14 +58,6 @@ public class ImdgPredicateBuilderHazelcast implements ImdgPredicateBuilder {
         return new ImdgPredicateHazelcast(Predicates.and(paramAMy.hazelcastPredicate, paramBMy.hazelcastPredicate));
     }
 
-    @Override
-    public ImdgPredicate and(ImdgPredicate paramA, ImdgPredicate paramB, ImdgPredicate paramC) {
-        ImdgPredicateHazelcast paramAMy = (ImdgPredicateHazelcast) paramA;
-        ImdgPredicateHazelcast paramBMy = (ImdgPredicateHazelcast) paramB;
-        ImdgPredicateHazelcast paramCMy = (ImdgPredicateHazelcast) paramC;
-        return new ImdgPredicateHazelcast(Predicates.and(paramAMy.hazelcastPredicate, paramBMy.hazelcastPredicate, paramCMy.hazelcastPredicate));
-    }
-
     @Override
     public ImdgPredicate and(ImdgPredicate... param) {
         Predicate[] paramMy = new Predicate[param.length];
@@ -83,11 +75,12 @@ public class ImdgPredicateBuilderHazelcast implements ImdgPredicateBuilder {
     }
 
     @Override
-    public ImdgPredicate or(ImdgPredicate paramA, ImdgPredicate paramB, ImdgPredicate paramC) {
+    public ImdgPredicate or(ImdgPredicate... param) {
-        ImdgPredicateHazelcast paramAMy = (ImdgPredicateHazelcast) paramA;
+        Predicate[] paramMy = new Predicate[param.length];
-        ImdgPredicateHazelcast paramBMy = (ImdgPredicateHazelcast) paramB;
+        for (int i = 0; i < param.length; i++) {
-        ImdgPredicateHazelcast paramCMy = (ImdgPredicateHazelcast) paramC;
+            paramMy[i] = ((ImdgPredicateHazelcast) param[i]).getRawPredicate();
-        return new ImdgPredicateHazelcast(Predicates.or(paramAMy.hazelcastPredicate, paramBMy.hazelcastPredicate, paramCMy.hazelcastPredicate));
+        }
+        return new ImdgPredicateHazelcast(Predicates.or(paramMy));
     }
 
     @Override

platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/predicate/ImdgPredicateBuilder.java

@@ -11,10 +11,9 @@ public interface ImdgPredicateBuilder {
 
     ImdgPredicate not(ImdgPredicate param);
     ImdgPredicate and(ImdgPredicate paramA, ImdgPredicate paramB);
-    ImdgPredicate and(ImdgPredicate paramA, ImdgPredicate paramB, ImdgPredicate paramC);
     ImdgPredicate and(ImdgPredicate... param);
     ImdgPredicate or(ImdgPredicate paramA, ImdgPredicate paramB);
-    ImdgPredicate or(ImdgPredicate paramA, ImdgPredicate paramB, ImdgPredicate paramC);
+    ImdgPredicate or(ImdgPredicate... param);
 
     ImdgPredicate sql(String sql);
     ImdgPredicate regex(String key, String regex);

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

@@ -75,6 +75,7 @@ public interface Consts {
     String CREATE_REPORT_FOR_SESSION_ID = "create-report-for-session-id";
     String CREATE_REPORT_FOR_SESSION_ID_LIST = "create-report-for-session-id-list";
     String CREATE_REPORT_FOR_PERIOD = "create-report-for-period";
+    String CREATE_REPORT_FOR_REGISTRY = "create-report-for-registry";
 
     @Deprecated
     String ACCOUNT_NEW_SDF01 = "account-new-sdf01";

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/reports/ReportRequestWithRegistryId.java

@@ -0,0 +1,32 @@
+package ru.spcex.clearing.platform.messaging.domain.cud.reports;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Запрос на построение отчета по SessionId, RegistryId
+ */
+public class ReportRequestWithRegistryId extends ReportRequest {
+    /**
+     * Идентификатор сессии
+     */
+    @JsonProperty
+    public Long sessionId;
+    @JsonProperty
+    public Long registryId;
+
+    public Long getSessionId() {
+        return sessionId;
+    }
+
+    public void setSessionId(Long sessionId) {
+        this.sessionId = sessionId;
+    }
+
+    public Long getRegistryId() {
+        return registryId;
+    }
+
+    public void setRegistryId(Long registryId) {
+        this.registryId = registryId;
+    }
+}