From 4f88c714bc2eb0c2c5b198de3502ca6182c47a38 Mon Sep 17 00:00:00 2001 From: ialbert Date: Mon, 7 Oct 2024 19:36:53 +0300 Subject: [PATCH] FinalMkrSession state machine integration --- .../FinalSessionStateMachineConfig.java | 3 ++- .../clearing/service/EventsReceiver.java | 24 +++++++++++++++++-- .../session/stage/SessionManager.java | 4 ++++ .../clearing/session/state/ActionHeader.java | 2 +- .../clearing/session/state/SsnEvent.java | 3 --- .../state/action/FinishingSessionAction.java | 11 +++++++-- .../factory/SessionStateMachineWrapper.java | 20 ++++++++++++---- 7 files changed, 54 insertions(+), 13 deletions(-) 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 082416f80..f66365039 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 @@ -99,6 +99,7 @@ public class FinalSessionStateMachineConfig this.endStageNotificationAction = endStageNotificationAction; this.finishingSessionAction.setSessionType(SESSION_TYPE); this.finishingSessionAction.setSection(SECTION); + this.finishingSessionAction.setPr("3"); //stages settings: ImdgPredicateBuilder rgsPrctBuilder = rgsImdg.predicateBuilder(); @@ -149,7 +150,7 @@ public class FinalSessionStateMachineConfig public void configure(StateMachineTransitionConfigurer transitions) throws Exception { transitions .withExternal() - .event(SsnEvent.Sdf57Processed) + .event(SsnEvent.SDF_57) .source(TaskType.StartRevise).target(TaskType.StartRevisePart1) .action(reviseStage1Action) .and() diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java index a947ca486..7a007b02a 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java @@ -1,6 +1,8 @@ package ru.spcex.clearing.service; import java.time.LocalTime; +import java.util.Collection; +import java.util.stream.Stream; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; @@ -50,6 +52,8 @@ import ru.spcex.clearing.session.stage.SessionTerminator; import ru.spcex.clearing.session.stage.TaskType; import ru.spcex.clearing.session.stage.UnitedSession; import ru.spcex.clearing.session.stage.impl.BalanceRevise; +import ru.spcex.clearing.session.state.SsnEvent; +import ru.spcex.clearing.session.state.factory.SessionStateMachineWrapper; import ru.spcex.clearing.statement.StatementServiceV2; import ru.spcex.platform.enumeration.Task; import ru.spcex.platform.utils.enumeration.IMessageResolver; @@ -72,6 +76,7 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean private final UnitedSession unitedSession; private final ReturnDepositSession returnDepositSession; private final SessionManager sessionManager; + private final SessionStateMachineWrapper stateMachineWrapper; private final Sdf06Executor sdf06Executor; private final Sdf10Executor sdf10Executor; private final BalanceRevise balanceRevise; @@ -91,7 +96,7 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean PrimaryAuctionBnSession primaryAuctionBnSession, SecondaryAuctionT0Session secondaryAuctionT0Session, CurrencySession currencySession, - PrimaryAuctionB0Session primaryAuctionB0Session, PrimaryAuctionT0Session primaryAuctionT0Session, IntermediateMkrSession intermediateMkrSession, FinalMkrSession finalMkrSession, UnitedSession unitedSession, ReturnDepositSession returnDepositSession, SessionManager sessionManager, + PrimaryAuctionB0Session primaryAuctionB0Session, PrimaryAuctionT0Session primaryAuctionT0Session, IntermediateMkrSession intermediateMkrSession, FinalMkrSession finalMkrSession, UnitedSession unitedSession, ReturnDepositSession returnDepositSession, SessionManager sessionManager, SessionStateMachineWrapper stateMachineWrapper, Sdf06Executor sdf06Executor, Sdf10Executor sdf10Executor, BalanceRevise balanceRevise, Sdf05Sender sdf05Sender, StatementServiceV2 statementService, SessionTerminator sessionTerminator, PaymentInstructionOutboundService pmtOutboundService, ExecutionCurrencyComponent execCurrUpload, ExecutionDepositComponent execDepUpload, ExecutionFondComponent execFondUpload) { super(kafkaQueue, kafkaResponseQueue); @@ -109,6 +114,7 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean this.unitedSession = unitedSession; this.returnDepositSession = returnDepositSession; this.sessionManager = sessionManager; + this.stateMachineWrapper = stateMachineWrapper; this.sdf06Executor = sdf06Executor; this.sdf10Executor = sdf10Executor; this.balanceRevise = balanceRevise; @@ -146,6 +152,7 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean .setFunction(r -> { try { sessionManager.defineAndStartSession(r); + stateMachineWrapper.defineAndStartSession(r); } catch (ValidationException ve) { return makeErrorResponse(r, ve); } @@ -167,9 +174,22 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean currencySession.continueSession(req); primaryAuctionB0Session.continueSession(req); intermediateMkrSession.continueSession(req); - finalMkrSession.continueSession(req); + //finalMkrSession.continueSession(req); unitedSession.continueSession(req); returnDepositSession.continueSession(req); + SessionContinueEvent payload = req.getRequestPayload(); + Stream + .ofNullable(payload.getSdfType()) + .flatMap(Collection::stream) + .forEach(sdf -> { + try { + SsnEvent ssnEvent = SsnEvent.valueOf(sdf.name()); + stateMachineWrapper.sendEvent(ssnEvent); + } catch (IllegalArgumentException e) { + log.info("sdf {} event skipped for state machine", sdf); + } + }); + ; }) .forDestination(Consts.CONTINUE_SESSION_BN_FIRST_PART, callbacks::put); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/SessionManager.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/SessionManager.java index d18415d74..f280a69bf 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/SessionManager.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/SessionManager.java @@ -72,6 +72,10 @@ public class SessionManager { log.warn("Unknown sessionType: {} or section: {}", section, sessionType); return; } + if (SessionType.FINL.equals(sessionType)) { + log.info("falling back to spring state machine"); + return; + } BaseRequest baseRequest = new BaseRequest<>(); AbstractSession session = sessionByType(sessionType); if (session != null) { 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 d220b2404..78b1c0724 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, pr; + sdf56FromTime; } 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 f772b8cce..57a81b4f8 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 @@ -1,9 +1,6 @@ package ru.spcex.clearing.session.state; 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/FinishingSessionAction.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/FinishingSessionAction.java index ef5cc3b01..1224fcb75 100644 --- 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 @@ -35,7 +35,6 @@ 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; @@ -74,6 +73,7 @@ public class FinishingSessionAction extends AbstractSessionActionForOkErrorHandl private final Sdf14Sender sdf14Sender; private SessionType sessionType; private Section section; + private String pr; @Autowired @@ -98,6 +98,11 @@ public class FinishingSessionAction extends AbstractSessionActionForOkErrorHandl public void setSection(Section section) { this.section = section; } + + public void setPr(String pr) { + this.pr = pr; + } + public void setSessionType(SessionType sessionType) { this.sessionType = sessionType; } @@ -106,7 +111,9 @@ public class FinishingSessionAction extends AbstractSessionActionForOkErrorHandl @Override protected void actualExecute(StateContext ctx) { Long sessionId = ctx.getExtendedState().get(DataEnum.sessionId, Long.class); - String pr = (String) ctx.getMessageHeader(ActionHeader.pr); + if (pr == null) { + throw new IllegalStateException("'pr' not set up"); + } Instant now = Instant.now(); //установка CLRD для обработанных регистров Collection allLiabilitiesAndOKClaims = selectClaimsAndLiabilities(sessionId); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/factory/SessionStateMachineWrapper.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/factory/SessionStateMachineWrapper.java index b04050076..b425cc6b4 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/factory/SessionStateMachineWrapper.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/factory/SessionStateMachineWrapper.java @@ -2,7 +2,6 @@ package ru.spcex.clearing.session.state.factory; import java.util.AbstractMap; import java.util.Map; -import java.util.NoSuchElementException; import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -57,6 +56,10 @@ public class SessionStateMachineWrapper { return; } StateMachineFactory factory = getByType(sessionType); + if (factory == null) { + log.info("no spring state machine factory for {}", sessionType); + return; + } if (this.currentSession != null) { Session session = currentSession .getExtendedState() @@ -82,6 +85,15 @@ public class SessionStateMachineWrapper { } } + public synchronized void sendEvent(SsnEvent event) { + if (this.currentSession != null) { + log.info("sending an event to session: {}", event); + this.currentSession.sendEvent(event); + } else { + log.warn("there is no currently active session"); + } + } + private synchronized void clearCurrentMachine() { this.currentSession.stop(); this.currentSession = null; @@ -89,9 +101,9 @@ public class SessionStateMachineWrapper { private StateMachineFactory getByType(SessionType sessionType) { StateMachineFactory factory = factories.get(sessionType); - if (factory == null) { - throw new NoSuchElementException("no state machine for session: " + sessionType); - } + //if (factory == null) { + // throw new NoSuchElementException("no state machine for session: " + sessionType); + //} return factory; }