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

FinalMkrSession state machine integration

Browse source
This commit is contained in:
ialbert 2024-10-07 19:36:53 +03:00
parent f98ee3f58e
commit 4f88c714bc
7 changed files with 54 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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<TaskType, SsnEvent> transitions) throws Exception {
transitions
.withExternal()
.event(SsnEvent.Sdf57Processed)
.event(SsnEvent.SDF_57)
.source(TaskType.StartRevise).target(TaskType.StartRevisePart1)
.action(reviseStage1Action)
.and()

24
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java
View file

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

4
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/SessionManager.java
View file

@ -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) {

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, pr;
sdf56FromTime;
}

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

@ -1,9 +1,6 @@
package ru.spcex.clearing.session.state;
public enum SsnEvent {
Revise,
Sdf57Processed,
SdfReceived,
SDF_01,
SDF_57,
SDF_04,

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

@ -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<TaskType, SsnEvent> 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<Registry> allLiabilitiesAndOKClaims = selectClaimsAndLiabilities(sessionId);

20
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/factory/SessionStateMachineWrapper.java
View file

@ -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<TaskType, SsnEvent> 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<TaskType, SsnEvent> getByType(SessionType sessionType) {
StateMachineFactory<TaskType, SsnEvent> 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;
}