From fa98b9f7afe27d78582be123be3cf7cdb3edf7bb Mon Sep 17 00:00:00 2001 From: AKurakin Date: Tue, 23 May 2023 12:25:12 +0300 Subject: [PATCH] clearing-service http://jira.mfd.msk:8088/browse/CLS-290 stage 10 --- .../builder/sql/RegistryCodeSqlBuilder.java | 24 +++ .../session/stage/impl/FinishingSession.java | 142 ++++++++++++++++++ .../ImdgPredicateBuilderHazelcast.java | 19 +-- .../api/predicate/ImdgPredicateBuilder.java | 3 +- .../platform/messaging/domain/Consts.java | 1 + .../reports/ReportRequestWithRegistryId.java | 32 ++++ 6 files changed, 206 insertions(+), 15 deletions(-) create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/FinishingSession.java create mode 100644 platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/reports/ReportRequestWithRegistryId.java diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/builder/sql/RegistryCodeSqlBuilder.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/builder/sql/RegistryCodeSqlBuilder.java index 4dd315001..376a9036c 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/builder/sql/RegistryCodeSqlBuilder.java +++ b/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 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()])); + } } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/FinishingSession.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/FinishingSession.java new file mode 100644 index 000000000..a14991891 --- /dev/null +++ b/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 registryImdg; + private final Imdg 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 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 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 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> 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); + } + +} diff --git a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/predicate/ImdgPredicateBuilderHazelcast.java b/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/predicate/ImdgPredicateBuilderHazelcast.java index 60f97795e..c3391a42d 100644 --- a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/predicate/ImdgPredicateBuilderHazelcast.java +++ b/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) { - ImdgPredicateHazelcast paramAMy = (ImdgPredicateHazelcast) paramA; - ImdgPredicateHazelcast paramBMy = (ImdgPredicateHazelcast) paramB; - ImdgPredicateHazelcast paramCMy = (ImdgPredicateHazelcast) paramC; - return new ImdgPredicateHazelcast(Predicates.or(paramAMy.hazelcastPredicate, paramBMy.hazelcastPredicate, paramCMy.hazelcastPredicate)); + public ImdgPredicate or(ImdgPredicate... param) { + Predicate[] paramMy = new Predicate[param.length]; + for (int i = 0; i < param.length; i++) { + paramMy[i] = ((ImdgPredicateHazelcast) param[i]).getRawPredicate(); + } + return new ImdgPredicateHazelcast(Predicates.or(paramMy)); } @Override diff --git a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/predicate/ImdgPredicateBuilder.java b/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/predicate/ImdgPredicateBuilder.java index e0c4196d2..e664ebe0b 100644 --- a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/predicate/ImdgPredicateBuilder.java +++ b/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); diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java index 7dac16640..4ddec1fd1 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java +++ b/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"; diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/reports/ReportRequestWithRegistryId.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/reports/ReportRequestWithRegistryId.java new file mode 100644 index 000000000..9dd2ff2b1 --- /dev/null +++ b/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; + } +}