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

SDF06/SDF10 timeouts

Browse source 
This reverts commit fd900e27
This commit is contained in:
ialbert 2024-01-17 13:44:39 +03:00
parent 58f78c39b4
commit 35151a172f
3 changed files with 221 additions and 314 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -15,6 +15,7 @@ import ru.spcex.clearing.platform.messaging.domain.cud.clearing.CreateRegistryRe
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.Sdf04Request; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.Sdf04Request;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.SessionContinueEvent; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.SessionContinueEvent;
import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonIdRequest; 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.payment.PIClearingOutbondActionNewRequest; import ru.spcex.clearing.platform.messaging.domain.cud.payment.PIClearingOutbondActionNewRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.registry.*; import ru.spcex.clearing.platform.messaging.domain.cud.registry.*;
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
@ -137,6 +138,14 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean {
callback(StatementRequest.class) callback(StatementRequest.class)
.setConsumer(sdf06Executor::execute) .setConsumer(sdf06Executor::execute)
.forDestination(Consts.STATEMENT_PROCESS_SDF06, callbacks::put); .forDestination(Consts.STATEMENT_PROCESS_SDF06, callbacks::put);
callback(AssetOperationApprovalRequest.class)
.setConsumer(req -> {
sdf06Executor.processGatewayResponse(req);
sdf10Executor.processGatewayResponse(req);
})
.forDestination(Consts.ASSET_OPERATION_APPROVAL, callbacks::put);
callback(STradesImportedRequest.class) callback(STradesImportedRequest.class)
.setConsumer(event -> clearingService.executeSTrade()) .setConsumer(event -> clearingService.executeSTrade())
.forDestination(S_TRADES_IMPORTED, callbacks::put); .forDestination(S_TRADES_IMPORTED, callbacks::put);

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

@ -17,7 +17,9 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.Consts; 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.ExportToFileRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest; import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationListRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationRequest; 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.domain.cud.gateway.SingleAssetResponse;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.service.AssetTrio; import ru.spcex.clearing.service.AssetTrio;
@ -27,14 +29,16 @@ import ru.spcex.clearing.service.registry.DmiService;
import ru.spcex.clearing.service.schedule.TradingTimeService; import ru.spcex.clearing.service.schedule.TradingTimeService;
import ru.spcex.clearing.service.validation.Sdf06NewValidationRule; import ru.spcex.clearing.service.validation.Sdf06NewValidationRule;
import ru.spcex.clearing.service.validation.ValidationStored; import ru.spcex.clearing.service.validation.ValidationStored;
import ru.spcex.clearing.session.stage.impl.GatewayBatchRequester; import ru.spcex.platform.classes.base.SpcexObjectBase;
import ru.spcex.platform.enumeration.*; import ru.spcex.platform.enumeration.*;
import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
import ru.spcex.platform.imdg.validation.ImdgValidationContext; import ru.spcex.platform.imdg.validation.ImdgValidationContext;
import ru.spcex.platform.utils.enumeration.EnumMessage; import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.enumeration.IMessageResolver; import ru.spcex.platform.utils.enumeration.IMessageResolver;
import ru.spcex.platform.utils.text.TextUtil;
import ru.spcex.platform.utils.validation.IValidator; import ru.spcex.platform.utils.validation.IValidator;
import ru.spcex.platform.utils.validation.ValidatorImpl; import ru.spcex.platform.utils.validation.ValidatorImpl;
@ -74,19 +78,19 @@ public class Sdf06Executor {
//отказ от gateway либо недостаточно средств у нас //отказ от gateway либо недостаточно средств у нас
private final static BigDecimal errorResult3 = new BigDecimal("3"); private final static BigDecimal errorResult3 = new BigDecimal("3");
private final static BigDecimal errorResult4 = new BigDecimal("4"); private final static BigDecimal errorResult4 = new BigDecimal("4");
private final GatewayBatchRequester gateway;
//not thread safe: used in EventReceiver single thread executor //not thread safe: used in EventReceiver single thread executor
//переменная контроля за тем что от гейтвея пришел ответ именно на наш запрос //переменная контроля за тем что от гейтвея пришел ответ именно на наш запрос
LinkedList<BaseRequest<StatementRequest>> cachedRequests = new LinkedList<>(); LinkedList<BaseRequest<StatementRequest>> cachedRequests = new LinkedList<>();
private Long sdf06GroupId;
private Long sdf07GroupId;
@Autowired @Autowired
public Sdf06Executor(ImdgProvider imdgProvider, public Sdf06Executor(ImdgProvider imdgProvider,
IMessageResolver messageResolver, IMessageResolver messageResolver,
@Qualifier("sdf06ValidatorNew") Function<SDf06, IValidator> sDf06Validator, @Qualifier("sdf06ValidatorNew") Function<SDf06, IValidator> sDf06Validator,
TradingTimeService tradingTimeService, KafkaSender kafkaSender, FilenameObtainer filenameObtainer, DmiService dmiService, AssetTBFProcessing assets, TradingTimeService tradingTimeService, KafkaSender kafkaSender, FilenameObtainer filenameObtainer, DmiService dmiService, AssetTBFProcessing assets) {
GatewayBatchRequester gateway) {
this.imdgProvider = imdgProvider; this.imdgProvider = imdgProvider;
this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class); this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class);
this.tradingTimeService = tradingTimeService; this.tradingTimeService = tradingTimeService;
@ -102,8 +106,6 @@ public class Sdf06Executor {
this.filenameObtainer = filenameObtainer; this.filenameObtainer = filenameObtainer;
this.dmiService = dmiService; this.dmiService = dmiService;
this.assets = assets; this.assets = assets;
this.gateway = gateway;
this.gateway.setName("Sdf06");
} }
public void execute(BaseRequest<StatementRequest> systemRequest) { public void execute(BaseRequest<StatementRequest> systemRequest) {
@ -116,13 +118,12 @@ public class Sdf06Executor {
log.info("start processing SDF06: groupId={} file={}", groupId, fileName); log.info("start processing SDF06: groupId={} file={}", groupId, fileName);
Instant now = Instant.now(); Instant now = Instant.now();
Long sdf07GroupId = idGenerator.nextId(); Long sdf07GroupId = idGenerator.nextId();
//if (sdf06GroupId != null) { if (sdf06GroupId != null) {
// log.warn("currently awaiting gateway response for groupId: {}. adding to cache groupId {}", sdf06GroupId, groupId); log.warn("currently awaiting gateway response for groupId: {}. adding to cache groupId {}", sdf06GroupId, groupId);
// cachedRequests.add(systemRequest); cachedRequests.add(systemRequest);
// return; return;
//} }
Collection<AssetOperationRequest> requests = new ArrayList<>(); Collection<AssetOperationRequest> requests = new ArrayList<>();
Collection<Trio> records = new ArrayList<>();
boolean sdf07WasCreated = false; boolean sdf07WasCreated = false;
for (SDf06 sDf06 : sdfs) { for (SDf06 sDf06 : sdfs) {
IValidator validator = sDf06Validator.apply(sDf06); IValidator validator = sDf06Validator.apply(sDf06);
@ -156,7 +157,6 @@ public class Sdf06Executor {
log.debug("Statement created: {}", stmt.getId()); log.debug("Statement created: {}", stmt.getId());
if (InOutDirection.out.getKey().equals(stmt.getInOutDirection()) && tradingTimeService.isTradingTime()) { if (InOutDirection.out.getKey().equals(stmt.getInOutDirection()) && tradingTimeService.isTradingTime()) {
requests.add(GatewayRequestCreator.from(stmt, company.getTradingCode(), tcr.getCode())); requests.add(GatewayRequestCreator.from(stmt, company.getTradingCode(), tcr.getCode()));
records.add(new Trio(sDf06, stmt, tcr));
} else { } else {
processedApproved(stmt, sDf06, now, sdf07GroupId); processedApproved(stmt, sDf06, now, sdf07GroupId);
if (tcr != null) { if (tcr != null) {
@ -171,80 +171,19 @@ public class Sdf06Executor {
} }
} }
if (requests.size() > 0) { if (requests.size() > 0) {
Instant updatedTime = Instant.now(); this.sdf06GroupId = groupId;
Optional<List<SingleAssetResponse>> gatewayResponse = gateway.gatewayRequestAndWaitBatch(requests); this.sdf07GroupId = sdf07GroupId;
if (gatewayResponse.isEmpty()) { AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest();
log.error("couldn't retrieve SDF06 gateway response for groupId={}/sdf07.groupId={}.", groupId, sdf07GroupId); assetOperationListRequest.setAssetOperationRequests(requests);
sdf07WasCreated = true; kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest);
records.forEach(trio -> { } else if (sdf07WasCreated) {
SDf06 sDf06 = trio.sdf06();
Statement stmt = trio.stmt();
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);
});
} else {
List<SingleAssetResponse> gtwRsp = gatewayResponse.get();
for (SingleAssetResponse gatewayMsg : gtwRsp) {
//получаем запрос для текущей группы sdf06
//находим группу
Long statementId = gatewayMsg.getEntityId();
Trio trio = records.stream()
.filter(s -> Objects.equals(s.stmt().getId(), statementId))
.findFirst()
.orElse(null);
if (trio == null) {
log.error("SDF06.groupId={}, SDF07.groupId={} gateway response statement.id {} with type {} not found",
groupId, sdf07GroupId, statementId, InOutSDfType.type6.getKey());
continue;
}
SDf06 sdf06 = trio.sdf06();
Statement stmt = trio.stmt();
TradingClearingRegistry tcr = trio.tcr();
if (gatewayMsg.isApproved()) {
processedApproved(stmt, sdf06, updatedTime, sdf07GroupId);
dmiService.setProcContract(tcr != null ? tcr.getId() : null,
CurrencyCode.RUB.getKey(),
safeBD(sdf06.getSum()),
sdf06.getNumber().toString()
);
if (tcr != null) {
Optional<AssetTrio> asts = assets.searchMoneyByAccAndCompany(tcr.getCompanyId(), tcr.getMoneyAccountId());
asts.ifPresent(a -> assets.process(a.a__b(), a.a__t(), a.a__f(), BigDecimal.ZERO));
}
} else {
SDf07 sdf07 = createSdf07(sdf06, now, errorResult3);
sdf07.setGenerationId(sdf07GroupId);
sdf07Imdg.insert(sdf07);
log.trace("statement.id={}, sdf07.id={}, sdf06.id={} sdf06.getGenerationId={} rejected (by gateway answer)",
statementId,
sdf07.getId(),
sdf06.getId(), sdf06.getGenerationId());
stmt.setOperationStatus(OperationStatus.Rejected.getKey());
stmt.setUpdated(updatedTime);
stmt.setOutSDfId(sdf07.getId());
statementImdg.update(stmt);
}
sdf07WasCreated = true;
}
}
}
if (sdf07WasCreated) {
sendToExporter(sdf07GroupId, fileName); sendToExporter(sdf07GroupId, fileName);
} }
if (cachedRequests.size() > 0) {
BaseRequest<StatementRequest> statementRequest = cachedRequests.removeFirst();
execute(statementRequest);
}
} }
//private void processedApproved(Statement stmt, SDf06 sDf06, Instant time) { private void processedApproved(Statement stmt, SDf06 sDf06, Instant time) {
// processedApproved(stmt, sDf06, time, sdf07GroupId); processedApproved(stmt, sDf06, time, sdf07GroupId);
//} }
private void processedApproved(Statement stmt, SDf06 sDf06, Instant time, Long sdf07GenerationId) { private void processedApproved(Statement stmt, SDf06 sDf06, Instant time, Long sdf07GenerationId) {
log.trace("statement.id={}, sdf07.id={}, sdf06.id={} executed", log.trace("statement.id={}, sdf07.id={}, sdf06.id={} executed",
@ -288,87 +227,92 @@ public class Sdf06Executor {
return result; return result;
} }
// public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) { public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) {
// //может прийти неограниченно позже 06го, после того как прошел клиринг например... //может прийти неограниченно позже 06го, после того как прошел клиринг например...
// Instant now = Instant.now(); Instant now = Instant.now();
// String fileName = null; String fileName = null;
// boolean requestIsIntendedForSdf06 = false; boolean requestIsIntendedForSdf06 = false;
// for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) { for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) {
// //получаем запрос для текущей группы sdf06 //получаем запрос для текущей группы sdf06
// //находим группу //находим группу
// Long statementId = gatewayMsg.getEntityId(); Long statementId = gatewayMsg.getEntityId();
// ImdgPredicateBuilder pb = statementImdg.predicateBuilder(); ImdgPredicateBuilder pb = statementImdg.predicateBuilder();
// Statement stmt = statementImdg.getFirstObjectByPredicate( Statement stmt = statementImdg.getFirstObjectByPredicate(
// pb.and( pb.and(
// pb.equals("id", statementId), pb.equals("id", statementId),
// pb.equals("inOutSDfType", InOutSDfType.type6.getKey()) pb.equals("inOutSDfType", InOutSDfType.type6.getKey())
// ) )
// ); );
// if (stmt == null) { if (stmt == null) {
// log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type6.getKey()); log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type6.getKey());
// return; return;
// } }
// requestIsIntendedForSdf06 = true; requestIsIntendedForSdf06 = true;
//
// Long sdf06Id = stmt.getInSDfId(); Long sdf06Id = stmt.getInSDfId();
// SDf06 sdf06 = sdf06Imdg.getSingleObjectByID(sdf06Id); SDf06 sdf06 = sdf06Imdg.getSingleObjectByID(sdf06Id);
// if (fileName == null && !TextUtil.isEmpty(sdf06.getFileName())) { if (fileName == null && !TextUtil.isEmpty(sdf06.getFileName())) {
// fileName = sdf06.getFileName(); fileName = sdf06.getFileName();
// } }
// if (sdf06 == null) { if (sdf06 == null) {
// log.error("Sdf06.id {} not found by statement.id {}", sdf06Id, statementId); log.error("Sdf06.id {} not found by statement.id {}", sdf06Id, statementId);
// return; return;
// } }
//
// Long groupId = sdf06.getGenerationId(); Long groupId = sdf06.getGenerationId();
// //сверяем группу SDF06 пришедшего запроса с ожидаемой //сверяем группу SDF06 пришедшего запроса с ожидаемой
// if (sdf06GroupId == null || !sdf06GroupId.equals(groupId)) { if (sdf06GroupId == null || !sdf06GroupId.equals(groupId)) {
// log.error("do not currently waiting for gateway response for statement.id {} sdf06 groupId {}; waiting for {}", log.error("do not currently waiting for gateway response for statement.id {} sdf06 groupId {}; waiting for {}",
// statementId, statementId,
// groupId, groupId,
// sdf06Id); sdf06Id);
// return; return;
// } }
//
// Instant updatedTime = Instant.now(); Instant updatedTime = Instant.now();
// if (gatewayMsg.isApproved()) { if (gatewayMsg.isApproved()) {
// processedApproved(stmt, sdf06, updatedTime); processedApproved(stmt, sdf06, updatedTime);
// Optional<TradingClearingRegistry> tcr = searchTcrOnGatewayResponse(sdf06); Optional<TradingClearingRegistry> tcr = searchTcrOnGatewayResponse(sdf06);
// dmiService.setProcContract(tcr.map(SpcexObjectBase::getId).orElse(null), dmiService.setProcContract(tcr.map(SpcexObjectBase::getId).orElse(null),
// CurrencyCode.RUB.getKey(), CurrencyCode.RUB.getKey(),
// safeBD(sdf06.getSum()), safeBD(sdf06.getSum()),
// sdf06.getNumber().toString() sdf06.getNumber().toString()
// ); );
// if (tcr.isPresent()) { if (tcr.isPresent()) {
// Optional<AssetTrio> asts = assets.searchMoneyByAccAndCompany(tcr.get().getCompanyId(), tcr.get().getMoneyAccountId()); Optional<AssetTrio> asts = assets.searchMoneyByAccAndCompany(tcr.get().getCompanyId(), tcr.get().getMoneyAccountId());
// asts.ifPresent(a -> assets.process(a.a__b(), a.a__t(), a.a__f(), BigDecimal.ZERO)); asts.ifPresent(a -> assets.process(a.a__b(), a.a__t(), a.a__f(), BigDecimal.ZERO));
// } }
// } else { } else {
// log.trace("statement.id={}, sdf07.id={}, sdf06.id={} rejected (by gateway answer)", log.trace("statement.id={}, sdf07.id={}, sdf06.id={} rejected (by gateway answer)",
// statementId, statementId,
// sdf06.getGenerationId(), sdf06.getGenerationId(),
// sdf06.getId()); sdf06.getId());
// SDf07 sdf07 = createSdf07(sdf06, now, errorResult3); SDf07 sdf07 = createSdf07(sdf06, now, errorResult3);
// sdf07.setGenerationId(sdf07GroupId); sdf07.setGenerationId(sdf07GroupId);
// sdf07Imdg.insert(sdf07); sdf07Imdg.insert(sdf07);
// stmt.setOperationStatus(OperationStatus.Rejected.getKey()); stmt.setOperationStatus(OperationStatus.Rejected.getKey());
// stmt.setUpdated(updatedTime); stmt.setUpdated(updatedTime);
// stmt.setOutSDfId(sdf07.getId()); stmt.setOutSDfId(sdf07.getId());
// statementImdg.update(stmt); statementImdg.update(stmt);
// } }
// } }
// if (requestIsIntendedForSdf06) { if (requestIsIntendedForSdf06) {
// sendToExporter(sdf07GroupId, fileName); sendToExporter(sdf07GroupId, fileName);
// log.debug("All gateway responses received for SDF06 groupId {}. SDF07 groupId {}", sdf06GroupId, sdf07GroupId); log.debug("All gateway responses received for SDF06 groupId {}. SDF07 groupId {}", sdf06GroupId, sdf07GroupId);
// clearContext(); clearContext();
// } else { } else {
// log.debug("gateway response was not for SDF06 executor"); log.debug("gateway response was not for SDF06 executor");
// } }
// if (cachedRequests.size() > 0) { if (cachedRequests.size() > 0) {
// BaseRequest<StatementRequest> statementRequest = cachedRequests.removeFirst(); BaseRequest<StatementRequest> statementRequest = cachedRequests.removeFirst();
// execute(statementRequest); execute(statementRequest);
// } }
// } }
private void clearContext() {
this.sdf06GroupId = null;
this.sdf07GroupId = null;
}
private void sendToExporter(Long generationId, String fileName) { private void sendToExporter(Long generationId, String fileName) {
ExportToFileRequest exportRequest = new ExportToFileRequest(); ExportToFileRequest exportRequest = new ExportToFileRequest();
@ -445,6 +389,4 @@ public class Sdf06Executor {
} }
return Optional.empty(); return Optional.empty();
} }
private static record Trio (SDf06 sdf06, Statement stmt, TradingClearingRegistry tcr) {};
} }

252
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf10Executor.java
View file

@ -19,7 +19,9 @@ import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.Consts; 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.balance.StatementRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationListRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationRequest; 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.domain.cud.gateway.SingleAssetResponse;
import ru.spcex.clearing.platform.messaging.domain.cud.importexport.SwtExporterRequest; import ru.spcex.clearing.platform.messaging.domain.cud.importexport.SwtExporterRequest;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
@ -27,12 +29,12 @@ import ru.spcex.clearing.service.registry.AssetTBFProcessing;
import ru.spcex.clearing.service.registry.RegistryManager; import ru.spcex.clearing.service.registry.RegistryManager;
import ru.spcex.clearing.service.schedule.TradingTimeService; import ru.spcex.clearing.service.schedule.TradingTimeService;
import ru.spcex.clearing.service.validation.ValidationStored; import ru.spcex.clearing.service.validation.ValidationStored;
import ru.spcex.clearing.session.stage.impl.GatewayBatchRequester;
import ru.spcex.clearing.session.stage.util.RegistryUtil; import ru.spcex.clearing.session.stage.util.RegistryUtil;
import ru.spcex.platform.enumeration.*; import ru.spcex.platform.enumeration.*;
import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
import ru.spcex.platform.imdg.api.predicate.specific.SecuritySelector; import ru.spcex.platform.imdg.api.predicate.specific.SecuritySelector;
import ru.spcex.platform.utils.enumeration.EnumMessage; import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.enumeration.IEnumId; import ru.spcex.platform.utils.enumeration.IEnumId;
@ -42,7 +44,10 @@ import ru.spcex.platform.utils.validation.IValidator;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.time.Instant; import java.time.Instant;
import java.time.LocalDate; import java.time.LocalDate;
import java.util.*; import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function; import java.util.function.Function;
import static ru.spcex.platform.utils.number.BigDecimalUtil.safeBD; import static ru.spcex.platform.utils.number.BigDecimalUtil.safeBD;
@ -65,12 +70,10 @@ public class Sdf10Executor {
private final KafkaSender kafkaSender; private final KafkaSender kafkaSender;
private final SecuritySelector<Security> scrtSlct; private final SecuritySelector<Security> scrtSlct;
private final AssetTBFProcessing assets; private final AssetTBFProcessing assets;
private final GatewayBatchRequester gateway;
private final static String OK = "OK"; private final static String OK = "OK";
private final static String SYNTAX_ERROR = "Синтаксическая ошибка (файл сформирован неверно)"; private final static String SYNTAX_ERROR = "Синтаксическая ошибка (файл сформирован неверно)";
private final static String GATEWAY_REJECTED = "Отрицательный ответ от Торговой Системы"; private final static String GATEWAY_REJECTED = "Отрицательный ответ от Торговой Системы";
private final static String GATEWAY_NO_ANSWER = "Не удалось получить ответ от Торговой Системы";
private final static String NOT_FOUND_ACCOUNT = "Не найден код раздела субсчета/счета депо"; private final static String NOT_FOUND_ACCOUNT = "Не найден код раздела субсчета/счета депо";
private final static String NOT_FOUND_COMPANY = "Не найдена компания"; private final static String NOT_FOUND_COMPANY = "Не найдена компания";
private final static String NOT_FOUND_TCR = "Не найден ТКР"; private final static String NOT_FOUND_TCR = "Не найден ТКР";
@ -80,11 +83,14 @@ public class Sdf10Executor {
private final static String SESSION_IS_ACTIVE = "Существует активная сессия"; private final static String SESSION_IS_ACTIVE = "Существует активная сессия";
private final static String GENERAL_ERROR = "Общая ошибка обработки SDF10"; private final static String GENERAL_ERROR = "Общая ошибка обработки SDF10";
private Long sdf10GroupId;
private Long sdf11GroupId;
@Autowired @Autowired
public Sdf10Executor(ImdgProvider imdgProvider, public Sdf10Executor(ImdgProvider imdgProvider,
RegistryManager rgsMng, IMessageResolver messageResolver, RegistryManager rgsMng, IMessageResolver messageResolver,
@Qualifier("sdf10Validator") Function<SDf10, IValidator> sDf10Validator, @Qualifier("sdf10Validator") Function<SDf10, IValidator> sDf10Validator,
TradingTimeService tradingTimeService, KafkaSender kafkaSender, AssetTBFProcessing assets, GatewayBatchRequester gateway) { TradingTimeService tradingTimeService, KafkaSender kafkaSender, AssetTBFProcessing assets) {
this.imdgProvider = imdgProvider; this.imdgProvider = imdgProvider;
this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class); this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class);
this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class);
@ -99,8 +105,6 @@ public class Sdf10Executor {
this.plannerAllTodayImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_PlannerAllToday, PlannerAllToday.class); this.plannerAllTodayImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_PlannerAllToday, PlannerAllToday.class);
this.scrtSlct = new SecuritySelector<>(imdgProvider, Security.class); this.scrtSlct = new SecuritySelector<>(imdgProvider, Security.class);
this.assets = assets; this.assets = assets;
this.gateway = gateway;
this.gateway.setName("sdf10");
} }
public void execute(BaseRequest<StatementRequest> systemRequest) { public void execute(BaseRequest<StatementRequest> systemRequest) {
@ -112,17 +116,16 @@ public class Sdf10Executor {
Instant now = Instant.now(); Instant now = Instant.now();
Long sdf11GroupId = idGenerator.nextId(); Long sdf11GroupId = idGenerator.nextId();
log.info("processing SDF10 groupId: {} sdf10s: {}", groupId, sdfs.size()); log.info("processing SDF10 groupId: {} sdf10s: {}", groupId, sdfs.size());
//if (sdf10GroupId != null) { if (sdf10GroupId != null) {
// log.warn("currently awaiting gateway response for groupId: {}. skipping groupId {}", sdf10GroupId, groupId); log.warn("currently awaiting gateway response for groupId: {}. skipping groupId {}", sdf10GroupId, groupId);
// sdfs.forEach(sdf10 -> { sdfs.forEach(sdf10 -> {
// SDf11 errorSdf11 = createSdf11(sdf10, sdf11GroupId, now, PREVIOUS_REQUEST_NOT_PROCESSED); SDf11 errorSdf11 = createSdf11(sdf10, sdf11GroupId, now, PREVIOUS_REQUEST_NOT_PROCESSED);
// this.sdf11Imdg.insert(errorSdf11); this.sdf11Imdg.insert(errorSdf11);
// }); });
// sendToExporter(sdf11GroupId); sendToExporter(sdf11GroupId);
// return; return;
//} }
Collection<AssetOperationRequest> requests = new ArrayList<>(); Collection<AssetOperationRequest> requests = new ArrayList<>();
List<Trio> records = new ArrayList<>();
boolean sdf11WasCreated = false; boolean sdf11WasCreated = false;
for (SDf10 sDf10 : sdfs) { for (SDf10 sDf10 : sdfs) {
IValidator validator = sDf10Validator.apply(sDf10); IValidator validator = sDf10Validator.apply(sDf10);
@ -162,7 +165,6 @@ public class Sdf10Executor {
log.debug("Statement created: {}", stmt.getId()); log.debug("Statement created: {}", stmt.getId());
if (InOutDirection.out.getKey().equals(stmt.getInOutDirection()) && tradingTimeService.isTradingTime()) { if (InOutDirection.out.getKey().equals(stmt.getInOutDirection()) && tradingTimeService.isTradingTime()) {
requests.add(requestFromStatement(stmt, company.getTradingCode(), tcr.getCode(), sDf10.getSecurityCode())); requests.add(requestFromStatement(stmt, company.getTradingCode(), tcr.getCode(), sDf10.getSecurityCode()));
records.add(new Trio(sDf10, stmt, security));
} else { } else {
SDf11 sDf11 = processedApproved(stmt, sDf10, now, sdf11GroupId); SDf11 sDf11 = processedApproved(stmt, sDf10, now, sdf11GroupId);
BigDecimal amount = InOutDirection.in.equalsByKey(stmt.getInOutDirection()) ? BigDecimal amount = InOutDirection.in.equalsByKey(stmt.getInOutDirection()) ?
@ -176,65 +178,12 @@ public class Sdf10Executor {
} }
} }
if (requests.size() > 0) { if (requests.size() > 0) {
Instant updatedTime = Instant.now(); this.sdf10GroupId = groupId;
Optional<List<SingleAssetResponse>> gatewayResponse = gateway.gatewayRequestAndWaitBatch(requests); this.sdf11GroupId = sdf11GroupId;
if (gatewayResponse.isEmpty()) { AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest();
log.error("couldn't retrieve SDF10 gateway response for groupId={}/sdf11.groupId={}", groupId, sdf11GroupId); assetOperationListRequest.setAssetOperationRequests(requests);
sdf11WasCreated = true; kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest);
records.forEach(trio -> { } else if (sdf11WasCreated) {
SDf10 sDf10 = trio.sDf10();
Statement stmt = trio.stmt();
SDf11 errorSdf11 = createSdf11(sDf10, sdf11GroupId, now, GATEWAY_NO_ANSWER);
errorSdf11.setGenerationId(sdf11GroupId);
sdf11Imdg.insert(errorSdf11);
stmt.setOperationStatus(OperationStatus.Rejected.getKey());
stmt.setUpdated(updatedTime);
stmt.setOutSDfId(errorSdf11.getId());
statementImdg.update(stmt);
});
} else {
List<SingleAssetResponse> gtwRsp = gatewayResponse.get();
Instant gtwReceived = Instant.now();
for (SingleAssetResponse gatewayMsg : gtwRsp) {
Long statementId = gatewayMsg.getEntityId();
Trio trio = records.stream()
.filter(r -> Objects.equals(r.stmt().getId(), statementId))
.findFirst()
.orElse(null);
if (trio == null) {
log.error("SDF10.groupId={}, SDF07.groupId={}" +
" gateway response statement.id {} with type {} not found",
groupId, sdf11GroupId, statementId, InOutSDfType.type10.getKey());
continue;
}
SDf10 sdf10 = trio.sDf10();
Statement stmt = trio.stmt();
Security security = trio.security();
if (gatewayMsg.isApproved()) {
createDs_iResponseFromGateway(
stmt.getAccountId(),
security != null ? security.getSecuritySymbol() : null,
stmt.getAmount().negate(),
sdf10.getOutDocument());
//tcrId, companyId, securitySymbol
processedApproved(stmt, sdf10, gtwReceived, sdf11GroupId);
} else {
log.trace("statement.id={}, sdf11.id={}, sdf10.id={} rejected (by gateway answer)",
statementId,
sdf10.getGenerationId(),
sdf10.getId());
SDf11 sdf11 = createSdf11(sdf10, sdf11GroupId, now, GATEWAY_REJECTED);
sdf11Imdg.insert(sdf11);
stmt.setOperationStatus(OperationStatus.Rejected.getKey());
stmt.setUpdated(gtwReceived);
stmt.setOutSDfId(sdf11.getId());
statementImdg.update(stmt);
}
sdf11WasCreated = true;
}
}
}
if (sdf11WasCreated) {
sendToExporter(sdf11GroupId); sendToExporter(sdf11GroupId);
} }
} }
@ -280,6 +229,10 @@ public class Sdf10Executor {
log.debug("created DS*I.id={}", rgsD.getId()); log.debug("created DS*I.id={}", rgsD.getId());
} }
private void processedApproved(Statement stmt, SDf10 sDf10, Instant time) {
processedApproved(stmt, sDf10, time, sdf11GroupId);
}
private SDf11 processedApproved(Statement stmt, SDf10 sDf10, Instant time, Long sdf11GenerationId) { private SDf11 processedApproved(Statement stmt, SDf10 sDf10, Instant time, Long sdf11GenerationId) {
log.trace("statement.id={}, sdf11.id={}, sdf10.id={} executed", log.trace("statement.id={}, sdf11.id={}, sdf10.id={} executed",
stmt.getId(), stmt.getId(),
@ -294,75 +247,80 @@ public class Sdf10Executor {
return sdf11; return sdf11;
} }
// public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) { public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) {
// Instant now = Instant.now(); Instant now = Instant.now();
// boolean requestIsIntendedForSdf10 = false; boolean requestIsIntendedForSdf10 = false;
// for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) { for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) {
// //получаем запрос для текущей группы sdf10 //получаем запрос для текущей группы sdf10
// //находим группу //находим группу
// Long statementId = gatewayMsg.getEntityId(); Long statementId = gatewayMsg.getEntityId();
// ImdgPredicateBuilder pb = statementImdg.predicateBuilder(); ImdgPredicateBuilder pb = statementImdg.predicateBuilder();
// Statement stmt = statementImdg.getFirstObjectByPredicate( Statement stmt = statementImdg.getFirstObjectByPredicate(
// pb.and( pb.and(
// pb.equals("id", statementId), pb.equals("id", statementId),
// pb.equals("inOutSDfType", InOutSDfType.type10.getKey()) pb.equals("inOutSDfType", InOutSDfType.type10.getKey())
// ) )
// ); );
// if (stmt == null) { if (stmt == null) {
// log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type10.getKey()); log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type10.getKey());
// return; return;
// } }
// requestIsIntendedForSdf10 = true; requestIsIntendedForSdf10 = true;
//
// Long sdf10Id = stmt.getInSDfId(); Long sdf10Id = stmt.getInSDfId();
// SDf10 sdf10 = sdf10Imdg.getSingleObjectByID(sdf10Id); SDf10 sdf10 = sdf10Imdg.getSingleObjectByID(sdf10Id);
// if (sdf10 == null) { if (sdf10 == null) {
// log.error("Sdf10.id {} not found by statement.id {}", sdf10Id, statementId); log.error("Sdf10.id {} not found by statement.id {}", sdf10Id, statementId);
// return; return;
// } }
//
// Long groupId = sdf10.getGenerationId(); Long groupId = sdf10.getGenerationId();
// //сверяем группу SDF10 пришедшего запроса с ожидаемой //сверяем группу SDF10 пришедшего запроса с ожидаемой
// if (sdf10GroupId == null || !sdf10GroupId.equals(groupId)) { if (sdf10GroupId == null || !sdf10GroupId.equals(groupId)) {
// log.error("do not currently waiting for gateway response for statement.id {} sdf10 groupId {}; waiting for {}", log.error("do not currently waiting for gateway response for statement.id {} sdf10 groupId {}; waiting for {}",
// statementId, statementId,
// groupId, groupId,
// sdf10Id); sdf10Id);
// return; return;
// } }
//
// Instant updatedTime = Instant.now(); Instant updatedTime = Instant.now();
// if (gatewayMsg.isApproved()) { if (gatewayMsg.isApproved()) {
// Security security = scrtSlct.selectSecurityById(stmt.getSecurityId()); Security security = scrtSlct.selectSecurityById(stmt.getSecurityId());
// createDs_iResponseFromGateway( createDs_iResponseFromGateway(
// stmt.getAccountId(), stmt.getAccountId(),
//// stmt.getAddresseeId(), // stmt.getAddresseeId(),
// security != null ? security.getSecuritySymbol() : null, security != null ? security.getSecuritySymbol() : null,
// stmt.getAmount().negate(), stmt.getAmount().negate(),
// sdf10.getOutDocument()); sdf10.getOutDocument());
// //tcrId, companyId, securitySymbol //tcrId, companyId, securitySymbol
// processedApproved(stmt, sdf10, updatedTime); processedApproved(stmt, sdf10, updatedTime);
// } else { } else {
// log.trace("statement.id={}, sdf11.id={}, sdf10.id={} rejected (by gateway answer)", log.trace("statement.id={}, sdf11.id={}, sdf10.id={} rejected (by gateway answer)",
// statementId, statementId,
// sdf10.getGenerationId(), sdf10.getGenerationId(),
// sdf10.getId()); sdf10.getId());
// SDf11 sdf11 = createSdf11(sdf10, sdf11GroupId, now, GATEWAY_REJECTED); SDf11 sdf11 = createSdf11(sdf10, sdf11GroupId, now, GATEWAY_REJECTED);
// sdf11Imdg.insert(sdf11); sdf11Imdg.insert(sdf11);
// stmt.setOperationStatus(OperationStatus.Rejected.getKey()); stmt.setOperationStatus(OperationStatus.Rejected.getKey());
// stmt.setUpdated(updatedTime); stmt.setUpdated(updatedTime);
// stmt.setOutSDfId(sdf11.getId()); stmt.setOutSDfId(sdf11.getId());
// statementImdg.update(stmt); statementImdg.update(stmt);
// } }
// } }
// if (requestIsIntendedForSdf10) { if (requestIsIntendedForSdf10) {
// sendToExporter(sdf11GroupId); sendToExporter(sdf11GroupId);
// log.debug("All gateway responses received for SDF10 groupId {}. SDF11 groupId {}", sdf10GroupId, sdf11GroupId); log.debug("All gateway responses received for SDF10 groupId {}. SDF11 groupId {}", sdf10GroupId, sdf11GroupId);
// clearContext(); clearContext();
// } else { } else {
// log.debug("gateway response was not for SDF10 executor"); log.debug("gateway response was not for SDF10 executor");
// } }
// } }
private void clearContext() {
this.sdf10GroupId = null;
this.sdf11GroupId = null;
}
private void sendToExporter(Long generationId) { private void sendToExporter(Long generationId) {
SwtExporterRequest swtReq = new SwtExporterRequest(); SwtExporterRequest swtReq = new SwtExporterRequest();
@ -448,6 +406,4 @@ public class Sdf10Executor {
} }
return GENERAL_ERROR; return GENERAL_ERROR;
} }
private static record Trio(SDf10 sDf10, Statement stmt, Security security) {}
} }