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

обработка SDF06 взаимодействие с gateway

Browse source
This commit is contained in:
ialbert 2023-06-19 16:55:11 +03:00
parent c98087ccd6
commit b81559002f
6 changed files with 176 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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<String, Object> 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);

127
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java
View file

@ -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<SDf06> {
public class Sdf06Executor {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Imdg<Statement> statementImdg;
@ -50,13 +53,16 @@ public class Sdf06Executor extends AbstractExecutor<SDf06> {
private final Imdg<DepoAccount> depoAccountImdg;
private final Imdg<TradingClearingRegistry> tradingClearingRegistryImdg;
private final Imdg<Company> companyImdg;
private final Imdg<SDf09> sdf09Imdg;
private final Imdg<SDf06> sdf06Imdg;
private final Imdg<SDf07> sdf07Imdg;
private final IMessageResolver messageResolver;
private final Function<SDf06, IValidator> sDf06Validator;
private final KafkaSender kafkaSender;
//not thread safe: used in EventReceiver single thread executor
private Long sdf06GroupId;
//@Qualifier("sdf08ValidatorNew") Function<SDf08, IValidator> sDf08Validator
@Autowired
public Sdf06Executor(ImdgProvider imdgProvider,
IMessageResolver messageResolver,
@Qualifier("sdf06ValidatorNew") Function<SDf06, IValidator> sDf06Validator, KafkaSender kafkaSender) {
@ -68,30 +74,19 @@ public class Sdf06Executor extends AbstractExecutor<SDf06> {
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<SDf06> sdfs, StatementRequest statementRequest) {
Result result = new Result();
Long generationIdForGroup = imdgProvider.getImdgIdGenerator().nextId();
result.setGenerationId(generationIdForGroup);
public void createStatementAndSendGatewayCommand(BaseRequest<StatementRequest> systemRequest) {
Imdg<SDf06> sdfImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf06, SDf06.class);
Long groupId = systemRequest.getRequestPayload().getGroupId();
Collection<SDf06> sdfs = sdfImdg.getCollectionObjectsByFieldValues(Map.of(
"generationId", groupId
));
Collection<AssetOperationRequest> requests = new ArrayList<>();
for (SDf06 sDf06 : sdfs) {
IValidator validator = sDf06Validator.apply(sDf06);
@ -100,6 +95,7 @@ public class Sdf06Executor extends AbstractExecutor<SDf06> {
//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<SDf06> {
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<AssetOperationApprovalRequest> 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);

22
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/DbfImportKafkaMessenger.java
View file

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

4
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java
View file

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

18
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/AssetOperationApprovalRequest.java Normal file
View file

@ -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<SingleAssetResponse> approvals;
public List<SingleAssetResponse> getApprovals() {
return approvals;
}
public void setApprovals(List<SingleAssetResponse> approvals) {
this.approvals = approvals;
}
}

26
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/gateway/SingleAssetResponse.java Normal file
View file

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