From dbec4604f9f4bfccb3e96ac1c881ee81a3b1cb2d Mon Sep 17 00:00:00 2001 From: AKurakin Date: Mon, 22 May 2023 12:34:51 +0300 Subject: [PATCH] clearing-service http://jira.mfd.msk:8088/browse/CLS-290 stage 11 --- .../service/ExecutionDepositComponent.java | 8 +- .../clearing/session/stage/TaskType.java | 7 ++ .../session/stage/impl/BalanceRevise.java | 3 +- .../stage/impl/EndStageNotification.java | 112 ++++++++++++++++++ .../task/EndStageNotificationPayload.java | 13 ++ .../platform/messaging/domain/Consts.java | 1 + 6 files changed, 140 insertions(+), 4 deletions(-) create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/EndStageNotification.java create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/task/EndStageNotificationPayload.java diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ExecutionDepositComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ExecutionDepositComponent.java index 81cddeb36..d7756f23a 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ExecutionDepositComponent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ExecutionDepositComponent.java @@ -54,7 +54,7 @@ public class ExecutionDepositComponent { //fixme ждать ТЗ Long tradeNum; Instant tradingDay; - private final IMessageResolver msgResolver = new SimpleMessageResolver(); + private final IMessageResolver msgResolver; private final Function stradesValidator; private final KafkaSender kafkaSender; @@ -62,12 +62,14 @@ public class ExecutionDepositComponent { @Autowired public ExecutionDepositComponent(ImdgProvider imdgProvider, Producer kafka, @Qualifier("sTradesValidator") Function stradesValidator, - @Qualifier("kafkaSenderWithoutRequestInfo") KafkaSender kafkaSender) { + @Qualifier("kafkaSenderWithoutRequestInfo") KafkaSender kafkaSender, + IMessageResolver msgResolver) { this.sTradeImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class); this.executionDepositImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ExecutionDeposit, ExecutionDeposit.class); this.listingImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Listing, Listing.class); this.stradesValidator = stradesValidator; this.kafkaSender = kafkaSender; + this.msgResolver = msgResolver; resetTradingDay(); } @@ -127,7 +129,7 @@ public class ExecutionDepositComponent { } ExecutionDeposit newED; try { - newED = createExecutionDeposit(sTrd, validator); + newED = createExecutionDeposit(sTrd, validator); executionDepositImdg.insert(newED); sendNotification(newED); log.debug("New executionDeposit.id={} was created.", newED.getId()); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/TaskType.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/TaskType.java index 72a2f42e8..a6a3bac15 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/TaskType.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/TaskType.java @@ -34,4 +34,11 @@ public enum TaskType { * step 8 */ UnlockResources, + + + + /** + * Step 11 + */ + EndStageNotification; } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/BalanceRevise.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/BalanceRevise.java index 4614f2276..1cebb2a3b 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/BalanceRevise.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/BalanceRevise.java @@ -43,12 +43,13 @@ public class BalanceRevise implements ISessionStage { private KafkaSender kafkaSender; @Autowired - public BalanceRevise(ImdgProvider imdgProvider) { + public BalanceRevise(ImdgProvider imdgProvider, KafkaSender kafkaSender) { this.imdgProvider = imdgProvider; this.idGenerator = imdgProvider.getImdgIdGenerator(); this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class); this.currencyImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Currency, Currency.class); this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); + this.kafkaSender = kafkaSender; } @Override diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/EndStageNotification.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/EndStageNotification.java new file mode 100644 index 000000000..ef76991c9 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/EndStageNotification.java @@ -0,0 +1,112 @@ +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.registry.Registry; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.clearing.SdfClearingRequest; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +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.EndStageNotificationPayload; +import ru.spcex.platform.enumeration.*; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; +import ru.spcex.platform.utils.enumeration.EnumMessage; +import ru.spcex.platform.utils.enumeration.IMessageResolver; + +import java.util.Collection; +import java.util.Objects; +import java.util.stream.Collectors; + +import static ru.spcex.clearing.error.ClearingErrorInternal.SessionGeneralError; + +@Service +public class EndStageNotification implements ISessionStage { + private final Logger log = LoggerFactory.getLogger(getClass()); + + private final Imdg registryImdg; + private final KafkaSender kafkaSender; + private final IMessageResolver msgResolver; + + @Autowired + public EndStageNotification(ImdgProvider imdgProvider, KafkaSender kafkaSender, IMessageResolver msgResolver) { + this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); + this.kafkaSender = kafkaSender; + this.msgResolver = msgResolver; + } + + @Override + public StageResult submit(Task task) { + EndStageNotificationPayload payload = (EndStageNotificationPayload) task.getData(); + switch (task.getTaskType()) { + case EndStageNotification -> { + return endStageNotification(payload.getSection()); + } + default -> { + throw new IllegalStateException("Unknown task type: " + task.getTaskType()); + } + } + } + + protected Collection selectRegistry() { + String registrySQL = "registryStatus=" + RegistryStatus.OK.getKey() + ")"; + Collection result = registryImdg.getCollectionObjectsBySQL(registrySQL); + log.trace("Selected {} registry's by sql: {}", result.size(), registrySQL); + return result; + } + + protected StageResult endStageNotification(String section) { + Collection forRegistries = selectRegistry(); + Collection groups = forRegistries.stream() + .map(Registry::getGroupId) + .filter(Objects::nonNull) + .distinct().collect(Collectors.toList()); + log.debug("Sending notifications for {} groups ({} registers) on section {}", + groups, forRegistries.size(), section); + for (Long groupId : groups) { + log.trace("For registry group {} send notification", + groupId); + StageResult sResult; + if (Section.FOND.equalsByKey(section)) { + sResult = notificationDF14(groupId); + } else if (Section.MKR.equalsByKey(section)) { + sResult = notificationDF05(groupId); + } else { + throw new IllegalArgumentException("Unsupported section " + section); + } + if (sResult.getError() != null) { + log.warn("When sending groupId={} has error: {}", groupId, msgResolver.resolve(sResult.getError())); + } + } + StageResult> res = new StageResult<>(null, true); + return res; + } + + protected StageResult notificationDF14(Long groupId) { + SdfClearingRequest sdf14Request = new SdfClearingRequest(); + sdf14Request.setGroupId(groupId); + Long msgKey = kafkaSender.sendRequestToQueue(Consts.SDF14_PROCESS, sdf14Request); + if (msgKey == null) { + log.error("failed to put SDF14 request to kafka queue"); + return new StageResult<>(new EnumMessage(SessionGeneralError), false); + } + return new StageResult<>(null, true); + } + + protected StageResult notificationDF05(Long groupId) { + SdfClearingRequest sdf05Request = new SdfClearingRequest(); + sdf05Request.setGroupId(groupId); + Long msgKey = kafkaSender.sendRequestToQueue(Consts.SDF05_PROCESS, sdf05Request); + if (msgKey == null) { + log.error("failed to put SDF05 request to kafka queue"); + return new StageResult<>(new EnumMessage(SessionGeneralError), false); + } + return new StageResult<>(null, true); + } + +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/task/EndStageNotificationPayload.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/task/EndStageNotificationPayload.java new file mode 100644 index 000000000..ba2b3836c --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/task/EndStageNotificationPayload.java @@ -0,0 +1,13 @@ +package ru.spcex.clearing.session.stage.task; + +public class EndStageNotificationPayload { + private String section; + + public String getSection() { + return section; + } + + public void setSection(String section) { + this.section = section; + } +} 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 7d80037cc..7dac16640 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 @@ -111,6 +111,7 @@ public interface Consts { String SDF05_PROCESS = "sdf05-process"; String SDF07_PROCESS = "sdf07-process"; String SDF11_PROCESS = "sdf11-process"; + String SDF14_PROCESS = "sdf14-process"; String SDF51_PROCESS = "sdf51-process"; String SDF53_PROCESS = "sdf53-process"; String SDF54_PROCESS = "sdf54-process";