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

http://jira.mfd.msk:8088/browse/CLS-606

Browse source
This commit is contained in:
ialbert 2023-12-27 11:32:10 +03:00
parent 97b8a4fd78
commit fd900e27b6
6 changed files with 471 additions and 221 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,7 +15,6 @@ 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;
@ -138,14 +137,6 @@ 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,9 +17,7 @@ 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;
@ -29,16 +27,14 @@ 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.platform.classes.base.SpcexObjectBase; import ru.spcex.clearing.session.stage.impl.GatewayBatchRequester;
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;
@ -78,19 +74,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;
@ -106,6 +102,8 @@ 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) {
@ -118,12 +116,13 @@ 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);
@ -157,6 +156,7 @@ 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,19 +171,80 @@ public class Sdf06Executor {
} }
} }
if (requests.size() > 0) { if (requests.size() > 0) {
this.sdf06GroupId = groupId; Instant updatedTime = Instant.now();
this.sdf07GroupId = sdf07GroupId; Optional<List<SingleAssetResponse>> gatewayResponse = gateway.gatewayRequestAndWaitBatch(requests);
AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest(); if (gatewayResponse.isEmpty()) {
assetOperationListRequest.setAssetOperationRequests(requests); log.error("couldn't retrieve SDF06 gateway response for groupId={}/sdf07.groupId={}.", groupId, sdf07GroupId);
kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest); sdf07WasCreated = true;
} else if (sdf07WasCreated) { records.forEach(trio -> {
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 {
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);
}
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",
@ -227,92 +288,87 @@ 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();
@ -389,4 +445,6 @@ 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,9 +19,7 @@ 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;
@ -29,12 +27,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;
@ -44,10 +42,7 @@ 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.ArrayList; import java.util.*;
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;
@ -70,10 +65,12 @@ 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 = "Не найден ТКР";
@ -83,14 +80,11 @@ 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) { TradingTimeService tradingTimeService, KafkaSender kafkaSender, 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.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class);
@ -105,6 +99,8 @@ 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) {
@ -116,16 +112,17 @@ 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);
@ -165,6 +162,7 @@ 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()) ?
@ -178,12 +176,65 @@ public class Sdf10Executor {
} }
} }
if (requests.size() > 0) { if (requests.size() > 0) {
this.sdf10GroupId = groupId; Instant updatedTime = Instant.now();
this.sdf11GroupId = sdf11GroupId; Optional<List<SingleAssetResponse>> gatewayResponse = gateway.gatewayRequestAndWaitBatch(requests);
AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest(); if (gatewayResponse.isEmpty()) {
assetOperationListRequest.setAssetOperationRequests(requests); log.error("couldn't retrieve SDF10 gateway response for groupId={}/sdf11.groupId={}", groupId, sdf11GroupId);
kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest); sdf11WasCreated = true;
} else if (sdf11WasCreated) { records.forEach(trio -> {
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);
} }
} }
@ -229,10 +280,6 @@ 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(),
@ -247,80 +294,75 @@ 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();
@ -406,4 +448,6 @@ public class Sdf10Executor {
} }
return GENERAL_ERROR; return GENERAL_ERROR;
} }
private static record Trio(SDf10 sDf10, Statement stmt, Security security) {}
} }

146
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/GatewayBatchRequester.java Normal file
View file

@ -0,0 +1,146 @@
package ru.spcex.clearing.session.stage.impl;
import org.apache.kafka.clients.consumer.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import ru.spcex.clearing.config.element.ClearingServiceSettings;
import ru.spcex.clearing.config.element.SessionStageSettings;
import ru.spcex.clearing.error.ClearingError;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.Consts;
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.gateway.AssetOperationApprovalRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.gateway.SingleAssetResponse;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.log.ExceptionUtils;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
@Component
public class GatewayBatchRequester extends QueueConsumer implements InitializingBean {
private final Logger log = LoggerFactory.getLogger(getClass());
private final KafkaSender kafkaSender;
private final Integer GATEWAY_TIMEOUT;
private String name;
private final Lock gatewayLock = new ReentrantLock();
private final Condition gatewayCondition = gatewayLock.newCondition();
private Long kafkaReqId = null;
private List<SingleAssetResponse> batchResponses = null;
@Autowired
public GatewayBatchRequester(@Qualifier("kafkaConsumerGateway") Supplier<Consumer<String, Object>> consumer,
ClearingServiceSettings settings,
KafkaSender kafkaSender) {
super(consumer.get());
this.kafkaSender = kafkaSender;
SessionStageSettings sessionSettings = settings.getSessionStage();
this.GATEWAY_TIMEOUT = sessionSettings != null ? sessionSettings.getInspectionGatewayTimeout() : 10;
}
@Override
public void afterPropertiesSet() throws Exception {
callback(AssetOperationApprovalRequest.class)
.setConsumer(this::receiveGatewayAnswer)
.forDestination(Consts.ASSET_OPERATION_APPROVAL, callbacks::put);
init();
}
public Optional<List<SingleAssetResponse>> gatewayRequestAndWaitBatch(Collection<AssetOperationRequest> requests) {
if (requests == null || requests.size() == 0) {
return Optional.of(Collections.emptyList());
}
List<SingleAssetResponse> responseReceived = null;
Long kafkaReqLogging = null;
gatewayLock.lock();
try {
if (this.kafkaReqId != null) {
log.error("{}.kafkaReqId={} already sent to gateway", getName(), this.kafkaReqId);
throw new IllegalStateException(" forbid reusing GatewayRequester instance in multiple threads");
}
//send Request
AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest();
assetOperationListRequest.setAssetOperationRequests(requests);
this.kafkaReqId = kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest);
log.debug("sent {}.id={} to gateway", getName(), this.kafkaReqId);
boolean gatewayReceived = gatewayCondition.await(GATEWAY_TIMEOUT, TimeUnit.SECONDS);
kafkaReqLogging = this.kafkaReqId;
if (gatewayReceived) {
responseReceived = batchResponses;
batchResponses = null;
} else {
kafkaReqId = null;
}
} catch (InterruptedException e) {
log.error(ExceptionUtils.getStackTrace(e));
} finally {
gatewayLock.unlock();
}
if (responseReceived != null) {
log.debug("{}.id={} answer from gateway received. size: {}", getName(), kafkaReqLogging, responseReceived.size());
return Optional.of(responseReceived);
} else {
log.error("{}.id={} answer from gateway not received.", getName(), kafkaReqLogging);
return Optional.empty();
}
}
private void receiveGatewayAnswer(BaseRequest<AssetOperationApprovalRequest> sysReq) {
AssetOperationApprovalRequest requestPayload = sysReq.getRequestPayload();
Long initialReqId = requestPayload.getInitialRequestId();
log.trace("gateway answer received, BaseRequest.id={} initialRequestId={}", sysReq.getId(), initialReqId);
List<SingleAssetResponse> approvals = requestPayload.getApprovals();
if (approvals == null) {
log.trace("gateway BaseRequest.id={} initialReqId={} approvals null", sysReq.getId(), initialReqId);
return;
}
gatewayLock.lock();
try {
if (kafkaReqId == null) {
log.trace("gateway BaseRequest.id={} initialReqId={} currently not waiting for {} answer", sysReq.getId(), initialReqId, getName());
return;
}
if (Objects.equals(this.kafkaReqId, initialReqId)) {
log.trace("gateway BaseRequest.id={} answer found for {}.id={}", sysReq.getId(), getName(), initialReqId);
this.kafkaReqId = null;
this.batchResponses = requestPayload.getApprovals() != null ? requestPayload.getApprovals() : Collections.emptyList();
gatewayCondition.signal();
}
} finally {
gatewayLock.unlock();
}
}
public String getName() {
return name != null ? name : "entity";
}
public void setName(String name) {
this.name = name;
}
public static EnumMessage mapError(Optional<Boolean> gtwRsp) {
if (gtwRsp.isEmpty()) {
return new EnumMessage(ClearingError.GatewayTimeout);
} else if (!gtwRsp.get()) {
return new EnumMessage(ClearingError.GatewayNotApproved);
} else throw new IllegalArgumentException("gateway response was OK. cannot create error message");
}
}

1
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/OperationService.java
View file

@ -53,6 +53,7 @@ public class OperationService {
} }
AssetOperationApprovalRequest assetOperationApprovalRequest = new AssetOperationApprovalRequest(); AssetOperationApprovalRequest assetOperationApprovalRequest = new AssetOperationApprovalRequest();
assetOperationApprovalRequest.setApprovals(responsesToClearing); assetOperationApprovalRequest.setApprovals(responsesToClearing);
assetOperationApprovalRequest.setInitialRequestId(requestId);
kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION_APPROVAL, assetOperationApprovalRequest); kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION_APPROVAL, assetOperationApprovalRequest);
sentAssets.remove(requestId); sentAssets.remove(requestId);
} }

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

@ -5,6 +5,8 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.List; import java.util.List;
public class AssetOperationApprovalRequest { public class AssetOperationApprovalRequest {
@JsonProperty
private Long initialRequestId;
@JsonProperty @JsonProperty
private List<SingleAssetResponse> approvals; private List<SingleAssetResponse> approvals;
@ -15,4 +17,12 @@ public class AssetOperationApprovalRequest {
public void setApprovals(List<SingleAssetResponse> approvals) { public void setApprovals(List<SingleAssetResponse> approvals) {
this.approvals = approvals; this.approvals = approvals;
} }
public Long getInitialRequestId() {
return initialRequestId;
}
public void setInitialRequestId(Long initialRequestId) {
this.initialRequestId = initialRequestId;
}
} }