treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

FinalMkrSession spring state machine

Browse source
This commit is contained in:
ialbert 2024-10-07 19:17:23 +03:00
parent 05d1135ae2
commit 68496730c6
19 changed files with 940 additions and 93 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

96
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/state_machine_2/specific/FinalSessionStateMachineConfig.java
View file

@ -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<TaskType, SsnEvent> {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Imdg<Session> sessionImdg;
private final Imdg<Registry> 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<List<String>> 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<Registry> 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<TaskType, SsnEvent> 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<TaskType, SsnEvent> config) throws Exception {
StateMachineListenerAdapter<TaskType, SsnEvent> loggingChangeStateListener = new StateMachineListenerAdapter<>() {
@Override
public void stateEntered(State<TaskType, SsnEvent> state) {
TaskType enteredState = state != null ? state.getId() : null;
log.info("State entered: {}", enteredState);
}
};
StateMachineListenerAdapter<TaskType, SsnEvent> loggingChangeStateListener
= new MachineMonitoringListener(SESSION_TYPE.getKey());
config
.withConfiguration()
.machineId(SessionType.FINL.getKey())
.machineId(SESSION_TYPE.getKey())
.listener(loggingChangeStateListener)
;
}

8
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/TaskType.java
View file

@ -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");
}
}

2
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/ActionHeader.java
View file

@ -1,5 +1,5 @@
package ru.spcex.clearing.session.state;
public enum ActionHeader {
sdf56FromTime;
sdf56FromTime, pr;
}

1
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/DataEnum.java
View file

@ -3,6 +3,7 @@ package ru.spcex.clearing.session.state;
public enum DataEnum {
sessionId, //Long
sessionType, //Session
section,
session, //Session
dealsPrepared, //List<ExecutionCommon>
counterPartyId, //Long

3
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/SsnEvent.java
View file

@ -4,4 +4,7 @@ public enum SsnEvent {
Revise,
Sdf57Processed,
SdfReceived,
SDF_01,
SDF_57,
SDF_04,
}

22
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/CreateSessionAction.java
View file

@ -17,27 +17,43 @@ import ru.spcex.platform.imdg.api.Imdg;
public class CreateSessionAction extends AbstractSessionActionForOkErrorHandling {
private final Imdg<Session> sessionImdg;
private final Logger log = LoggerFactory.getLogger(CreateSessionAction.class);
private SessionType sessionType;
private Section section;
public CreateSessionAction(Imdg<Session> 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<TaskType, SsnEvent> 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);
}
}

123
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/EndStageNotificationAction.java Normal file
View file

@ -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<Registry> registryImdg;
private final Imdg<Session> 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<TaskType, SsnEvent> 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<Registry> forRegistries = selectRegistry();
Collection<Long> 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<EnumMessage> 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<Registry> selectRegistry() {
String registrySQL = "registryStatus='" + RegistryStatus.OK.getKey() + "'";
Collection<Registry> result = registryImdg.getCollectionObjectsBySQL(registrySQL);
log.trace("Selected {} registry's by sql: {}", result.size(), registrySQL);
return result;
}
protected Optional<EnumMessage> 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<EnumMessage> 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();
}
}

376
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/FinishingSessionAction.java Normal file
View file

@ -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<Registry> registryImdg;
private final Imdg<ExecutionFond> executionFondImdg;
private final Imdg<ExecutionDeposit> executionDepositImdg;
private final Imdg<ExecutionCurrency> executionCurrencyImdg;
private final Imdg<Session> sessionImdg;
private final Imdg<SDf05> 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<TaskType, SsnEvent> ctx) {
Long sessionId = ctx.getExtendedState().get(DataEnum.sessionId, Long.class);
String pr = (String) ctx.getMessageHeader(ActionHeader.pr);
Instant now = Instant.now();
//установка CLRD для обработанных регистров
Collection<Registry> allLiabilitiesAndOKClaims = selectClaimsAndLiabilities(sessionId);
Collection<Registry> claimsAndLiabilities = allLiabilitiesAndOKClaims
.stream()
.filter(rgs -> RegistryDesignation.T.equalsByKey(rgs.getRegistryDesignation())
|| RegistryStatus.OK.equalsByKey(rgs.getRegistryStatus()))
.toList();
{
Map<Long, Registry> 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<Registry> oRegs = findORegistryBySessionId(allLiabilitiesAndOKClaims);
Map<Long, RegistryStatus> 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<Long, CoverageStatus> 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<Long> 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<Long, ExecutionDeposit> depos = new HashMap<>();
Collection<ExecutionDeposit> 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<? extends ExecutionCommon> executions = findExecutionFondBySessionId(sessionId, sessionType);
Map<Long, ExecutionCurrency> execCurr = new HashMap<>();
Map<Long, ExecutionFond> 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<Registry> forRegistries = selectRegistry(sessionId);
log.debug("Found {} registries for sessionId={}", forRegistries.size(), sessionId);
Optional<EnumMessage> 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<EnumMessage> 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<EnumMessage> 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<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 Collection<Registry> 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<Registry> result = registryImdg.getCollectionObjectsByPredicate(prdct);
log.debug("found {} claims and liabilities", result.size());
return result;
}
protected Collection<Registry> findORegistryBySessionId(Collection<Registry> rgss) {
List<Registry> 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<? extends ExecutionCommon> findExecutionFondBySessionId(Long sessionId, SessionType sessionType) {
if (sessionType.equals(SessionType.UNIT)) {
List<ExecutionCommon> res = new ArrayList<>();
res.addAll(findExecutionFondBySessionId(sessionId, executionCurrencyImdg));
res.addAll(findExecutionFondBySessionId(sessionId, executionFondImdg));
return res;
} else {
Imdg<? extends ExecutionCommon> excImdg;
if (sessionType.equals(SessionType.CURR)) {
excImdg = executionCurrencyImdg;
} else {
excImdg = executionFondImdg;
}
return findExecutionFondBySessionId(sessionId, excImdg);
}
}
protected Collection<? extends ExecutionCommon> findExecutionFondBySessionId(
Long sessionId,
Imdg<? extends ExecutionCommon> excImdg) {
ImdgPredicateBuilder pb = excImdg.predicateBuilder();
ImdgPredicate prdct = pb.or(
pb.equals("sessionId", sessionId),
pb.equals("settlementDate", LocalDate.now())
);
Collection<? extends ExecutionCommon> 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<ExecutionDeposit> findExecutionDepositBySessionId(Long sessionId) {
ImdgPredicateBuilder pb = executionDepositImdg.predicateBuilder();
ImdgPredicate prdct = pb.or(
pb.equals("sessionId", sessionId),
pb.equals("firstLegSettlementDate", LocalDate.now())
);
Collection<ExecutionDeposit> result = executionDepositImdg.getCollectionObjectsByPredicate(prdct);
log.trace("found {} ExecutionDeposit by {}", result.size(), prdct);
return result;
}
}

19
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/SdfReceivedAction.java Normal file
View file

@ -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<TaskType, SsnEvent> {
private final Logger log = LoggerFactory.getLogger(getClass());
public static final SdfReceivedAction instance = new SdfReceivedAction();
@Override
public void execute(StateContext<TaskType, SsnEvent> ctx) {
log.info("event {} flag saving received", ctx.getEvent().name());
ctx.getExtendedState().getVariables().put(ctx.getEvent().name(), true);
}
}

22
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/PaymentsWereCreatedGuard.java Normal file
View file

@ -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<TaskType, SsnEvent> {
public static final PaymentsWereCreatedGuard instance = new PaymentsWereCreatedGuard();
private PaymentsWereCreatedGuard() {
}
@Override
public boolean evaluate(StateContext<TaskType, SsnEvent> ctx) {
PaymentInfo pmts = ctx.getExtendedState().get(DataEnum.paymentInfo, PaymentInfo.class);
return pmts != null && !pmts.getPaymentInstructions().isEmpty();
}
}

22
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/PaymentsWereNotCreatedGuard.java Normal file
View file

@ -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<TaskType, SsnEvent> {
public static final PaymentsWereNotCreatedGuard instance = new PaymentsWereNotCreatedGuard();
private PaymentsWereNotCreatedGuard() {
}
@Override
public boolean evaluate(StateContext<TaskType, SsnEvent> ctx) {
PaymentInfo pmts = ctx.getExtendedState().get(DataEnum.paymentInfo, PaymentInfo.class);
return pmts == null || pmts.getPaymentInstructions().isEmpty();
}
}

45
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/SdfGuard.java Normal file
View file

@ -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<TaskType, SsnEvent> {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Function<StateContext<TaskType, SsnEvent>, Set<SdfTable>> sdfsSetExtractor;
public SdfGuard(Function<StateContext<TaskType, SsnEvent>, Set<SdfTable>> sdfsSetExtractor) {
this.sdfsSetExtractor = sdfsSetExtractor;
}
@Override
public boolean evaluate(StateContext<TaskType, SsnEvent> ctx) {
ExtendedState state = ctx.getExtendedState();
Function<SdfTable, Boolean> was = sdf -> state.get(sdf.getKey(), Object.class) != null;
Set<SdfTable> 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;
}
}
}

76
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/guard/SdfGuardExtractorAfterAssets.java Normal file
View file

@ -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<StateContext<TaskType, SsnEvent>, Set<SdfTable>> {
private final Logger log = LoggerFactory.getLogger(getClass());
public static final SdfGuardExtractorAfterAssets instance = new SdfGuardExtractorAfterAssets();
@Override
public Set<SdfTable> apply(StateContext<TaskType, SsnEvent> 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);
// }
//}
}

2
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/interceptor/SessionStatusChangingInterceptor.java
View file

@ -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();

37
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/listener/MachineMonitoringListener.java Normal file
View file

@ -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<TaskType, SsnEvent> {
private final Logger log = LoggerFactory.getLogger(getClass());
private final String machineId;
public MachineMonitoringListener(String machineId) {
this.machineId = machineId;
}
@Override
public void stateEntered(State<TaskType, SsnEvent> state) {
TaskType enteredState = state != null ? state.getId() : null;
log.info("LISTENER {} stateEntered: {}", machineId, enteredState);
}
@Override
public void transitionStarted(Transition<TaskType, SsnEvent> 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<SsnEvent> event) {
log.info("LISTENER {} eventNotAccepted: {}", machineId, event.getPayload());
}
}

106
clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/StateMachineTest.java
View file

@ -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<Event> preEvent(Message<Event> message, StateMachine<State, Event> 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, Event> state, Message<Event> message, Transition<State, Event> transition, StateMachine<State, Event> stateMachine, StateMachine<State, Event> rootStateMachine) {
log.info("INTERCEPTOR preStateChange catched");
super.preStateChange(state, message, transition, stateMachine, rootStateMachine);
}
@Override
public void postStateChange(org.springframework.statemachine.state.State<State, Event> state, Message<Event> message, Transition<State, Event> transition, StateMachine<State, Event> stateMachine, StateMachine<State, Event> rootStateMachine) {
log.info("INTERCEPTOR postStateChange catched");
super.postStateChange(state, message, transition, stateMachine, rootStateMachine);
}
@Override
public StateContext<State, Event> preTransition(StateContext<State, Event> stateContext) {
log.info("INTERCEPTOR preTransition catched");
org.springframework.statemachine.state.State<State, Event> 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<State, Event> postTransition(StateContext<State, Event> stateContext) {
log.info("INTERCEPTOR postTransition catched");
return super.postTransition(stateContext);
}
@Override
public Exception stateMachineError(StateMachine<State, Event> stateMachine, Exception exception) {
log.info("INTERCEPTOR error caught!!!");
return exception;
}
}));
// stateMachine
// .getStateMachineAccessor()
// .doWithRegion(function -> function.addStateMachineInterceptor(
// new StateMachineInterceptorAdapter<>() {
// @Override
// public Message<Event> preEvent(Message<Event> message, StateMachine<State, Event> 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, Event> state, Message<Event> message, Transition<State, Event> transition, StateMachine<State, Event> stateMachine, StateMachine<State, Event> rootStateMachine) {
// log.info("INTERCEPTOR preStateChange catched");
// super.preStateChange(state, message, transition, stateMachine, rootStateMachine);
// }
//
// @Override
// public void postStateChange(org.springframework.statemachine.state.State<State, Event> state, Message<Event> message, Transition<State, Event> transition, StateMachine<State, Event> stateMachine, StateMachine<State, Event> rootStateMachine) {
// log.info("INTERCEPTOR postStateChange catched");
// super.postStateChange(state, message, transition, stateMachine, rootStateMachine);
// }
//
// @Override
// public StateContext<State, Event> preTransition(StateContext<State, Event> stateContext) {
// log.info("INTERCEPTOR preTransition catched");
// org.springframework.statemachine.state.State<State, Event> 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<State, Event> postTransition(StateContext<State, Event> stateContext) {
// log.info("INTERCEPTOR postTransition catched");
// return super.postTransition(stateContext);
// }
//
// @Override
// public Exception stateMachineError(StateMachine<State, Event> 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) {

1
clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/Event.java
View file

@ -2,5 +2,6 @@ package ru.spcex.clearing.session.teststate.config;
public enum Event {
startSession, continueRevise,
SDF_04, SDF_01, SDF_57,
triggerInternal
}

4
clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/State.java
View file

@ -5,5 +5,7 @@ public enum State {
startRevise,
continueRevise,
stateWithInternalTransition,
stateAfterInternalTransition,
stateAutoTransitionAfterInternalAutoTransition,
stateWaitingForSdf,
stateAfterSdfReceived,
}

68
clearing-parent/clearing-service/src/test/java/ru/spcex/clearing/session/teststate/config/TestStateMachineConfig.java
View file

@ -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<St
.state(State.startRevise)
.state(State.continueRevise, Event.triggerInternal)
.state(State.stateWithInternalTransition)
.state(State.stateAfterInternalTransition)
.state(State.stateAutoTransitionAfterInternalAutoTransition)
.state(State.stateWaitingForSdf)
.state(State.stateAfterSdfReceived)
;
}
@ -73,7 +79,7 @@ public class TestStateMachineConfig extends EnumStateMachineConfigurerAdapter<St
.target(State.stateWithInternalTransition)
.action(context -> {
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<St
.withExternal()
.source(State.stateWithInternalTransition)
.timerOnce(1)
.target(State.stateAfterInternalTransition)
.action(ctx -> log.info("ACTION after internal transition"))
;
.target(State.stateAutoTransitionAfterInternalAutoTransition)
.action(ctx -> {
log.info("ACTION after internal transition");
List<Object> 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<State, Event> eventReceivedAction() {
return ctx -> {
log.info("event {} flag saving received", ctx.getEvent().name());
ctx.getExtendedState().getVariables().put(ctx.getEvent().name(), true);
};
}
@Override