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

http://jira.mfd.msk:8088/browse/CLS-357 SDF06 executor

Browse source
This commit is contained in:
ialbert 2023-06-20 19:46:24 +03:00
parent 8f7bd69f4e
commit b19ec97a5d
5 changed files with 167 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java
View file

@ -208,12 +208,14 @@ public class ValidationConfig {
context.setValidatedObject(sDf06);
context.addImdg(IMDGDistributedNames.Map_Account, imdgAccount);
context.addImdg(IMDGDistributedNames.Map_CompanySymbols, imdgCompanySymbols);
context.addImdg(IMDGDistributedNames.Map_Registry, imdgCompany);
context.addImdg(IMDGDistributedNames.Map_Company, imdgCompany);
context.addImdg(IMDGDistributedNames.Map_Registry, imdgRegistry);
context.addImdg(IMDGDistributedNames.Map_Session, imdgSession);
return new ValidatorImpl<>(context,
Sdf06NewValidationRule.CompanySymbolPresent,
Sdf06NewValidationRule.AccountPresent
// Sdf06NewValidationRule.MarketIsMkr
// Sdf06NewValidationRule.Balance
Sdf06NewValidationRule.CompanySymbolAndCompanyPresent,
Sdf06NewValidationRule.AccountPresent,
Sdf06NewValidationRule.Session,
Sdf06NewValidationRule.Balance
);
};

2
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java
View file

@ -27,6 +27,8 @@ public enum ClearingError implements IErrorEnumId {
AccountNotFoundB(5217L),
MarketIsNotMkr(5217L),
BalanceInsufficient(5222L),
ActiveSessionIsPresent(-1L),
;
private final Long id;

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

@ -15,9 +15,11 @@ import ru.clearing.classes.statics.data.registry.TradingClearingRegistry;
import ru.clearing.classes.statics.data.sdf.SDf06;
import ru.clearing.classes.statics.data.sdf.SDf07;
import ru.clearing.classes.statics.data.statement.Statement;
import ru.spcex.clearing.error.ClearingError;
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.ExportToFileRequest;
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;
@ -26,6 +28,7 @@ import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.service.validation.ValidationStored;
import ru.spcex.platform.enumeration.*;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.enumeration.IMessageResolver;
@ -55,12 +58,24 @@ public class Sdf06Executor {
private final Imdg<Company> companyImdg;
private final Imdg<SDf06> sdf06Imdg;
private final Imdg<SDf07> sdf07Imdg;
private final ImdgId idGenerator;
private final IMessageResolver messageResolver;
private final Function<SDf06, IValidator> sDf06Validator;
private final KafkaSender kafkaSender;
private final static BigDecimal successResult = BigDecimal.ZERO;
//ошибка проверок
private final static BigDecimal errorResult2 = new BigDecimal("3");
//ошибка проверок баланса или наличия сессии - в этом сценарии session создается
private final static BigDecimal errorResult1 = new BigDecimal("1");
//отказ от gateway
private final static BigDecimal errorResult3 = new BigDecimal("3");
//not thread safe: used in EventReceiver single thread executor
//переменная контроля за тем что от гейтвея пришел ответ именно на наш запрос
private Long sdf06GroupId;
private Long sdf07GroupId;
@Autowired
public Sdf06Executor(ImdgProvider imdgProvider,
@ -74,6 +89,7 @@ public class Sdf06Executor {
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.idGenerator = imdgProvider.getImdgIdGenerator();
this.messageResolver = messageResolver;
this.sDf06Validator = sDf06Validator;
this.kafkaSender = kafkaSender;
@ -87,43 +103,73 @@ public class Sdf06Executor {
Collection<SDf06> sdfs = sdfImdg.getCollectionObjectsByFieldValues(Map.of(
"generationId", groupId
));
Instant now = Instant.now();
Long sdf07GroupId = idGenerator.nextId();
if (sdf06GroupId != null) {
log.warn("currently awaiting gateway response for groupId: {}. skipping groupId {}", sdf06GroupId, groupId);
sdfs.forEach(sdf06 -> {
SDf07 errorSdf07 = createSdf07(sdf06, now, errorResult2);
errorSdf07.setGenerationId(sdf07GroupId);
sdf07Imdg.insert(errorSdf07);
});
sendToExporter(groupId);
return;
}
Collection<AssetOperationRequest> requests = new ArrayList<>();
boolean sdf07WasCreated = false;
for (SDf06 sDf06 : sdfs) {
IValidator validator = sDf06Validator.apply(sDf06);
Optional<EnumMessage> err = validator.tillFirstError();
if (err.isPresent()) {
//todo если ошибка про нехватку баланса - делаем statement с ошибочным статусом
//todo 2 делаем ошибочный SDF07
log.debug("Validation error: {}", messageResolver.resolve(err.get()));
createErrorSdf07(null, sDf06.getId(), groupId, null);
log.debug("sdf06.id={} validation error: {}", sDf06.getId(), messageResolver.resolve(err.get()));
SDf07 errorSdf07;
BigDecimal errorResult;
Statement stmt = null;
if (needToCreateStatement(err.get())) {
errorResult = errorResult1;
stmt = createStatementBySdf06(sDf06,
((Company) validator.getStored(ValidationStored.Sdf06Company)).getId(),
validator.getStored(ValidationStored.Sdf06Account));
} else {
errorResult = errorResult2;
}
errorSdf07 = createSdf07(sDf06, now, errorResult);
errorSdf07.setGenerationId(sdf07GroupId);
sdf07Imdg.insert(errorSdf07);
sdf07WasCreated = true;
if (stmt != null) {
stmt.setOutSDfId(errorSdf07.getId());
statementImdg.insert(stmt);
log.debug("Statement created: {}", stmt.getId());
}
continue;
}
Long companyId = validator.getStored(ValidationStored.Sdf06CompanyId);
Company company = validator.getStored(ValidationStored.Sdf06Company);
Account account = validator.getStored(ValidationStored.Sdf06Account);
Company company = companyImdg.getSingleObjectByID(companyId);
TradingClearingRegistry tcr = tradingClearingRegistryImdg.getSingleObjectByFieldValues(
Map.of("moneyAccountId", account.getId())
);
if (company == null || tcr == null) {
log.error("Company.id {} {}, TCR by account.id {} {} ", companyId,
company != null ? "present" : "not found",
account.getId(),
tcr != null ? "present" : "not found");
continue;
}
//проверка существует ли statement пока убрал
Statement stmt = createStatementBySdf06(sDf06, companyId, account);
Statement stmt = createStatementBySdf06(sDf06, company.getId(), account);
statementImdg.insert(stmt);
log.debug("Statement created: {}", stmt.getId());
requests.add(requestFromStatement(stmt, company.getTradingCode(), tcr.getCode()));
}
if (requests.size() > 0) {
sdf06GroupId = groupId;
this.sdf06GroupId = groupId;
this.sdf07GroupId = sdf07GroupId;
kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, requests);
} else if (sdf07WasCreated) {
sendToExporter(groupId);
}
}
private boolean needToCreateStatement(EnumMessage err) {
return ClearingError.ActiveSessionIsPresent.equals(err.getSubject()) || ClearingError.BalanceInsufficient.equals(err.getSubject());
}
public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) {
Instant now = Instant.now();
for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) {
//получаем запрос для текущей группы sdf06
//находим группу
@ -153,55 +199,65 @@ public class Sdf06Executor {
Instant updatedTime = Instant.now();
if (gatewayMsg.isApproved()) {
SDf07 sdf07 = createSuccessSdf07(stmt, updatedTime);
log.trace("statement.id={}, sdf07.id={}, sdf06.id={} executed",
statementId,
sdf06.getGenerationId(),
sdf06.getId());
SDf07 sdf07 = createSdf07(sdf06, updatedTime, successResult);
sdf07.setGenerationId(sdf07GroupId);
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);
log.trace("statement.id={}, sdf07.id={}, sdf06.id={} rejected (by gateway answer)",
statementId,
sdf06.getGenerationId(),
sdf06.getId());
SDf07 sdf07 = createSdf07(sdf06, now, errorResult3);
sdf07.setGenerationId(sdf07GroupId);
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);
}
sendToExporter(sdf07GroupId);
log.debug("All gateway responses received for SDF06 groupId {}. SDF07 groupId {}", sdf06GroupId, sdf07GroupId);
clearContext();
}
private SDf07 createSuccessSdf07(Statement stmt, Instant time) {
private void clearContext() {
this.sdf06GroupId = null;
this.sdf07GroupId = null;
}
private void sendToExporter(Long generationId) {
ExportToFileRequest exportRequest = new ExportToFileRequest();
exportRequest.setSdfGroupId(generationId);
exportRequest.setNameOfTable("DF-07");
kafkaSender.sendRequestToQueue(Consts.EXPORT_PROCESS, exportRequest);
}
private SDf07 createSdf07(SDf06 sdf06, Instant time, BigDecimal result) {
SDf07 sDf07 = new SDf07();
sDf07.setInSDfId(stmt.getInSDfId());
sDf07.setResult(BigDecimal.ZERO);
sDf07.setSum(stmt.getAmount());
sDf07.setInSDfId(sdf06.getId());
sDf07.setGenerationTime(time);
sDf07.setAccount(sdf06.getAccount());
sDf07.setSum(sdf06.getSum());
sDf07.setMarket(sdf06.getMarket());
sDf07.setType(sdf06.getType());
sDf07.setDeal(sdf06.getDeal());
sDf07.setClientN(sdf06.getClientN());
sDf07.setInn(sdf06.getInn());
sDf07.setBic(sdf06.getBic());
sDf07.setNumber(sdf06.getNumber());
sDf07.setResult(result);
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);
@ -210,8 +266,13 @@ public class Sdf06Executor {
stmt.setComment(sdf06.getSpec());
stmt.setAccountId(account.getId());
stmt.setAccount(account.getAccount());
stmt.setInOutDirection(InOutDirection.in.getKey());
stmt.setAmount(sdf06.getSum());
BigDecimal amount = BigDecimalUtil.safeBD(sdf06.getSum());
if (amount.compareTo(BigDecimal.ZERO) >= 0) {
stmt.setInOutDirection(InOutDirection.in.getKey());
} else {
stmt.setInOutDirection(InOutDirection.out.getKey());
}
stmt.setAmount(amount.abs());
stmt.setOperationStatus(OperationStatus.Pending.getKey());
stmt.setInSDfId(sdf06.getId());
stmt.setInOutSDfType(InOutSDfType.type6.getKey());
@ -223,15 +284,11 @@ public class Sdf06Executor {
private AssetOperationRequest requestFromStatement(Statement stmt, String tradingCode, String tcrCode) {
AssetOperationRequest req = new AssetOperationRequest();
req.setStatementId(stmt.getId());
req.setAmount(BigDecimalUtil.safeBD(stmt.getAmount()).abs());
req.setAmount(stmt.getAmount());
req.setSecuritySymbol(CurrencyCode.RUB.getKey()); //fixme retrieve security symbol from validator
req.setTradingCode(tradingCode);
req.setCode(tcrCode);
if (BigDecimalUtil.safeBD(stmt.getAmount()).compareTo(BigDecimal.ZERO) >= 0) {
req.setDirection(InOutDirection.in.getKey());
} else {
req.setDirection(InOutDirection.out.getKey());
}
req.setDirection(stmt.getInOutDirection());
return req;
}
}

65
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/Sdf06NewValidationRule.java
View file

@ -1,14 +1,23 @@
package ru.spcex.clearing.service.validation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.clearing.classes.statics.data.account.Account;
import ru.clearing.classes.statics.data.company.Company;
import ru.clearing.classes.statics.data.company.CompanySymbols;
import ru.clearing.classes.statics.data.misc.Session;
import ru.clearing.classes.statics.data.registry.Registry;
import ru.clearing.classes.statics.data.sdf.SDf06;
import ru.spcex.clearing.error.ClearingError;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.platform.enumeration.AccountType;
import ru.spcex.platform.enumeration.CompanySymbol;
import ru.spcex.platform.enumeration.RegistryTradingParams;
import ru.spcex.platform.enumeration.WorkflowStatus;
import ru.spcex.platform.imdg.api.Imdg;
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.imdg.validation.ImdgValidationContext;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.number.BigDecimalUtil;
@ -16,15 +25,19 @@ import ru.spcex.platform.utils.text.TextUtil;
import ru.spcex.platform.utils.validation.IValidationRule;
import java.math.BigDecimal;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
public enum Sdf06NewValidationRule implements IValidationRule<ImdgValidationContext<SDf06>> {
CompanySymbolPresent() {
CompanySymbolAndCompanyPresent() {
@Override
public Optional<EnumMessage> validate(ImdgValidationContext<SDf06> context) {
SDf06 sdf06 = context.getValidatedObject();
Imdg<CompanySymbols> companySymbolsImdg = context.obtainMap(IMDGDistributedNames.Map_CompanySymbols, CompanySymbols.class);
Imdg<Company> companyImdg = context.obtainMap(IMDGDistributedNames.Map_Company, Company.class);
CompanySymbols companySymbols = null;
if (sdf06.getInn() != null) { //TextUtil.isEmpty
companySymbols = companySymbolsImdg.getSingleObjectByFieldValues(Map.of(
@ -38,7 +51,13 @@ public enum Sdf06NewValidationRule implements IValidationRule<ImdgValidationCont
if (companySymbols == null) {
return of(ClearingError.CompanyNotFoundB, "inn=" + sdf06.getInn() + "/bic=" + sdf06.getBic());
}
context.storeObject(ValidationStored.Sdf06CompanyId, companySymbols.getCompanyId());
Company company = companyImdg.getSingleObjectByID(companySymbols.getCompanyId());
if (company == null) {
return of(ClearingError.CompanyNotFoundB, "inn=" + sdf06.getInn()
+ "/bic=" + sdf06.getBic()
+ "/companyId=" + companySymbols.getCompanyId());
}
context.storeObject(ValidationStored.Sdf06Company, company);
return empty();
}
},
@ -62,12 +81,15 @@ public enum Sdf06NewValidationRule implements IValidationRule<ImdgValidationCont
return empty();
}
},
MarketIsMkr() {
Session() {
@Override
public Optional<EnumMessage> validate(ImdgValidationContext<SDf06> context) {
SDf06 sdf06 = context.getValidatedObject();
if (sdf06.getMarket() == null || !sdf06.getMarket().equals("U")) {
return of(ClearingError.MarketIsNotMkr);
Imdg<Session> sessionImdg = context.obtainMap(IMDGDistributedNames.Map_Session, Session.class);
Collection<Session> activeSessions = sessionImdg.getCollectionObjectsByFieldValues(Map.of(
"workflowStatus", WorkflowStatus.Active.getKey()
));
if (activeSessions.size() > 0) {
return of(ClearingError.ActiveSessionIsPresent);
}
return empty();
}
@ -79,24 +101,33 @@ public enum Sdf06NewValidationRule implements IValidationRule<ImdgValidationCont
if (BigDecimalUtil.safeBD(sdf06.getSum()).compareTo(BigDecimal.ZERO) >= 0) {
return empty();
}
Company company = context.getStoredObject(ValidationStored.Sdf06Company);
Imdg<Registry> registryImdg = context.obtainMap(IMDGDistributedNames.Map_Registry, Registry.class);
Registry rgs = registryImdg.getSingleObjectByFieldValues(Map.of(
"account", sdf06.getAccount(),
"accountType", AccountType.Clrn.getKey()));
if (rgs == null) {
return of(ClearingError.BalanceInsufficient, sdf06.getAccount());
}
//fixme getFreeBalance???
if (BigDecimalUtil.safeBD(rgs.getCheckBalance()).compareTo(BigDecimalUtil.safeBD(sdf06.getSum()).abs()) < 0) {
return of(ClearingError.BalanceInsufficient, sdf06.getAccount());
ImdgPredicateBuilder prdctBldr = registryImdg.predicateBuilder();
Function<RegistryTradingParams, ImdgPredicate> prdctByRegistryCode = code -> prdctBldr.and(
prdctBldr.equals("account", sdf06.getAccount()),
prdctBldr.equals("companyId", company.getId()),
prdctBldr.sql(RegistryCodeSqlBuilder.getInstance(code).build()));
Registry rgsAmf = registryImdg.getSingleObjectByPredicate(prdctByRegistryCode.apply(RegistryTradingParams.AM_F));
Registry rgsDmt = registryImdg.getSingleObjectByPredicate(prdctByRegistryCode.apply(RegistryTradingParams.DM_T));
log.trace("validating sdf06 id={} account={} companyId={} found registries: AM*F.id={} DM*T.id={}",
sdf06.getId(), sdf06.getAccount(), company.getId(),
rgsAmf == null ? null : rgsAmf.getId(),
rgsDmt == null ? null : rgsDmt.getId()
);
BigDecimal amfBalance = rgsAmf == null ? BigDecimal.ZERO : BigDecimalUtil.safeBD(rgsAmf.getBalance());
BigDecimal dmtBalance = rgsDmt == null ? BigDecimal.ZERO : BigDecimalUtil.safeBD(rgsDmt.getBalance());
BigDecimal stmtAmount = BigDecimalUtil.safeBD(sdf06.getSum());
if (amfBalance.subtract(dmtBalance).compareTo(stmtAmount.abs()) < 0) {
return of(ClearingError.BalanceInsufficient);
}
return empty();
}
}
;
private final static Logger log = LoggerFactory.getLogger(Sdf06NewValidationRule.class);
@Override
public String ruleName() {
return "Sdf08NewValidationRule." + name();
return "Sdf06NewValidationRule." + name();
}
}

2
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java
View file

@ -10,5 +10,5 @@ public enum ValidationStored {
Sdf08Company, Sdf08Account,
Sdf06CompanyId, Sdf06Account
Sdf06Company, Sdf06Account
}