From b81559002f79ec060bf89f88ac3d120b02daf7cd Mon Sep 17 00:00:00 2001 From: ialbert Date: Mon, 19 Jun 2023 16:55:11 +0300 Subject: [PATCH] =?UTF-8?q?=D0=BE=D0=B1=D1=80=D0=B0=D0=B1=D0=BE=D1=82?= =?UTF-8?q?=D0=BA=D0=B0=20SDF06=20=D0=B2=D0=B7=D0=B0=D0=B8=D0=BC=D0=BE?= =?UTF-8?q?=D0=B4=D0=B5=D0=B9=D1=81=D1=82=D0=B2=D0=B8=D0=B5=20=D1=81=20gat?= =?UTF-8?q?eway?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../clearing/service/EventsReceiver.java | 16 ++- .../service/executors/Sdf06Executor.java | 127 ++++++++++++++---- .../logic/stages/DbfImportKafkaMessenger.java | 22 +-- .../platform/messaging/domain/Consts.java | 4 +- .../AssetOperationApprovalRequest.java | 18 +++ .../cud/gateway/SingleAssetResponse.java | 26 ++++ 6 files changed, 176 insertions(+), 37 deletions(-) create mode 100644 platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java create mode 100644 platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/SingleAssetResponse.java 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 dcf7b58e7..abee9dbcc 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 @@ -7,12 +7,15 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.CreateRegistryRequest; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.Sdf04Request; import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonIdRequest; +import ru.spcex.clearing.platform.messaging.domain.cud.gateway.AssetOperationApprovalRequest; import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; import ru.spcex.clearing.platform.messaging.domain.cud.utilities.STradesImportedRequest; import ru.spcex.clearing.platform.messaging.service.QueueConsumer; +import ru.spcex.clearing.service.executors.Sdf06Executor; import ru.spcex.clearing.session.stage.*; import ru.spcex.platform.enumeration.Task; @@ -31,6 +34,7 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean { private final FinalMkrSession finalMkrSession; private final ReturnDepositSession returnDepositSession; private final SessionManager sessionManager; + private final Sdf06Executor sdf06Executor; @Autowired public EventsReceiver(Consumer kafkaQueue, @@ -38,7 +42,8 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean { RegistryService registryService, PrimaryAuctionBnSession primaryAuctionBnSession, SecondaryAuctionT0Session secondaryAuctionT0Session, - PrimaryAuctionB0Session primaryAuctionB0Session, PrimaryAuctionT0Session primaryAuctionT0Session, IntermediateMkrSession intermediateMkrSession, FinalMkrSession finalMkrSession, ReturnDepositSession returnDepositSession, SessionManager sessionManager) { + PrimaryAuctionB0Session primaryAuctionB0Session, PrimaryAuctionT0Session primaryAuctionT0Session, IntermediateMkrSession intermediateMkrSession, FinalMkrSession finalMkrSession, ReturnDepositSession returnDepositSession, SessionManager sessionManager, + Sdf06Executor sdf06Executor) { super(kafkaQueue); this.clearingService = clearingService; this.registryService = registryService; @@ -50,6 +55,7 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean { this.finalMkrSession = finalMkrSession; this.returnDepositSession = returnDepositSession; this.sessionManager = sessionManager; + this.sdf06Executor = sdf06Executor; } @Override @@ -86,6 +92,14 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean { }) .forDestination(Consts.CONTINUE_SESSION_BN_FIRST_PART, callbacks::put); + callback(StatementRequest.class) + .setConsumer(sdf06Executor::createStatementAndSendGatewayCommand) + .forDestination(Consts.STATEMENT_PROCESS_SDF06, callbacks::put); + + callback(AssetOperationApprovalRequest.class) + .setConsumer(sdf06Executor::processGatewayResponse) + .forDestination(Consts.ASSET_OPERATION_APPROVAL, callbacks::put); + callback(STradesImportedRequest.class) .setConsumer(event -> clearingService.executeSTrade()) .forDestination(S_TRADES_IMPORTED, callbacks::put); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java index c308ec8f8..46dc9254b 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java @@ -2,6 +2,7 @@ package ru.spcex.clearing.service.executors; 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.stereotype.Service; import ru.clearing.classes.statics.data.account.Account; @@ -12,14 +13,16 @@ import ru.clearing.classes.statics.data.instrument.issue.FixedIncomeSecurity; import ru.clearing.classes.statics.data.registry.Registry; import ru.clearing.classes.statics.data.registry.TradingClearingRegistry; import ru.clearing.classes.statics.data.sdf.SDf06; -import ru.clearing.classes.statics.data.sdf.SDf09; +import ru.clearing.classes.statics.data.sdf.SDf07; import ru.clearing.classes.statics.data.statement.Statement; import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import ru.spcex.clearing.platform.messaging.domain.Consts; import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationRequest; +import ru.spcex.clearing.platform.messaging.domain.cud.gateway.AssetOperationApprovalRequest; +import ru.spcex.clearing.platform.messaging.domain.cud.gateway.SingleAssetResponse; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; -import ru.spcex.clearing.service.model.Result; import ru.spcex.clearing.service.validation.ValidationStored; import ru.spcex.platform.enumeration.*; import ru.spcex.platform.imdg.api.Imdg; @@ -39,7 +42,7 @@ import java.util.Optional; import java.util.function.Function; @Service -public class Sdf06Executor extends AbstractExecutor { +public class Sdf06Executor { private final Logger log = LoggerFactory.getLogger(getClass()); private final Imdg statementImdg; @@ -50,13 +53,16 @@ public class Sdf06Executor extends AbstractExecutor { private final Imdg depoAccountImdg; private final Imdg tradingClearingRegistryImdg; private final Imdg companyImdg; - private final Imdg sdf09Imdg; + private final Imdg sdf06Imdg; + private final Imdg sdf07Imdg; private final IMessageResolver messageResolver; private final Function sDf06Validator; private final KafkaSender kafkaSender; + //not thread safe: used in EventReceiver single thread executor + private Long sdf06GroupId; - //@Qualifier("sdf08ValidatorNew") Function sDf08Validator + @Autowired public Sdf06Executor(ImdgProvider imdgProvider, IMessageResolver messageResolver, @Qualifier("sdf06ValidatorNew") Function sDf06Validator, KafkaSender kafkaSender) { @@ -68,30 +74,19 @@ public class Sdf06Executor extends AbstractExecutor { this.depoAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_DepoAccount, DepoAccount.class); this.tradingClearingRegistryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class); this.companyImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class); - this.sdf09Imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf09, SDf09.class); this.messageResolver = messageResolver; this.sDf06Validator = sDf06Validator; this.kafkaSender = kafkaSender; + this.sdf06Imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf06, SDf06.class); + this.sdf07Imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf07, SDf07.class); } - @Override - public String exportTableName() { - return "DF-06"; - } - - @Override - public boolean isNeedToSendCommand() { - return false; - } - - @Override - public void sendCommand(KafkaSender kafkaSender, Result result) { - } - - public Result execute(Collection sdfs, StatementRequest statementRequest) { - Result result = new Result(); - Long generationIdForGroup = imdgProvider.getImdgIdGenerator().nextId(); - result.setGenerationId(generationIdForGroup); + public void createStatementAndSendGatewayCommand(BaseRequest systemRequest) { + Imdg sdfImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf06, SDf06.class); + Long groupId = systemRequest.getRequestPayload().getGroupId(); + Collection sdfs = sdfImdg.getCollectionObjectsByFieldValues(Map.of( + "generationId", groupId + )); Collection requests = new ArrayList<>(); for (SDf06 sDf06 : sdfs) { IValidator validator = sDf06Validator.apply(sDf06); @@ -100,6 +95,7 @@ public class Sdf06Executor extends AbstractExecutor { //todo если ошибка про нехватку баланса - делаем statement с ошибочным статусом //todo 2 делаем ошибочный SDF07 log.debug("Validation error: {}", messageResolver.resolve(err.get())); + createErrorSdf07(null, sDf06.getId(), groupId, null); continue; } Long companyId = validator.getStored(ValidationStored.Sdf06CompanyId); @@ -121,10 +117,89 @@ public class Sdf06Executor extends AbstractExecutor { log.debug("Statement created: {}", stmt.getId()); requests.add(requestFromStatement(stmt, company.getTradingCode(), tcr.getCode())); } - requests.forEach(r -> kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, r)); - return result; + sdf06GroupId = groupId; + kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, requests); } + public void processGatewayResponse(BaseRequest req) { + for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) { + //получаем запрос для текущей группы sdf06 + //находим группу + Long statementId = gatewayMsg.getStatementId(); + Statement stmt = statementImdg.getSingleObjectByID(statementId); + if (stmt == null) { + log.error("Statement.id {} not found", statementId); + return; + } + + Long sdf06Id = stmt.getInSDfId(); + SDf06 sdf06 = sdf06Imdg.getSingleObjectByID(sdf06Id); + if (sdf06 == null) { + log.error("Sdf06.id {} not found by statement.id {}", sdf06Id, statementId); + return; + } + + Long groupId = sdf06.getGenerationId(); + //сверяем группу SDF06 пришедшего запроса с ожидаемой + if (sdf06GroupId == null || !sdf06GroupId.equals(groupId)) { + log.error("do not currently waiting for gateway response for statement.id {} sdf06 groupId {}; waiting for {}", + statementId, + groupId, + sdf06Id); + return; + } + + Instant updatedTime = Instant.now(); + if (gatewayMsg.isApproved()) { + SDf07 sdf07 = createSuccessSdf07(stmt, updatedTime); + sdf07Imdg.insert(sdf07); + stmt.setOperationStatus(OperationStatus.Executed.getKey()); + stmt.setUpdated(updatedTime); + stmt.setOutSDfId(sdf07.getId()); + statementImdg.update(stmt); + } else { + SDf07 sdf07 = createErrorSdf07(stmt, stmt.getInSDfId(), groupId, null); + sdf07Imdg.insert(sdf07); + stmt.setOperationStatus(OperationStatus.Rejected.getKey()); + stmt.setUpdated(updatedTime); + stmt.setOutSDfId(sdf07.getId()); + statementImdg.update(stmt); + } + sdf06GroupId = null; + log.debug("All gateway responses received for groupId {}", groupId); + } + } + + private SDf07 createSuccessSdf07(Statement stmt, Instant time) { + SDf07 sDf07 = new SDf07(); + sDf07.setInSDfId(stmt.getInSDfId()); + sDf07.setResult(BigDecimal.ZERO); + sDf07.setSum(stmt.getAmount()); + sDf07.setGenerationTime(time); + return sDf07; + } + + private final static BigDecimal errorResult = new BigDecimal("3"); + private SDf07 createErrorSdf07(Statement stmt, Long sdf06Id, Long sdf06GroupId, EnumMessage err) { + SDf07 sDf07 = new SDf07(); + sDf07.setResult(errorResult); + sDf07.setInSDfId(sdf06Id); + //пока использую такой же groupId как для 06. + //при необходимости определить вместе с gatewayWaitInfo + //(т.к. будут создаваться ошибочные sdf07 в при ошибках валидации Statement, там должен быть такой же generationId) + sDf07.setGenerationId(sdf06GroupId); + if (err != null) { + //fixme нет полей для ошибки + } + if (stmt != null) { + sDf07.setAccount(stmt.getAccount()); + sDf07.setSpec(stmt.getComment()); + } + return sDf07; + } + + + private Statement createStatementBySdf06(SDf06 sdf06, Long companyId, Account account) { Statement stmt = new Statement(); stmt.setAddresseeId(companyId); diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/DbfImportKafkaMessenger.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/DbfImportKafkaMessenger.java index 93d3177c2..fd5959115 100644 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/DbfImportKafkaMessenger.java +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/DbfImportKafkaMessenger.java @@ -29,12 +29,12 @@ public class DbfImportKafkaMessenger implements InitializingBean { @Override public void afterPropertiesSet() { - messengers.put(ETable.DF_01, groupId -> messageBalance(groupId, SdfTable.SDF_01)); - messengers.put(ETable.DF_06, groupId -> messageBalance(groupId, SdfTable.SDF_06)); - messengers.put(ETable.DF_09, groupId -> messageBalance(groupId, SdfTable.SDF_09)); - messengers.put(ETable.DF_16, groupId -> messageBalance(groupId, SdfTable.SDF_16)); - messengers.put(ETable.DF_57, groupId -> messageBalance(groupId, SdfTable.SDF_57)); - messengers.put(ETable.DF_04, groupId -> messageBalance(groupId, SdfTable.SDF_04)); + messengers.put(ETable.DF_01, groupId -> messageStatement(groupId, SdfTable.SDF_01)); + messengers.put(ETable.DF_06, groupId -> messageStatement(groupId, SdfTable.SDF_06, Consts.STATEMENT_PROCESS_SDF06)); + messengers.put(ETable.DF_09, groupId -> messageStatement(groupId, SdfTable.SDF_09)); + messengers.put(ETable.DF_16, groupId -> messageStatement(groupId, SdfTable.SDF_16)); + messengers.put(ETable.DF_57, groupId -> messageStatement(groupId, SdfTable.SDF_57)); + messengers.put(ETable.DF_04, groupId -> messageStatement(groupId, SdfTable.SDF_04)); } /** @@ -48,13 +48,17 @@ public class DbfImportKafkaMessenger implements InitializingBean { } } - private void messageBalance(Long groupId, SdfTable table) { + private void messageStatement(Long groupId, SdfTable table) { + messageStatement(groupId, table, Consts.STATEMENT_PROCESS); + } + + private void messageStatement(Long groupId, SdfTable table, String destination) { StatementRequest statementRequest = new StatementRequest(); statementRequest.setGroupId(groupId); statementRequest.setTable(table); - Long msgId = kafka.get().sendRequestToQueue(Consts.STATEMENT_PROCESS, statementRequest); + Long msgId = kafka.get().sendRequestToQueue(destination, statementRequest); log.debug("Send StatementRequest({}, {}) message id={} to kafka \"{}\"", - groupId, table, msgId, Consts.STATEMENT_PROCESS); + groupId, table, msgId, destination); } private void messageDf04(Long groupId) { diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java index a51caf40a..aed7aac24 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java @@ -124,6 +124,7 @@ public interface Consts { String USER_SETTINGS_UPDATE = "user-settings-update"; String STATEMENT_PROCESS = "statement-process"; + String STATEMENT_PROCESS_SDF06 = "statement-process-sdf06"; String SDF02_PROCESS = "sdf02-process"; String SDF03_PROCESS = "sdf03-process"; String SDF04_PROCESS = "sdf04-process"; @@ -143,7 +144,8 @@ public interface Consts { String EXPORT_COMPLETED = "export_completed"; String S_TRADES_IMPORTED = "s_trades-imported"; String LIM_EXPORTED = "lim_exported"; - String ASSET_OPERATION = "asset_operation"; + String ASSET_OPERATION = "asset-operation"; + String ASSET_OPERATION_APPROVAL = "asset-operation-approval"; String JOURNAL_SERVICE = "journal-service-exported"; String ACCOUNT_TERMINATION = "account-termination"; String ACCOUNT_TERMINATION_STEP2 = "account-termination-step2"; diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java new file mode 100644 index 000000000..a20b6e9a1 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java @@ -0,0 +1,18 @@ +package ru.spcex.clearing.platform.messaging.domain.cud.gateway; + +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.List; + +public class AssetOperationApprovalRequest { + @JsonProperty + private List approvals; + + public List getApprovals() { + return approvals; + } + + public void setApprovals(List approvals) { + this.approvals = approvals; + } +} diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/SingleAssetResponse.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/SingleAssetResponse.java new file mode 100644 index 000000000..06fe0a7fe --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/SingleAssetResponse.java @@ -0,0 +1,26 @@ +package ru.spcex.clearing.platform.messaging.domain.cud.gateway; + +import com.fasterxml.jackson.annotation.JsonProperty; + +public class SingleAssetResponse { + @JsonProperty + private Long statementId; + @JsonProperty + private boolean approved; + + public Long getStatementId() { + return statementId; + } + + public void setStatementId(Long statementId) { + this.statementId = statementId; + } + + public boolean isApproved() { + return approved; + } + + public void setApproved(boolean approved) { + this.approved = approved; + } +}