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.SessionContinueEvent;
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.registry.*;
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
@ -138,14 +137,6 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean {
callback(StatementRequest.class)
.setConsumer(sdf06Executor::execute)
.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)
.setConsumer(event -> clearingService.executeSTrade())
.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.cud.balance.ExportToFileRequest;
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.gateway.AssetOperationApprovalRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.gateway.SingleAssetResponse;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.service.AssetTrio;
@ -29,16 +27,14 @@ import ru.spcex.clearing.service.registry.DmiService;
import ru.spcex.clearing.service.schedule.TradingTimeService;
import ru.spcex.clearing.service.validation.Sdf06NewValidationRule;
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.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId;
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.utils.enumeration.EnumMessage;
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.ValidatorImpl;
@ -78,19 +74,19 @@ public class Sdf06Executor {
//отказ от gateway либо недостаточно средств у нас
private final static BigDecimal errorResult3 = new BigDecimal("3");
private final static BigDecimal errorResult4 = new BigDecimal("4");
private final GatewayBatchRequester gateway;
//not thread safe: used in EventReceiver single thread executor
//переменная контроля за тем что от гейтвея пришел ответ именно на наш запрос
LinkedList<BaseRequest<StatementRequest>> cachedRequests = new LinkedList<>();
private Long sdf06GroupId;
private Long sdf07GroupId;
@Autowired
public Sdf06Executor(ImdgProvider imdgProvider,
IMessageResolver messageResolver,
@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.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.class);
this.tradingTimeService = tradingTimeService;
@ -106,6 +102,8 @@ public class Sdf06Executor {
this.filenameObtainer = filenameObtainer;
this.dmiService = dmiService;
this.assets = assets;
this.gateway = gateway;
this.gateway.setName("Sdf06");
}
public void execute(BaseRequest<StatementRequest> systemRequest) {
@ -118,12 +116,13 @@ public class Sdf06Executor {
log.info("start processing SDF06: groupId={} file={}", groupId, fileName);
Instant now = Instant.now();
Long sdf07GroupId = idGenerator.nextId();
if (sdf06GroupId != null) {
log.warn("currently awaiting gateway response for groupId: {}. adding to cache groupId {}", sdf06GroupId, groupId);
cachedRequests.add(systemRequest);
return;
}
//if (sdf06GroupId != null) {
// log.warn("currently awaiting gateway response for groupId: {}. adding to cache groupId {}", sdf06GroupId, groupId);
// cachedRequests.add(systemRequest);
// return;
//}
Collection<AssetOperationRequest> requests = new ArrayList<>();
Collection<Trio> records = new ArrayList<>();
boolean sdf07WasCreated = false;
for (SDf06 sDf06 : sdfs) {
IValidator validator = sDf06Validator.apply(sDf06);
@ -157,6 +156,7 @@ public class Sdf06Executor {
log.debug("Statement created: {}", stmt.getId());
if (InOutDirection.out.getKey().equals(stmt.getInOutDirection()) && tradingTimeService.isTradingTime()) {
requests.add(GatewayRequestCreator.from(stmt, company.getTradingCode(), tcr.getCode()));
records.add(new Trio(sDf06, stmt, tcr));
} else {
processedApproved(stmt, sDf06, now, sdf07GroupId);
if (tcr != null) {
@ -171,19 +171,80 @@ public class Sdf06Executor {
}
}
if (requests.size() > 0) {
this.sdf06GroupId = groupId;
this.sdf07GroupId = sdf07GroupId;
AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest();
assetOperationListRequest.setAssetOperationRequests(requests);
kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest);
} else if (sdf07WasCreated) {
Instant updatedTime = Instant.now();
Optional<List<SingleAssetResponse>> gatewayResponse = gateway.gatewayRequestAndWaitBatch(requests);
if (gatewayResponse.isEmpty()) {
log.error("couldn't retrieve SDF06 gateway response for groupId={}/sdf07.groupId={}.", groupId, sdf07GroupId);
sdf07WasCreated = true;
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);
}
if (cachedRequests.size() > 0) {
BaseRequest<StatementRequest> statementRequest = cachedRequests.removeFirst();
execute(statementRequest);
}
}
private void processedApproved(Statement stmt, SDf06 sDf06, Instant time) {
processedApproved(stmt, sDf06, time, sdf07GroupId);
}
//private void processedApproved(Statement stmt, SDf06 sDf06, Instant time) {
// processedApproved(stmt, sDf06, time, sdf07GroupId);
//}
private void processedApproved(Statement stmt, SDf06 sDf06, Instant time, Long sdf07GenerationId) {
log.trace("statement.id={}, sdf07.id={}, sdf06.id={} executed",
@ -227,92 +288,87 @@ public class Sdf06Executor {
return result;
}
public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) {
//может прийти неограниченно позже 06го, после того как прошел клиринг например...
Instant now = Instant.now();
String fileName = null;
boolean requestIsIntendedForSdf06 = false;
for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) {
//получаем запрос для текущей группы sdf06
//находим группу
Long statementId = gatewayMsg.getEntityId();
ImdgPredicateBuilder pb = statementImdg.predicateBuilder();
Statement stmt = statementImdg.getFirstObjectByPredicate(
pb.and(
pb.equals("id", statementId),
pb.equals("inOutSDfType", InOutSDfType.type6.getKey())
)
);
if (stmt == null) {
log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type6.getKey());
return;
}
requestIsIntendedForSdf06 = true;
Long sdf06Id = stmt.getInSDfId();
SDf06 sdf06 = sdf06Imdg.getSingleObjectByID(sdf06Id);
if (fileName == null && !TextUtil.isEmpty(sdf06.getFileName())) {
fileName = sdf06.getFileName();
}
if (sdf06 == null) {
log.error("Sdf06.id {} not found by statement.id {}", sdf06Id, statementId);
return;
}
Long groupId = sdf06.getGenerationId();
//сверяем группу SDF06 пришедшего запроса с ожидаемой
if (sdf06GroupId == null || !sdf06GroupId.equals(groupId)) {
log.error("do not currently waiting for gateway response for statement.id {} sdf06 groupId {}; waiting for {}",
statementId,
groupId,
sdf06Id);
return;
}
Instant updatedTime = Instant.now();
if (gatewayMsg.isApproved()) {
processedApproved(stmt, sdf06, updatedTime);
Optional<TradingClearingRegistry> tcr = searchTcrOnGatewayResponse(sdf06);
dmiService.setProcContract(tcr.map(SpcexObjectBase::getId).orElse(null),
CurrencyCode.RUB.getKey(),
safeBD(sdf06.getSum()),
sdf06.getNumber().toString()
);
if (tcr.isPresent()) {
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));
}
} 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);
}
}
if (requestIsIntendedForSdf06) {
sendToExporter(sdf07GroupId, fileName);
log.debug("All gateway responses received for SDF06 groupId {}. SDF07 groupId {}", sdf06GroupId, sdf07GroupId);
clearContext();
} else {
log.debug("gateway response was not for SDF06 executor");
}
if (cachedRequests.size() > 0) {
BaseRequest<StatementRequest> statementRequest = cachedRequests.removeFirst();
execute(statementRequest);
}
}
private void clearContext() {
this.sdf06GroupId = null;
this.sdf07GroupId = null;
}
// public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) {
// //может прийти неограниченно позже 06го, после того как прошел клиринг например...
// Instant now = Instant.now();
// String fileName = null;
// boolean requestIsIntendedForSdf06 = false;
// for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) {
// //получаем запрос для текущей группы sdf06
// //находим группу
// Long statementId = gatewayMsg.getEntityId();
// ImdgPredicateBuilder pb = statementImdg.predicateBuilder();
// Statement stmt = statementImdg.getFirstObjectByPredicate(
// pb.and(
// pb.equals("id", statementId),
// pb.equals("inOutSDfType", InOutSDfType.type6.getKey())
// )
// );
// if (stmt == null) {
// log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type6.getKey());
// return;
// }
// requestIsIntendedForSdf06 = true;
//
// Long sdf06Id = stmt.getInSDfId();
// SDf06 sdf06 = sdf06Imdg.getSingleObjectByID(sdf06Id);
// if (fileName == null && !TextUtil.isEmpty(sdf06.getFileName())) {
// fileName = sdf06.getFileName();
// }
// if (sdf06 == null) {
// log.error("Sdf06.id {} not found by statement.id {}", sdf06Id, statementId);
// return;
// }
//
// Long groupId = sdf06.getGenerationId();
// //сверяем группу SDF06 пришедшего запроса с ожидаемой
// if (sdf06GroupId == null || !sdf06GroupId.equals(groupId)) {
// log.error("do not currently waiting for gateway response for statement.id {} sdf06 groupId {}; waiting for {}",
// statementId,
// groupId,
// sdf06Id);
// return;
// }
//
// Instant updatedTime = Instant.now();
// if (gatewayMsg.isApproved()) {
// processedApproved(stmt, sdf06, updatedTime);
// Optional<TradingClearingRegistry> tcr = searchTcrOnGatewayResponse(sdf06);
// dmiService.setProcContract(tcr.map(SpcexObjectBase::getId).orElse(null),
// CurrencyCode.RUB.getKey(),
// safeBD(sdf06.getSum()),
// sdf06.getNumber().toString()
// );
// if (tcr.isPresent()) {
// 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));
// }
// } 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);
// }
// }
// if (requestIsIntendedForSdf06) {
// sendToExporter(sdf07GroupId, fileName);
// log.debug("All gateway responses received for SDF06 groupId {}. SDF07 groupId {}", sdf06GroupId, sdf07GroupId);
// clearContext();
// } else {
// log.debug("gateway response was not for SDF06 executor");
// }
// if (cachedRequests.size() > 0) {
// BaseRequest<StatementRequest> statementRequest = cachedRequests.removeFirst();
// execute(statementRequest);
// }
// }
private void sendToExporter(Long generationId, String fileName) {
ExportToFileRequest exportRequest = new ExportToFileRequest();
@ -389,4 +445,6 @@ public class Sdf06Executor {
}
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.Consts;
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.gateway.AssetOperationApprovalRequest;
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.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.schedule.TradingTimeService;
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.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.imdg.api.predicate.ImdgPredicateBuilder;
import ru.spcex.platform.imdg.api.predicate.specific.SecuritySelector;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.enumeration.IEnumId;
@ -44,10 +42,7 @@ import ru.spcex.platform.utils.validation.IValidator;
import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.*;
import java.util.function.Function;
import static ru.spcex.platform.utils.number.BigDecimalUtil.safeBD;
@ -70,10 +65,12 @@ public class Sdf10Executor {
private final KafkaSender kafkaSender;
private final SecuritySelector<Security> scrtSlct;
private final AssetTBFProcessing assets;
private final GatewayBatchRequester gateway;
private final static String OK = "OK";
private final static String SYNTAX_ERROR = "Синтаксическая ошибка (файл сформирован неверно)";
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_COMPANY = "Не найдена компания";
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 GENERAL_ERROR = "Общая ошибка обработки SDF10";
private Long sdf10GroupId;
private Long sdf11GroupId;
@Autowired
public Sdf10Executor(ImdgProvider imdgProvider,
RegistryManager rgsMng, IMessageResolver messageResolver,
@Qualifier("sdf10Validator") Function<SDf10, IValidator> sDf10Validator,
TradingTimeService tradingTimeService, KafkaSender kafkaSender, AssetTBFProcessing assets) {
TradingTimeService tradingTimeService, KafkaSender kafkaSender, AssetTBFProcessing assets, GatewayBatchRequester gateway) {
this.imdgProvider = imdgProvider;
this.statementImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Statement, Statement.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.scrtSlct = new SecuritySelector<>(imdgProvider, Security.class);
this.assets = assets;
this.gateway = gateway;
this.gateway.setName("sdf10");
}
public void execute(BaseRequest<StatementRequest> systemRequest) {
@ -116,16 +112,17 @@ public class Sdf10Executor {
Instant now = Instant.now();
Long sdf11GroupId = idGenerator.nextId();
log.info("processing SDF10 groupId: {} sdf10s: {}", groupId, sdfs.size());
if (sdf10GroupId != null) {
log.warn("currently awaiting gateway response for groupId: {}. skipping groupId {}", sdf10GroupId, groupId);
sdfs.forEach(sdf10 -> {
SDf11 errorSdf11 = createSdf11(sdf10, sdf11GroupId, now, PREVIOUS_REQUEST_NOT_PROCESSED);
this.sdf11Imdg.insert(errorSdf11);
});
sendToExporter(sdf11GroupId);
return;
}
//if (sdf10GroupId != null) {
// log.warn("currently awaiting gateway response for groupId: {}. skipping groupId {}", sdf10GroupId, groupId);
// sdfs.forEach(sdf10 -> {
// SDf11 errorSdf11 = createSdf11(sdf10, sdf11GroupId, now, PREVIOUS_REQUEST_NOT_PROCESSED);
// this.sdf11Imdg.insert(errorSdf11);
// });
// sendToExporter(sdf11GroupId);
// return;
//}
Collection<AssetOperationRequest> requests = new ArrayList<>();
List<Trio> records = new ArrayList<>();
boolean sdf11WasCreated = false;
for (SDf10 sDf10 : sdfs) {
IValidator validator = sDf10Validator.apply(sDf10);
@ -165,6 +162,7 @@ public class Sdf10Executor {
log.debug("Statement created: {}", stmt.getId());
if (InOutDirection.out.getKey().equals(stmt.getInOutDirection()) && tradingTimeService.isTradingTime()) {
requests.add(requestFromStatement(stmt, company.getTradingCode(), tcr.getCode(), sDf10.getSecurityCode()));
records.add(new Trio(sDf10, stmt, security));
} else {
SDf11 sDf11 = processedApproved(stmt, sDf10, now, sdf11GroupId);
BigDecimal amount = InOutDirection.in.equalsByKey(stmt.getInOutDirection()) ?
@ -178,12 +176,65 @@ public class Sdf10Executor {
}
}
if (requests.size() > 0) {
this.sdf10GroupId = groupId;
this.sdf11GroupId = sdf11GroupId;
AssetOperationListRequest assetOperationListRequest = new AssetOperationListRequest();
assetOperationListRequest.setAssetOperationRequests(requests);
kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest);
} else if (sdf11WasCreated) {
Instant updatedTime = Instant.now();
Optional<List<SingleAssetResponse>> gatewayResponse = gateway.gatewayRequestAndWaitBatch(requests);
if (gatewayResponse.isEmpty()) {
log.error("couldn't retrieve SDF10 gateway response for groupId={}/sdf11.groupId={}", groupId, sdf11GroupId);
sdf11WasCreated = true;
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);
}
}
@ -229,10 +280,6 @@ public class Sdf10Executor {
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) {
log.trace("statement.id={}, sdf11.id={}, sdf10.id={} executed",
stmt.getId(),
@ -247,80 +294,75 @@ public class Sdf10Executor {
return sdf11;
}
public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) {
Instant now = Instant.now();
boolean requestIsIntendedForSdf10 = false;
for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) {
//получаем запрос для текущей группы sdf10
//находим группу
Long statementId = gatewayMsg.getEntityId();
ImdgPredicateBuilder pb = statementImdg.predicateBuilder();
Statement stmt = statementImdg.getFirstObjectByPredicate(
pb.and(
pb.equals("id", statementId),
pb.equals("inOutSDfType", InOutSDfType.type10.getKey())
)
);
if (stmt == null) {
log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type10.getKey());
return;
}
requestIsIntendedForSdf10 = true;
Long sdf10Id = stmt.getInSDfId();
SDf10 sdf10 = sdf10Imdg.getSingleObjectByID(sdf10Id);
if (sdf10 == null) {
log.error("Sdf10.id {} not found by statement.id {}", sdf10Id, statementId);
return;
}
Long groupId = sdf10.getGenerationId();
//сверяем группу SDF10 пришедшего запроса с ожидаемой
if (sdf10GroupId == null || !sdf10GroupId.equals(groupId)) {
log.error("do not currently waiting for gateway response for statement.id {} sdf10 groupId {}; waiting for {}",
statementId,
groupId,
sdf10Id);
return;
}
Instant updatedTime = Instant.now();
if (gatewayMsg.isApproved()) {
Security security = scrtSlct.selectSecurityById(stmt.getSecurityId());
createDs_iResponseFromGateway(
stmt.getAccountId(),
// stmt.getAddresseeId(),
security != null ? security.getSecuritySymbol() : null,
stmt.getAmount().negate(),
sdf10.getOutDocument());
//tcrId, companyId, securitySymbol
processedApproved(stmt, sdf10, updatedTime);
} 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(updatedTime);
stmt.setOutSDfId(sdf11.getId());
statementImdg.update(stmt);
}
}
if (requestIsIntendedForSdf10) {
sendToExporter(sdf11GroupId);
log.debug("All gateway responses received for SDF10 groupId {}. SDF11 groupId {}", sdf10GroupId, sdf11GroupId);
clearContext();
} else {
log.debug("gateway response was not for SDF10 executor");
}
}
private void clearContext() {
this.sdf10GroupId = null;
this.sdf11GroupId = null;
}
// public void processGatewayResponse(BaseRequest<AssetOperationApprovalRequest> req) {
// Instant now = Instant.now();
// boolean requestIsIntendedForSdf10 = false;
// for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) {
// //получаем запрос для текущей группы sdf10
// //находим группу
// Long statementId = gatewayMsg.getEntityId();
// ImdgPredicateBuilder pb = statementImdg.predicateBuilder();
// Statement stmt = statementImdg.getFirstObjectByPredicate(
// pb.and(
// pb.equals("id", statementId),
// pb.equals("inOutSDfType", InOutSDfType.type10.getKey())
// )
// );
// if (stmt == null) {
// log.debug("Statement.id {} with type {} not found", statementId, InOutSDfType.type10.getKey());
// return;
// }
// requestIsIntendedForSdf10 = true;
//
// Long sdf10Id = stmt.getInSDfId();
// SDf10 sdf10 = sdf10Imdg.getSingleObjectByID(sdf10Id);
// if (sdf10 == null) {
// log.error("Sdf10.id {} not found by statement.id {}", sdf10Id, statementId);
// return;
// }
//
// Long groupId = sdf10.getGenerationId();
// //сверяем группу SDF10 пришедшего запроса с ожидаемой
// if (sdf10GroupId == null || !sdf10GroupId.equals(groupId)) {
// log.error("do not currently waiting for gateway response for statement.id {} sdf10 groupId {}; waiting for {}",
// statementId,
// groupId,
// sdf10Id);
// return;
// }
//
// Instant updatedTime = Instant.now();
// if (gatewayMsg.isApproved()) {
// Security security = scrtSlct.selectSecurityById(stmt.getSecurityId());
// createDs_iResponseFromGateway(
// stmt.getAccountId(),
//// stmt.getAddresseeId(),
// security != null ? security.getSecuritySymbol() : null,
// stmt.getAmount().negate(),
// sdf10.getOutDocument());
// //tcrId, companyId, securitySymbol
// processedApproved(stmt, sdf10, updatedTime);
// } 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(updatedTime);
// stmt.setOutSDfId(sdf11.getId());
// statementImdg.update(stmt);
// }
// }
// if (requestIsIntendedForSdf10) {
// sendToExporter(sdf11GroupId);
// log.debug("All gateway responses received for SDF10 groupId {}. SDF11 groupId {}", sdf10GroupId, sdf11GroupId);
// clearContext();
// } else {
// log.debug("gateway response was not for SDF10 executor");
// }
// }
private void sendToExporter(Long generationId) {
SwtExporterRequest swtReq = new SwtExporterRequest();
@ -406,4 +448,6 @@ public class Sdf10Executor {
}
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.setApprovals(responsesToClearing);
assetOperationApprovalRequest.setInitialRequestId(requestId);
kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION_APPROVAL, assetOperationApprovalRequest);
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;
public class AssetOperationApprovalRequest {
@JsonProperty
private Long initialRequestId;
@JsonProperty
private List<SingleAssetResponse> approvals;
@ -15,4 +17,12 @@ public class AssetOperationApprovalRequest {
public void setApprovals(List<SingleAssetResponse> approvals) {
this.approvals = approvals;
}
public Long getInitialRequestId() {
return initialRequestId;
}
public void setInitialRequestId(Long initialRequestId) {
this.initialRequestId = initialRequestId;
}
}