diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/state_machine_2/specific/FinalSessionStateMachineConfig.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/state_machine_2/specific/FinalSessionStateMachineConfig.java index fc28ad198..082416f80 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/state_machine_2/specific/FinalSessionStateMachineConfig.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/state_machine_2/specific/FinalSessionStateMachineConfig.java @@ -2,13 +2,9 @@ package ru.spcex.clearing.config.state_machine_2.specific; import java.time.LocalDate; import java.util.Arrays; +import java.util.EnumSet; import java.util.HashSet; -import java.util.List; -import java.util.function.Supplier; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Configuration; import org.springframework.statemachine.config.EnableStateMachineFactory; import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter; @@ -16,7 +12,6 @@ import org.springframework.statemachine.config.builders.StateMachineConfiguratio import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; import org.springframework.statemachine.listener.StateMachineListenerAdapter; -import org.springframework.statemachine.state.State; import ru.clearing.classes.statics.data.misc.Session; import ru.clearing.classes.statics.data.registry.Registry; import ru.spcex.clearing.imdg.IMDGDistributedNames; @@ -24,6 +19,8 @@ import ru.spcex.clearing.session.stage.TaskType; import ru.spcex.clearing.session.state.SsnEvent; import ru.spcex.clearing.session.state.action.CreateSessionAction; import ru.spcex.clearing.session.state.action.DealsPrepareAction; +import ru.spcex.clearing.session.state.action.EndStageNotificationAction; +import ru.spcex.clearing.session.state.action.FinishingSessionAction; import ru.spcex.clearing.session.state.action.FormingPaymentInstructionAssetsAction; import ru.spcex.clearing.session.state.action.FormingPaymentInstructionReturnMkrAction; import ru.spcex.clearing.session.state.action.InclusionToPoolAction; @@ -33,9 +30,16 @@ import ru.spcex.clearing.session.state.action.ObligationAdmissionAction; import ru.spcex.clearing.session.state.action.RequirementAndObligationCreationAction; import ru.spcex.clearing.session.state.action.ReviseStage1; import ru.spcex.clearing.session.state.action.Sdf56Action; +import ru.spcex.clearing.session.state.action.SdfReceivedAction; +import ru.spcex.clearing.session.state.guard.PaymentsWereCreatedGuard; +import ru.spcex.clearing.session.state.guard.PaymentsWereNotCreatedGuard; +import ru.spcex.clearing.session.state.guard.SdfGuard; +import ru.spcex.clearing.session.state.guard.SdfGuardExtractorAfterAssets; +import ru.spcex.clearing.session.state.listener.MachineMonitoringListener; import ru.spcex.clearing.util.StateMachineUtil; import ru.spcex.platform.classes.base.interfaces.ExecutionType; import ru.spcex.platform.enumeration.RegistryStatus; +import ru.spcex.platform.enumeration.Section; import ru.spcex.platform.enumeration.SessionType; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; @@ -45,10 +49,8 @@ import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; @EnableStateMachineFactory(name = "FINL") public class FinalSessionStateMachineConfig extends EnumStateMachineConfigurerAdapter { - private final Logger log = LoggerFactory.getLogger(getClass()); private final Imdg sessionImdg; - private final Imdg rgsImdg; private final Sdf56Action sendSdf56Action; private final ReviseStage1 reviseStage1Action; private final DealsPrepareAction dealsPrepareAction; @@ -59,13 +61,16 @@ public class FinalSessionStateMachineConfig private final InspectionObligationsV2Action inspectionObligationsV2Action; private final FormingPaymentInstructionReturnMkrAction formingPaymentInstructionReturnMkrAction; private final FormingPaymentInstructionAssetsAction formingPaymentInstructionAssetsAction; + private final FinishingSessionAction finishingSessionAction; + private final EndStageNotificationAction endStageNotificationAction; + + private final SessionType SESSION_TYPE = SessionType.FINL; + private final Section SECTION = Section.MKR; @Autowired public FinalSessionStateMachineConfig( ImdgProvider imdgProvider, - @Qualifier("marketCodesForBn") - Supplier> marketCodes, Sdf56Action sendSdf56Action, ReviseStage1 reviseStage1Action, DealsPrepareAction dealsPrepareAction, @@ -75,11 +80,12 @@ public class FinalSessionStateMachineConfig InspectionObligationsDepositReturnAction inspOblDepositReturnAction, InspectionObligationsV2Action inspectionObligationsV2Action, FormingPaymentInstructionReturnMkrAction formingPaymentInstructionReturnMkrAction, - FormingPaymentInstructionAssetsAction formingPaymentInstructionAssetsAction + FormingPaymentInstructionAssetsAction formingPaymentInstructionAssetsAction, + FinishingSessionAction finishingSessionAction, EndStageNotificationAction endStageNotificationAction ) { this.sendSdf56Action = sendSdf56Action; this.sessionImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Session, Session.class); - this.rgsImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); + Imdg rgsImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); this.reviseStage1Action = reviseStage1Action; this.dealsPrepareAction = dealsPrepareAction; this.reqAndOblAction = reqAndOblAction; @@ -89,6 +95,10 @@ public class FinalSessionStateMachineConfig this.inspectionObligationsV2Action = inspectionObligationsV2Action; this.formingPaymentInstructionReturnMkrAction = formingPaymentInstructionReturnMkrAction; this.formingPaymentInstructionAssetsAction = formingPaymentInstructionAssetsAction; + this.finishingSessionAction = finishingSessionAction; + this.endStageNotificationAction = endStageNotificationAction; + this.finishingSessionAction.setSessionType(SESSION_TYPE); + this.finishingSessionAction.setSection(SECTION); //stages settings: ImdgPredicateBuilder rgsPrctBuilder = rgsImdg.predicateBuilder(); @@ -102,16 +112,19 @@ public class FinalSessionStateMachineConfig rgsPrctBuilder.lessEqual("valueDate", LocalDate.now()) ) ); - this.inspOblDepositReturnAction.setSessionType(SessionType.FINL); - this.inspectionObligationsV2Action.setSessionType(SessionType.FINL); + this.inspOblDepositReturnAction.setSessionType(SESSION_TYPE); + this.inspectionObligationsV2Action.setSessionType(SESSION_TYPE); } @Override public void configure(StateMachineStateConfigurer states) throws Exception { states .withStates() - .initial(TaskType.StartRevise, StateMachineUtil.chain(new CreateSessionAction(sessionImdg), sendSdf56Action)) - //.state(TaskType.StartRevise, sendSdf56Action) //entry action + .initial(TaskType.StartRevise, StateMachineUtil.chain( + new CreateSessionAction(sessionImdg) + .sessionType(SESSION_TYPE) + .section(SECTION), + sendSdf56Action)) .states(new HashSet<>(Arrays.asList( TaskType.StartRevise, TaskType.StartRevisePart1, @@ -125,7 +138,8 @@ public class FinalSessionStateMachineConfig TaskType.FormingPaymentInstruction, TaskType.FormingPaymentInstruction, TaskType.FinishingSession, - TaskType.EndStageNotification + TaskType.EndStageNotification, + TaskType.PSEUDO_waitingForSdf ))) .end(TaskType.EndStageNotification); } @@ -164,26 +178,46 @@ public class FinalSessionStateMachineConfig .source(TaskType.InspectionObligations).target(TaskType.FormingPaymentInstruction) .action(formingPaymentInstructionReturnMkrAction) .action(formingPaymentInstructionAssetsAction) - ; - - - - //TaskType.RequirementsAndObligationsCreate - + .and() + //если не создалось paymentInstruction'ов + .withExternal() + .source(TaskType.FormingPaymentInstruction) + .target(TaskType.FinishingSession) + .guard(PaymentsWereNotCreatedGuard.instance) + .action(finishingSessionAction) + .and() + //если создались paymentInstruction, переходим в режим ожидания + .withExternal() + .source(TaskType.FormingPaymentInstruction) + .target(TaskType.PSEUDO_waitingForSdf) + .guard(PaymentsWereCreatedGuard.instance) + .and() + .withExternal() + .source(TaskType.PSEUDO_waitingForSdf) + .target(TaskType.FinishingSession) + .guard(new SdfGuard(SdfGuardExtractorAfterAssets.instance)) + .action(finishingSessionAction) + .and() + .withExternal() + .source(TaskType.FinishingSession) + .target(TaskType.EndStageNotification) + .action(endStageNotificationAction); + for (var event: EnumSet.of(SsnEvent.SDF_01, SsnEvent.SDF_57, SsnEvent.SDF_04)) { + transitions + .withInternal() + .source(TaskType.PSEUDO_waitingForSdf) + .event(event) + .action(SdfReceivedAction.instance); + } } @Override public void configure(StateMachineConfigurationConfigurer config) throws Exception { - StateMachineListenerAdapter loggingChangeStateListener = new StateMachineListenerAdapter<>() { - @Override - public void stateEntered(State state) { - TaskType enteredState = state != null ? state.getId() : null; - log.info("State entered: {}", enteredState); - } - }; + StateMachineListenerAdapter loggingChangeStateListener + = new MachineMonitoringListener(SESSION_TYPE.getKey()); config .withConfiguration() - .machineId(SessionType.FINL.getKey()) + .machineId(SESSION_TYPE.getKey()) .listener(loggingChangeStateListener) ; } 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 da90680de..a78cb2250 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 @@ -55,7 +55,9 @@ public enum TaskType implements IEnumKey { /** * Step 11 */ - EndStageNotification("CL11"); + EndStageNotification("CL11"), + PSEUDO_waitingForSdf("PSEUDO_waitingForSdf") + ; private String key; @@ -67,4 +69,8 @@ public enum TaskType implements IEnumKey { public String getKey() { return key; } + + public boolean isPseudoStatus() { + return this.key.startsWith("PSEUDO"); + } } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/ActionHeader.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/ActionHeader.java index 78b1c0724..d220b2404 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/ActionHeader.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/ActionHeader.java @@ -1,5 +1,5 @@ package ru.spcex.clearing.session.state; public enum ActionHeader { - sdf56FromTime; + sdf56FromTime, pr; } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/DataEnum.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/DataEnum.java index 0edf00fc3..ae5680bab 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/DataEnum.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/DataEnum.java @@ -3,6 +3,7 @@ package ru.spcex.clearing.session.state; public enum DataEnum { sessionId, //Long sessionType, //Session + section, session, //Session dealsPrepared, //List counterPartyId, //Long diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/SsnEvent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/SsnEvent.java index ff471b212..f772b8cce 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/SsnEvent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/SsnEvent.java @@ -4,4 +4,7 @@ public enum SsnEvent { Revise, Sdf57Processed, SdfReceived, + SDF_01, + SDF_57, + SDF_04, } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/CreateSessionAction.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/CreateSessionAction.java index 944537ecd..2a41fdb9c 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/CreateSessionAction.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/CreateSessionAction.java @@ -17,27 +17,43 @@ import ru.spcex.platform.imdg.api.Imdg; public class CreateSessionAction extends AbstractSessionActionForOkErrorHandling { private final Imdg sessionImdg; private final Logger log = LoggerFactory.getLogger(CreateSessionAction.class); + private SessionType sessionType; + private Section section; public CreateSessionAction(Imdg sessionImdg) { this.sessionImdg = sessionImdg; } + public CreateSessionAction sessionType(SessionType sessionType) { + this.sessionType = sessionType; + return this; + } + + public CreateSessionAction section(Section section) { + this.section = section; + return this; + } + @Override public void actualExecute(StateContext context) { Session existActiveSession = sessionImdg.getFirstObjectByFieldValues(Map.of("workflowStatus", SessionStatus.ACTV.getKey())); if (existActiveSession != null) { throw new RuntimeException("ActiveSessionIsPresent " + existActiveSession); } + if (section == null | sessionType == null) { + throw new RuntimeException("CreateSessionAction is misconfigured: no section/sessionType"); + } Session newSession = new Session(); - newSession.setSection(Section.FOND.getKey());//fixme - newSession.setSessionType(SessionType.IPOB.getKey());//fixme + newSession.setSection(section.getKey()); + newSession.setSessionType(sessionType.getKey()); newSession.setSessionStatus(TaskType.StartRevise.getKey()); newSession.setWorkflowStatus(SessionStatus.ACTV.getKey()); newSession.setClearingDate(LocalDate.now()); sessionImdg.insert(newSession); log.info("started new session.id={}", newSession.getId()); context.getExtendedState().getVariables().put(DataEnum.sessionId.name(), newSession.getId()); - context.getExtendedState().getVariables().put(DataEnum.sessionType.name(), SessionType.IPOB); + context.getExtendedState().getVariables().put(DataEnum.sessionType.name(), sessionType); + context.getExtendedState().getVariables().put(DataEnum.section.name(), section); context.getExtendedState().getVariables().put(DataEnum.session.name(), newSession); } } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/EndStageNotificationAction.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/EndStageNotificationAction.java new file mode 100644 index 000000000..034e47f9e --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/EndStageNotificationAction.java @@ -0,0 +1,123 @@ +package ru.spcex.clearing.session.state.action; + +import java.time.Instant; +import java.util.Collection; +import java.util.Objects; +import java.util.Optional; +import java.util.stream.Collectors; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.context.annotation.Scope; +import org.springframework.statemachine.StateContext; +import org.springframework.stereotype.Service; +import ru.clearing.classes.statics.data.misc.Session; +import ru.clearing.classes.statics.data.registry.Registry; +import static ru.spcex.clearing.error.ClearingErrorInternal.SessionGeneralError; +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.TaskType; +import ru.spcex.clearing.session.state.DataEnum; +import ru.spcex.clearing.session.state.SsnEvent; +import ru.spcex.platform.enumeration.RegistryStatus; +import ru.spcex.platform.enumeration.Section; +import ru.spcex.platform.enumeration.SessionStatus; +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; + +@Service +@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) +public class EndStageNotificationAction extends AbstractSessionActionForOkErrorHandling { + 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 EndStageNotificationAction(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 + protected void actualExecute(StateContext ctx) { + Long sessionId = ctx.getExtendedState().get(DataEnum.sessionId, Long.class); + String section = ctx.getExtendedState().get(DataEnum.section, String.class); + // 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()); + } + + 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); + Optional 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); + } + sResult.ifPresent(enumMessage -> + log.error("When sending groupId={} has error: {}", groupId, msgResolver.resolve(enumMessage))); + } + } + + 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 Optional 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 Optional.of(new EnumMessage(SessionGeneralError)); + } + return Optional.empty(); + } + + protected Optional 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 Optional.of(new EnumMessage(SessionGeneralError)); + } + return Optional.empty(); + } + +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/FinishingSessionAction.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/FinishingSessionAction.java new file mode 100644 index 000000000..ef5cc3b01 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/FinishingSessionAction.java @@ -0,0 +1,376 @@ +package ru.spcex.clearing.session.state.action; + +import java.time.Instant; +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.context.annotation.Scope; +import org.springframework.statemachine.StateContext; +import org.springframework.stereotype.Service; +import ru.clearing.classes.statics.data.execution.ExecutionCommon; +import ru.clearing.classes.statics.data.execution.ExecutionCurrency; +import ru.clearing.classes.statics.data.execution.ExecutionDeposit; +import ru.clearing.classes.statics.data.execution.ExecutionFond; +import ru.clearing.classes.statics.data.misc.Session; +import ru.clearing.classes.statics.data.registry.Registry; +import ru.clearing.classes.statics.data.sdf.SDf05; +import static ru.spcex.clearing.error.ClearingErrorInternal.SessionGeneralError; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +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.Sdf05Sender; +import ru.spcex.clearing.service.Sdf14Sender; +import ru.spcex.clearing.session.stage.TaskType; +import ru.spcex.clearing.session.state.ActionHeader; +import ru.spcex.clearing.session.state.DataEnum; +import ru.spcex.clearing.session.state.SsnEvent; +import ru.spcex.platform.classes.base.interfaces.ExecutionType; +import ru.spcex.platform.enumeration.CoverageStatus; +import ru.spcex.platform.enumeration.RegistryDesignation; +import ru.spcex.platform.enumeration.RegistryStatus; +import ru.spcex.platform.enumeration.RegistryTradingParams; +import ru.spcex.platform.enumeration.Section; +import ru.spcex.platform.enumeration.SessionType; +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.imdg.api.predicate.specific.RegistryCodeSqlBuilder; +import ru.spcex.platform.utils.collection.Pair; +import ru.spcex.platform.utils.enumeration.EnumMessage; +import ru.spcex.platform.utils.enumeration.IEnumKey; +import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey; +import ru.spcex.platform.utils.enumeration.IMessageResolver; + +@Service +@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) +public class FinishingSessionAction extends AbstractSessionActionForOkErrorHandling { + private final Logger log = LoggerFactory.getLogger(getClass()); + + private final ImdgProvider imdgProvider; + private final Imdg registryImdg; + private final Imdg executionFondImdg; + private final Imdg executionDepositImdg; + private final Imdg executionCurrencyImdg; + private final Imdg sessionImdg; + private final Imdg sDf05Imdg; + private final KafkaSender kafkaSender; + private final IMessageResolver msgResolver; + private final Sdf05Sender sdf05Sender; + private final Sdf14Sender sdf14Sender; + private SessionType sessionType; + private Section section; + + + @Autowired + public FinishingSessionAction(ImdgProvider imdgProvider, + KafkaSender kafkaSender, + IMessageResolver msgResolver, + Sdf05Sender sdf05Sender, + Sdf14Sender sdf14Sender) { + this.imdgProvider = imdgProvider; + this.sessionImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Session, Session.class); + this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); + this.executionFondImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ExecutionFond, ExecutionFond.class); + this.executionDepositImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ExecutionDeposit, ExecutionDeposit.class); + this.executionCurrencyImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ExecutionCurrency, ExecutionCurrency.class); + this.sDf05Imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf05, SDf05.class); + this.kafkaSender = kafkaSender; + this.msgResolver = msgResolver; + this.sdf05Sender = sdf05Sender; + this.sdf14Sender = sdf14Sender; + } + + public void setSection(Section section) { + this.section = section; + } + public void setSessionType(SessionType sessionType) { + this.sessionType = sessionType; + } + + + @Override + protected void actualExecute(StateContext ctx) { + Long sessionId = ctx.getExtendedState().get(DataEnum.sessionId, Long.class); + String pr = (String) ctx.getMessageHeader(ActionHeader.pr); + Instant now = Instant.now(); + //установка CLRD для обработанных регистров + Collection allLiabilitiesAndOKClaims = selectClaimsAndLiabilities(sessionId); + Collection claimsAndLiabilities = allLiabilitiesAndOKClaims + .stream() + .filter(rgs -> RegistryDesignation.T.equalsByKey(rgs.getRegistryDesignation()) + || RegistryStatus.OK.equalsByKey(rgs.getRegistryStatus())) + .toList(); + { + Map rgsToUpdate = new HashMap<>(); + claimsAndLiabilities.forEach(rgs -> { + rgs.setRegistryStatus(RegistryStatus.CLRD.getKey()); + rgs.setUpdated(now); + // registryImdg.update(rgs); + rgsToUpdate.put(rgs.getId(), rgs); + }); + registryImdg.putAll(rgsToUpdate, 200); + } + log.debug("For sessionId={} was updated {} registers for status={}", sessionId, claimsAndLiabilities.size(), RegistryStatus.CLRD.getKey()); + + // Отметка статуса в Execution* + Session session = sessionImdg.getSingleObjectByID(sessionId); + if (session == null) { + log.error("Session id={} not found. Can not update Execution's.", sessionId); + } else { + Collection oRegs = findORegistryBySessionId(allLiabilitiesAndOKClaims); + Map obligationStatusByGroupId; + obligationStatusByGroupId = oRegs.stream() + .map(rgs -> new Pair<>(rgs.getGroupId(), getEnumByKey(RegistryStatus.class, rgs.getRegistryStatus()))) + .filter(pair -> pair.getSecond() != null) + .collect(Collectors.toMap(Pair::getFirst, Pair::getSecond, (o1, o2) -> o1)); + Function statusByExchangeId = id -> { + RegistryStatus obligationStatus = obligationStatusByGroupId.get(id); + if (obligationStatus == null) return null; + return switch (obligationStatus) { + case CLRD -> CoverageStatus.ALWD; + case FAIL, UNCV, NACK, NACC -> CoverageStatus.DEND; + default -> null; + }; + }; + Set exchangeExecutionIdsPreviousDay = oRegs.stream() + //.filter(rgs -> Objects.nonNull(rgs.getSettlementDate())) + .filter(rgs -> Objects.nonNull(rgs.getTradingDate())) + .filter(rgs -> rgs.getSettlementDate().isAfter(rgs.getTradingDate())) + .map(Registry::getGroupId) + .collect(Collectors.toSet()); + log.trace("obligations number with second leg in the past: {}", exchangeExecutionIdsPreviousDay.size()); + int allowed = 0; + int notAllowed = 0; + SessionType sessionType = getEnumByKey(SessionType.class, session.getSessionType()); + if (IEnumKey.contains(sessionType, SessionType.FINL, SessionType.MEDM, SessionType.UNIT)) { + Map depos = new HashMap<>(); + Collection executions = findExecutionDepositBySessionId(sessionId); + log.trace("loaded {} ExecutionDeposits for sessionId {}", executions.size(), sessionId); + for (ExecutionDeposit execution : executions) { + CoverageStatus toStatus = statusByExchangeId.apply(execution.getExchangeExecutionId()); + if (toStatus == null) { + log.trace("CoverageStatus not defined for execution.id={}", execution.getId()); + continue; + } + if (toStatus == CoverageStatus.DEND) { + notAllowed++; + } + if (toStatus == CoverageStatus.ALWD) { + allowed++; + } + execution.setCoverageStatus(toStatus.getKey()); + execution.setUpdated(Instant.now()); + depos.put(execution.getId(), execution); +// executionDepositImdg.update(execution); + } + executionDepositImdg.putAll(depos, 200); + log.trace("By sessionId={} processed {} ExecutionDeposit: {} allowed, {} denied, skipped {}", + sessionId, + executions.size(), + allowed, + notAllowed, + executions.size() - (allowed + notAllowed)); + } + if (IEnumKey.contains(sessionType, SessionType.CURR, SessionType.TRDT, SessionType.IPOB, SessionType.IPO0, SessionType.IPOT, SessionType.UNIT)) { + Collection executions = findExecutionFondBySessionId(sessionId, sessionType); + Map execCurr = new HashMap<>(); + Map execFond = new HashMap<>(); + log.trace("loaded {} Executions for sessionId {}", executions.size(), sessionId); + for (ExecutionCommon execution : executions) { + CoverageStatus toStatus = statusByExchangeId.apply(execution.getExchangeExecutionId()); + if (toStatus == null) { + log.trace("CoverageStatus not defined for execution.id={}", execution.getId()); + continue; + } + if (toStatus == CoverageStatus.DEND) { + notAllowed++; + } + if (toStatus == CoverageStatus.ALWD) { + allowed++; + } + execution.setCoverageStatus(toStatus.getKey()); + execution.setUpdated(Instant.now()); + if (execution.type().equals(ExecutionType.ExecutionCurrency)) { + execCurr.put(execution.getId(), (ExecutionCurrency) execution); + //executionCurrencyImdg.update((ExecutionCurrency) execution); + } else { + execFond.put(execution.getId(), (ExecutionFond) execution); + //executionFondImdg.update((ExecutionFond) execution); + } + } + executionCurrencyImdg.putAll(execCurr, 200); + executionFondImdg.putAll(execFond, 200); + log.trace("By sessionId={} processed {} ExecutionFond: allowed {}, denied {}, skipped {}", + sessionId, + executions.size(), + allowed, + notAllowed, + executions.size() - (allowed + notAllowed)); + } else { + log.warn("Unsupported session[{}].SessionType={} for update Executions", sessionId, session.getSessionType()); + } + } + + // 2. Отправка сообщений + Collection forRegistries = selectRegistry(sessionId); + log.debug("Found {} registries for sessionId={}", forRegistries.size(), sessionId); + + Optional sResult = toReportSession(sessionId); + if (sResult.isPresent()) { + String err = msgResolver.resolve(sResult.get()); + log.error("When sending sessionId={} has error: {}", sessionId, err); + throw new RuntimeException(err); + } + for (Registry registryA : forRegistries) { + if (!RegistryDesignation.A.equalsByKey(registryA.getRegistryUnit())) { //fixme Unit -> Designation + log.warn("For registry {}.RegistryDesignation is not A.", registryA.getId()); + continue; + } + sResult = toReportMoney(sessionId, registryA); + if (sResult.isPresent()) { + log.warn("When sending sessionId={}, registry.id={} has error: {}", sessionId, registryA.getId(), msgResolver.resolve(sResult.get())); + throw new RuntimeException(msgResolver.resolve(sResult.get())); + } + } + sdf05Sender.sendSdf05(pr); + if (Section.FOND.equals(section) || Section.MULT.equals(section)) { + sdf14Sender.sendSdf14(sessionId); + } + } + + /** + * формирование операционного отчета об обязательствах + **/ + protected Optional 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 Optional.of(new EnumMessage(SessionGeneralError)); + } + return Optional.empty(); + } + + /** + * формирование операционного отчета о денежных средствах + **/ + protected Optional 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 Optional.of(new EnumMessage(SessionGeneralError)); + } + return Optional.empty(); + } + + 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 Collection selectClaimsAndLiabilities(Long sessionId) { + ImdgPredicateBuilder pb = registryImdg.predicateBuilder(); + ImdgPredicate prdct = pb.and( + pb.equals("sessionId", sessionId), + pb.or( + pb.and( + pb.equals("registryStatus", RegistryStatus.OK.getKey()), + pb.equals("registryDesignation", RegistryDesignation.T.getKey()) + //pb.equals("registryDesignation", RegistryDesignation.C.getKey()), + //pb.equals("registryDesignation", RegistryDesignation.L.getKey()) + ), + pb.equals("registryDesignation", RegistryDesignation.O.getKey()) + ) + ); + log.debug("searching claims and liabilities by predicate: {}...", prdct); + Collection result = registryImdg.getCollectionObjectsByPredicate(prdct); + log.debug("found {} claims and liabilities", result.size()); + return result; + } + + protected Collection findORegistryBySessionId(Collection rgss) { + List result = rgss + .stream() + .filter(rgs -> RegistryDesignation.O.equalsByKey(rgs.getRegistryDesignation())) + .filter(registry -> Objects.equals(registry.getSettlementDate(), LocalDate.now())) + .filter(registry -> Objects.equals(registry.getValueDate(), registry.getSettlementDate())) + .collect(Collectors.toList()); + log.debug("{} liabilities", result.size()); + return result; + } + + protected Collection findExecutionFondBySessionId(Long sessionId, SessionType sessionType) { + if (sessionType.equals(SessionType.UNIT)) { + List res = new ArrayList<>(); + res.addAll(findExecutionFondBySessionId(sessionId, executionCurrencyImdg)); + res.addAll(findExecutionFondBySessionId(sessionId, executionFondImdg)); + return res; + } else { + Imdg excImdg; + if (sessionType.equals(SessionType.CURR)) { + excImdg = executionCurrencyImdg; + } else { + excImdg = executionFondImdg; + } + return findExecutionFondBySessionId(sessionId, excImdg); + } + } + + protected Collection findExecutionFondBySessionId( + Long sessionId, + Imdg excImdg) { + ImdgPredicateBuilder pb = excImdg.predicateBuilder(); + ImdgPredicate prdct = pb.or( + pb.equals("sessionId", sessionId), + pb.equals("settlementDate", LocalDate.now()) + ); + Collection result = excImdg.getCollectionObjectsByPredicate(prdct) + .stream() + .filter(eFond -> Objects.nonNull(eFond.getSessionId())) + .filter(eFond -> { + if (sessionId.equals(eFond.getSessionId())) return true; + Session session = sessionImdg.getSingleObjectByID(eFond.getSessionId()); + return this.sessionType.equalsByKey(session.getSessionType()); + }) + .toList(); + log.trace("found {} Executions by sessionId={}", result.size(), sessionId); + return result; + } + + protected Collection findExecutionDepositBySessionId(Long sessionId) { + ImdgPredicateBuilder pb = executionDepositImdg.predicateBuilder(); + ImdgPredicate prdct = pb.or( + pb.equals("sessionId", sessionId), + pb.equals("firstLegSettlementDate", LocalDate.now()) + ); + Collection result = executionDepositImdg.getCollectionObjectsByPredicate(prdct); + log.trace("found {} ExecutionDeposit by {}", result.size(), prdct); + return result; + } +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/SdfReceivedAction.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/SdfReceivedAction.java new file mode 100644 index 000000000..beec33333 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/SdfReceivedAction.java @@ -0,0 +1,19 @@ +package ru.spcex.clearing.session.state.action; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.action.Action; +import ru.spcex.clearing.session.stage.TaskType; +import ru.spcex.clearing.session.state.SsnEvent; + +public class SdfReceivedAction implements Action { + private final Logger log = LoggerFactory.getLogger(getClass()); + public static final SdfReceivedAction instance = new SdfReceivedAction(); + + @Override + public void execute(StateContext ctx) { + log.info("event {} flag saving received", ctx.getEvent().name()); + ctx.getExtendedState().getVariables().put(ctx.getEvent().name(), true); + } +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/PaymentsWereCreatedGuard.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/PaymentsWereCreatedGuard.java new file mode 100644 index 000000000..5679fe2a6 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/PaymentsWereCreatedGuard.java @@ -0,0 +1,22 @@ +package ru.spcex.clearing.session.state.guard; + +import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.guard.Guard; +import ru.spcex.clearing.session.stage.TaskType; +import ru.spcex.clearing.session.stage.impl.PaymentInfo; +import ru.spcex.clearing.session.state.DataEnum; +import ru.spcex.clearing.session.state.SsnEvent; + +public class PaymentsWereCreatedGuard implements Guard { + + public static final PaymentsWereCreatedGuard instance = new PaymentsWereCreatedGuard(); + + private PaymentsWereCreatedGuard() { + } + + @Override + public boolean evaluate(StateContext ctx) { + PaymentInfo pmts = ctx.getExtendedState().get(DataEnum.paymentInfo, PaymentInfo.class); + return pmts != null && !pmts.getPaymentInstructions().isEmpty(); + } +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/PaymentsWereNotCreatedGuard.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/PaymentsWereNotCreatedGuard.java new file mode 100644 index 000000000..9497703ff --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/PaymentsWereNotCreatedGuard.java @@ -0,0 +1,22 @@ +package ru.spcex.clearing.session.state.guard; + +import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.guard.Guard; +import ru.spcex.clearing.session.stage.TaskType; +import ru.spcex.clearing.session.stage.impl.PaymentInfo; +import ru.spcex.clearing.session.state.DataEnum; +import ru.spcex.clearing.session.state.SsnEvent; + +public class PaymentsWereNotCreatedGuard implements Guard { + + public static final PaymentsWereNotCreatedGuard instance = new PaymentsWereNotCreatedGuard(); + + private PaymentsWereNotCreatedGuard() { + } + + @Override + public boolean evaluate(StateContext ctx) { + PaymentInfo pmts = ctx.getExtendedState().get(DataEnum.paymentInfo, PaymentInfo.class); + return pmts == null || pmts.getPaymentInstructions().isEmpty(); + } +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/SdfGuard.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/SdfGuard.java new file mode 100644 index 000000000..aa5f1b2f2 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/SdfGuard.java @@ -0,0 +1,45 @@ +package ru.spcex.clearing.session.state.guard; + +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.statemachine.ExtendedState; +import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.guard.Guard; +import ru.spcex.clearing.session.stage.TaskType; +import ru.spcex.clearing.session.state.SsnEvent; +import ru.spcex.platform.enumeration.SdfTable; + +public class SdfGuard implements Guard { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final Function, Set> sdfsSetExtractor; + + public SdfGuard(Function, Set> sdfsSetExtractor) { + this.sdfsSetExtractor = sdfsSetExtractor; + } + + @Override + public boolean evaluate(StateContext ctx) { + ExtendedState state = ctx.getExtendedState(); + Function was = sdf -> state.get(sdf.getKey(), Object.class) != null; + Set sdfs = sdfsSetExtractor.apply(ctx); + boolean everySdfConditionMet = sdfs + .stream() + .allMatch(was::apply); + if (everySdfConditionMet) { + log.info("all of {} conditions were met", sdfs); + sdfs.forEach(sdf -> state.getVariables().remove(sdf.getKey())); + return true; + } else { + log.info("conditions: {}", + sdfs + .stream() + .map(sdf -> "%s: %s".formatted(sdf.getKey(), was.apply(sdf) ? "received" : "waiting")) + .collect(Collectors.joining(";", "[", "]")) + ); + return false; + } + } +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/SdfGuardExtractorAfterAssets.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/SdfGuardExtractorAfterAssets.java new file mode 100644 index 000000000..5df0a12c4 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/SdfGuardExtractorAfterAssets.java @@ -0,0 +1,76 @@ +package ru.spcex.clearing.session.state.guard; + +import java.util.Collections; +import java.util.EnumSet; +import java.util.Set; +import java.util.function.Function; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.statemachine.StateContext; +import ru.spcex.clearing.session.stage.TaskType; +import ru.spcex.clearing.session.stage.impl.PaymentInfo; +import ru.spcex.clearing.session.state.DataEnum; +import ru.spcex.clearing.session.state.SsnEvent; +import ru.spcex.platform.enumeration.SdfTable; +import ru.spcex.platform.enumeration.Section; + +public class SdfGuardExtractorAfterAssets implements Function, Set> { + + private final Logger log = LoggerFactory.getLogger(getClass()); + + public static final SdfGuardExtractorAfterAssets instance = new SdfGuardExtractorAfterAssets(); + + @Override + public Set apply(StateContext ctx) { + Section section = ctx.getExtendedState().get(DataEnum.section, Section.class); + if (section == null) { + throw new IllegalStateException("section is null"); + } + PaymentInfo pmts = ctx.getExtendedState().get(DataEnum.paymentInfo, PaymentInfo.class); + if (pmts == null || pmts.getPaymentInstructions().isEmpty()) { + log.warn("SdfGuard defined, but no DataEnum.paymentInfo present in extended state"); + return Collections.emptySet(); + } + switch (section) { + case MKR -> { + return EnumSet.of( + SdfTable.SDF_04, + SdfTable.SDF_01, + SdfTable.SDF_57 + ); + } + case FOND, MULT -> { + return EnumSet.of( + SdfTable.SDF_04, + SdfTable.SDF_13, + SdfTable.SDF_08, + SdfTable.SDF_21, + SdfTable.SDF_01, + SdfTable.SDF_57 + ); + } + default -> throw new IllegalStateException("unknown wait conditions for section " + section); + } + } + + //public static SessionMonitor waitStep7(Section section) { + // switch (section) { + // case MKR -> { + // return SessionMonitor.create() + // .addCondition(new SdfCondition(SdfTable.SDF_04)) + // .addCondition(new SdfCondition(SdfTable.SDF_01)) + // .addCondition(new SdfCondition(SdfTable.SDF_57)); + // } + // case FOND, MULT -> { + // return SessionMonitor.create() + // .addCondition(new SdfCondition(SdfTable.SDF_04)) + // .addCondition(new SdfCondition(SdfTable.SDF_13)) + // .addCondition(new SdfCondition(SdfTable.SDF_08)) + // .addCondition(new SdfCondition(SdfTable.SDF_21)) + // .addCondition(new SdfCondition(SdfTable.SDF_01)) + // .addCondition(new SdfCondition(SdfTable.SDF_57)); + // } + // default -> throw new IllegalStateException("unknown wait conditions for section " + section); + // } + //} +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/interceptor/SessionStatusChangingInterceptor.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/interceptor/SessionStatusChangingInterceptor.java index 06386496d..5919f2791 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/interceptor/SessionStatusChangingInterceptor.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/interceptor/SessionStatusChangingInterceptor.java @@ -64,6 +64,8 @@ public class SessionStatusChangingInterceptor extends StateMachineInterceptorAda source != null ? source.getId() : "unknown", target.getId() )); + } else if (target.getId().isPseudoStatus()) { + return ctx; } String sessionStatus = target.getId().getKey(); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/listener/MachineMonitoringListener.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/listener/MachineMonitoringListener.java new file mode 100644 index 000000000..f128bdf17 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/listener/MachineMonitoringListener.java @@ -0,0 +1,37 @@ +package ru.spcex.clearing.session.state.listener; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.messaging.Message; +import org.springframework.statemachine.listener.StateMachineListenerAdapter; +import org.springframework.statemachine.state.State; +import org.springframework.statemachine.transition.Transition; +import ru.spcex.clearing.session.stage.TaskType; +import ru.spcex.clearing.session.state.SsnEvent; + +public class MachineMonitoringListener extends StateMachineListenerAdapter { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final String machineId; + + public MachineMonitoringListener(String machineId) { + this.machineId = machineId; + } + + @Override + public void stateEntered(State state) { + TaskType enteredState = state != null ? state.getId() : null; + log.info("LISTENER {} stateEntered: {}", machineId, enteredState); + } + + @Override + public void transitionStarted(Transition transition) { + TaskType source = transition.getSource() != null ? transition.getSource().getId() : null; + TaskType target = transition.getTarget().getId(); + log.info("LISTENER {} transitionStarted: {} => {}", machineId, source, target); + } + + @Override + public void eventNotAccepted(Message event) { + log.info("LISTENER {} eventNotAccepted: {}", machineId, event.getPayload()); + } +} diff --git a/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/StateMachineTest.java b/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/StateMachineTest.java index 85b32b3ce..4a436a165 100644 --- a/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/StateMachineTest.java +++ b/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/StateMachineTest.java @@ -6,12 +6,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.messaging.Message; -import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.config.StateMachineFactory; -import org.springframework.statemachine.support.StateMachineInterceptorAdapter; -import org.springframework.statemachine.transition.Transition; import ru.spcex.clearing.session.teststate.config.Event; import ru.spcex.clearing.session.teststate.config.State; import ru.spcex.clearing.session.teststate.config.TestStateMachineConfig; @@ -40,55 +36,63 @@ class StateMachineTest { @Test void test() { log.debug("test???"); - stateMachine - .getStateMachineAccessor() - .doWithRegion(function -> function.addStateMachineInterceptor( - new StateMachineInterceptorAdapter<>() { - @Override - public Message preEvent(Message message, StateMachine stateMachine) { - Event payload = message.getPayload(); - log.info("INTERCEPTOR preEvent catched event {}", payload); - return super.preEvent(message, stateMachine); - } - - @Override - public void preStateChange(org.springframework.statemachine.state.State state, Message message, Transition transition, StateMachine stateMachine, StateMachine rootStateMachine) { - log.info("INTERCEPTOR preStateChange catched"); - super.preStateChange(state, message, transition, stateMachine, rootStateMachine); - } - - @Override - public void postStateChange(org.springframework.statemachine.state.State state, Message message, Transition transition, StateMachine stateMachine, StateMachine rootStateMachine) { - log.info("INTERCEPTOR postStateChange catched"); - super.postStateChange(state, message, transition, stateMachine, rootStateMachine); - } - - @Override - public StateContext preTransition(StateContext stateContext) { - log.info("INTERCEPTOR preTransition catched"); - org.springframework.statemachine.state.State target = stateContext.getTarget(); - //if (target.getId().equals(State.continueRevise)) { - // IllegalStateException ex = new IllegalStateException("pre transition interceptor exception!!!"); - // stateContext.getStateMachine().setStateMachineError(ex); - // throw ex; - //} - return super.preTransition(stateContext); - } - - @Override - public StateContext postTransition(StateContext stateContext) { - log.info("INTERCEPTOR postTransition catched"); - return super.postTransition(stateContext); - } - - @Override - public Exception stateMachineError(StateMachine stateMachine, Exception exception) { - log.info("INTERCEPTOR error caught!!!"); - return exception; - } - })); +// stateMachine +// .getStateMachineAccessor() +// .doWithRegion(function -> function.addStateMachineInterceptor( +// new StateMachineInterceptorAdapter<>() { +// @Override +// public Message preEvent(Message message, StateMachine stateMachine) { +// Event payload = message.getPayload(); +// log.info("INTERCEPTOR preEvent catched event {}", payload); +// return super.preEvent(message, stateMachine); +// } +// +// @Override +// public void preStateChange(org.springframework.statemachine.state.State state, Message message, Transition transition, StateMachine stateMachine, StateMachine rootStateMachine) { +// log.info("INTERCEPTOR preStateChange catched"); +// super.preStateChange(state, message, transition, stateMachine, rootStateMachine); +// } +// +// @Override +// public void postStateChange(org.springframework.statemachine.state.State state, Message message, Transition transition, StateMachine stateMachine, StateMachine rootStateMachine) { +// log.info("INTERCEPTOR postStateChange catched"); +// super.postStateChange(state, message, transition, stateMachine, rootStateMachine); +// } +// +// @Override +// public StateContext preTransition(StateContext stateContext) { +// log.info("INTERCEPTOR preTransition catched"); +// org.springframework.statemachine.state.State target = stateContext.getTarget(); +// //if (target.getId().equals(State.continueRevise)) { +// // IllegalStateException ex = new IllegalStateException("pre transition interceptor exception!!!"); +// // stateContext.getStateMachine().setStateMachineError(ex); +// // throw ex; +// //} +// return super.preTransition(stateContext); +// } +// +// @Override +// public StateContext postTransition(StateContext stateContext) { +// log.info("INTERCEPTOR postTransition catched"); +// return super.postTransition(stateContext); +// } +// +// @Override +// public Exception stateMachineError(StateMachine stateMachine, Exception exception) { +// log.info("INTERCEPTOR error caught!!!"); +// return exception; +// } +// })); stateMachine.sendEvent(Event.startSession); + try { + Thread.sleep(5000); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + stateMachine.sendEvent(Event.SDF_01); + stateMachine.sendEvent(Event.SDF_04); + stateMachine.sendEvent(Event.SDF_57); try { Thread.sleep(10000); } catch (InterruptedException e) { diff --git a/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/Event.java b/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/Event.java index dd1f29dc5..b7d7b6a33 100644 --- a/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/Event.java +++ b/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/Event.java @@ -2,5 +2,6 @@ package ru.spcex.clearing.session.teststate.config; public enum Event { startSession, continueRevise, + SDF_04, SDF_01, SDF_57, triggerInternal } diff --git a/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/State.java b/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/State.java index 371a485b5..a5790536c 100644 --- a/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/State.java +++ b/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/State.java @@ -5,5 +5,7 @@ public enum State { startRevise, continueRevise, stateWithInternalTransition, - stateAfterInternalTransition, + stateAutoTransitionAfterInternalAutoTransition, + stateWaitingForSdf, + stateAfterSdfReceived, } diff --git a/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/TestStateMachineConfig.java b/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/TestStateMachineConfig.java index 16a37fe67..017a2443c 100644 --- a/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/TestStateMachineConfig.java +++ b/clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/TestStateMachineConfig.java @@ -1,5 +1,8 @@ package ru.spcex.clearing.session.teststate.config; +import java.util.ArrayList; +import java.util.EnumSet; +import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -8,6 +11,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.core.task.TaskExecutor; import org.springframework.messaging.Message; import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.action.Action; import org.springframework.statemachine.config.EnableStateMachineFactory; import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter; import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer; @@ -42,7 +46,9 @@ public class TestStateMachineConfig extends EnumStateMachineConfigurerAdapter { log.info("ACTION without an event"); -// context.getStateMachine().sendEvent(Event.triggerInternal); +// context.getStateMachine().sendEvent(Event.triggerInternal); }) .and() .withInternal() @@ -88,9 +94,61 @@ public class TestStateMachineConfig extends EnumStateMachineConfigurerAdapter log.info("ACTION after internal transition")) - ; + .target(State.stateAutoTransitionAfterInternalAutoTransition) + .action(ctx -> { + log.info("ACTION after internal transition"); + List list = new ArrayList<>(); + list.add("fasdfasdf"); + ctx.getExtendedState().getVariables().put("payments", list); + }) + .and() + .withExternal() + .source(State.stateAutoTransitionAfterInternalAutoTransition) + .target(State.stateAfterSdfReceived) + .action(ctx -> log.info("ACTION stateAutoTransitionAfterInternalAutoTransition to stateAfterSdfReceived without waiting")) + .guard(ctx -> { + List list = ctx.getExtendedState().get("payments", List.class); + return list != null && list.isEmpty(); + }) + .and() + .withExternal() + .source(State.stateAutoTransitionAfterInternalAutoTransition) + .target(State.stateWaitingForSdf) + .action(ctx -> log.info("ACTION stateAutoTransitionAfterInternalAutoTransition to stateWaitingForSdf")) + .guard(ctx -> { + List list = ctx.getExtendedState().get("payments", List.class); + return list != null && !list.isEmpty(); + }) + .and() + .withExternal() + .source(State.stateWaitingForSdf) + .target(State.stateAfterSdfReceived) + .guard(ctx -> { + Boolean sdf01Received = ctx.getExtendedState().get(Event.SDF_01.name(), Boolean.class); + Boolean sdf04Received = ctx.getExtendedState().get(Event.SDF_04.name(), Boolean.class); + Boolean sdf57Received = ctx.getExtendedState().get(Event.SDF_57.name(), Boolean.class); + log.info("GUARD received sdfs: sdf01={}, sdf04={}, sdf57={}", + sdf01Received, sdf04Received, sdf57Received); + return Boolean.TRUE.equals(sdf01Received) + && Boolean.TRUE.equals(sdf04Received) + && Boolean.TRUE.equals(sdf57Received); + }); + + for (var event : EnumSet.of(Event.SDF_01, Event.SDF_04, Event.SDF_57)) { + transitions + .withInternal() + .source(State.stateWaitingForSdf) + .event(event) + .action(eventReceivedAction()); + } + } + + + public Action eventReceivedAction() { + return ctx -> { + log.info("event {} flag saving received", ctx.getEvent().name()); + ctx.getExtendedState().getVariables().put(ctx.getEvent().name(), true); + }; } @Override