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

account-service http://jira.mfd.msk:8088/browse/CLS-489 продолжение обработки до SDF_53.

Browse source
This commit is contained in:
AKurakin 2023-08-23 17:06:44 +03:00
parent 523d5b4bc1
commit f253caf3d2
4 changed files with 276 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/ClearingAccountService.java
View file

@ -1,5 +1,7 @@
package ru.spcex.clearing.account.service;
import org.apache.commons.lang3.tuple.MutableTriple;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
@ -51,6 +53,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
private final Logger log = LoggerFactory.getLogger(getClass());
private final KafkaSender kafkaSender;
private final AccountService accountService;
private final SDFProcessService sdfProcessService;
private final ValidationHelper validationHelper;
private final ImdgProvider imdgProvider;
private final IMessageResolver messageResolver;
@ -58,7 +61,6 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
private final Function<ClearingAccountNewRequest, IValidator> clearingAccountNewRequestValidator;
private final Function<ClearingAccountUpdateRequest, IValidator> clearingAccountUpdateRequestValidator;
private final Imdg<SDf52> sdf52Imdg;
private final Imdg<Account> accountImdg;
private final Imdg<Company> companyImdg;
private final Imdg<Relation> relationImdg;
@ -69,6 +71,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
Producer<String, Object> kafkaResponseQueue,
KafkaSender kafkaSender,
AccountService accountService,
SDFProcessService sdfProcessService,
ValidationHelper validationHelper,
ImdgProvider imdgProvider,
IMessageResolver messageResolver,
@ -80,6 +83,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
super(kafkaQueue, kafkaResponseQueue);
this.kafkaSender = kafkaSender;
this.accountService = accountService;
this.sdfProcessService = sdfProcessService;
this.validationHelper = validationHelper;
this.imdgProvider = imdgProvider;
this.messageResolver = messageResolver;
@ -87,7 +91,6 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
this.clearingAccountNewRequestValidator = clearingAccountNewRequestValidator;
this.clearingAccountUpdateRequestValidator = clearingAccountUpdateRequestValidator;
this.sdf52Imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf52, SDf52.class);
this.accountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Account, Account.class);
this.companyImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class);
this.relationImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Relation, Relation.class);
@ -314,22 +317,6 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
return serviceType;
}
/**
* @param status sdf.getStatus()
* @return AccountStatus или null
*/
AccountStatus parseSdf52Status(Long status) {
if (Long.valueOf(1L).equals(status)) {
return AccountStatus.ACTIVE;
} else if (Long.valueOf(0L).equals(status)) {
return AccountStatus.BLOCKED;
}
if (Long.valueOf(2L).equals(status)) {
return AccountStatus.CLOSE;
}
return null;
}
public RequestInfoUpdate accountUpdateSdf52(BaseRequest<StatementRequest> systemRequest) {
log.debug("accountUpdateSdf52 StatementRequest received, id={}", systemRequest.getId());
@ -339,9 +326,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
log.warn("Unsupported table {} received on accountUpdateSdf52. Expected only {}.",
req.getTable(), SdfTable.SDF_52);
}
Collection<SDf52> sdfs = sdf52Imdg.getCollectionObjectsByFieldValues(Map.of(
"generationId", groupId
));
Collection<SDf52> sdfs = sdfProcessService.sdfsByGroupId(groupId);
if (sdfs.isEmpty()) {
log.info("Do not processing SDF52: S_DF52 not found by groupId={}", groupId);
return null;
@ -349,17 +334,20 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
log.info("start processing {} SDF52: groupId={}", sdfs.size(), groupId);
List<Pair<SDf52, Account>> toUpdate = new ArrayList<>();
List<Triple<SDf52, Account, String>> toProcessSDF53 = new ArrayList<>();
{ // 1. Выборка данных
for (SDf52 sDf52 : sdfs) {
if (parseSdf52Status(sDf52.getStatus()) == null) {
if (sdfProcessService.parseSdf52Status(sDf52.getStatus()) == null) {
String msg = messageResolver.resolve(new EnumMessage(AccountError.WrongFieldValue, "status"));
log.warn("By generationId={} s_df52[{}] error: {}", groupId, sDf52.getId(), msg);
toProcessSDF53.add(new MutableTriple<>(sDf52, null, SDFProcessService.SDF_STATUS_ERROR));
continue;
}
Company company = sDf52.getDeal() == null ? null : companyImdg.getSingleObjectByFieldValues(Map.of("tradingCode", sDf52.getDeal()));
if (company == null) {
String msg = messageResolver.resolve(new EnumMessage(AccountError.CompanyNotFound, sDf52.getDeal()));
log.warn("By generationId={} s_df52[{}] error: {}", groupId, sDf52.getId(), msg);
toProcessSDF53.add(new MutableTriple<>(sDf52, null, SDFProcessService.SDF_STATUS_ERROR_COMPANY_NOT_FOUND));
continue;
}
String serviceType = serviceTypeByCMCOfCompany(company.getId());
@ -370,6 +358,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
if (relation == null) {
String msg = messageResolver.resolve(new EnumMessage(AccountError.ClearingCategoryNotFound, company.getTradingCode(), serviceType));
log.warn("By generationId={} s_df52[{}] error: {}", groupId, sDf52.getId(), msg);
toProcessSDF53.add(new MutableTriple<>(sDf52, null, SDFProcessService.SDF_STATUS_ERROR_COMPANY_NOT_FOUND));
continue;
}
Account account = sDf52.getAccount() == null ? null : accountImdg.getFirstObjectByFieldValues(Map.of(
@ -380,6 +369,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
if (account == null) {
String msg = messageResolver.resolve(new EnumMessage(AccountError.AccountNotFound, sDf52.getAccount()));
log.warn("By generationId={} s_df52[{}] error: {}", groupId, sDf52.getId(), msg);
toProcessSDF53.add(new MutableTriple<>(sDf52, null, SDFProcessService.SDF_STATUS_ERROR_COMPANY_NOT_FOUND));
continue;
}
toUpdate.add(new Pair<>(sDf52, account));
@ -397,11 +387,13 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
for (Pair<SDf52, Account> item : toUpdate) {
SDf52 sdf = item.getFirst();
Account account = item.getSecond();
AccountStatus newStatus = parseSdf52Status(sdf.getStatus());
AccountStatus newStatus = sdfProcessService.parseSdf52Status(sdf.getStatus());
if (newStatus == null) {
throw new IllegalArgumentException("Can not parse sdf status " + sdf.getStatus());
}
toProcessSDF53.add(new MutableTriple<>(sdf, account, SDFProcessService.SDF_STATUS_OK));
if (!newStatus.equalsByKey(account.getStatus())) {
// Обновление счёта
account.setStatus(newStatus.getKey());
account.setUpdated(now);
accountImdg.update(account);
@ -418,19 +410,25 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
imdgTransaction.rollbackTransaction();
}
}
if (txOk) {
sdfProcessService.process(req, toProcessSDF53);
}
log.debug("successfully processed, grouping id={}. Updated {} of {} accounts.",
groupId, countOfUpdated, toUpdate.size());
return null;
}
public void sendStatementRequestBack(Long groupingSdf01Id, Long groupSdf02Id, List<AccountSdfToStatementRequestPart> results) {
final String destination = Consts.STATEMENT_PROCESS;
StatementRequest request = new StatementRequest();
request.setGroupId(groupingSdf01Id);
request.setChildGenerationId(groupSdf02Id);
request.setAccountCreationResults(results);
request.setContinueSdf(true);
request.setTable(SdfTable.SDF_01); // по нему запрос получили
log.debug("Send message to kafka \"{}\": {}", Consts.STATEMENT_PROCESS, LogFormatter.toStringWrapper(request));
kafkaSender.sendRequestToQueue(Consts.STATEMENT_PROCESS, request);
log.debug("Send message to kafka \"{}\": {}", destination, LogFormatter.toStringWrapper(request));
kafkaSender.sendRequestToQueue(destination, request);
}
}

143
clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/SDFProcessService.java Normal file
View file

@ -0,0 +1,143 @@
package ru.spcex.clearing.account.service;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import ru.clearing.classes.statics.data.account.Account;
import ru.clearing.classes.statics.data.sdf.SDf52;
import ru.clearing.classes.statics.data.sdf.SDf53;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
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.service.sender.KafkaSender;
import ru.spcex.platform.enumeration.AccountStatus;
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.ImdgTransaction;
import java.time.Instant;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@Service
public class SDFProcessService {
public static final String SDF_STATUS_OK = "0"; // (Операция выполнена успешно) - если account обновлена по sDf52;
public static final String SDF_STATUS_ERROR_REPEAT = "1"; // (Ошибка. Попытка повторно исполнить операцию)
public static final String SDF_STATUS_ERROR_COMPANY_NOT_FOUND = "2"; // (Ошибка. Участник не найден) - если получена ошибка (5013) "Компания %s не найдена" (т.е. account НЕ обновлена по sDf52);
public static final String SDF_STATUS_ERROR_LIMIT_SUMM = "3"; // (Ошибка. Сумма списания превышает сумму средств на торговом счете участника в ТС)
public static final String SDF_STATUS_ERROR_NO_TRADE = "4"; // (Ошибка. Торги не идут)
public static final String SDF_STATUS_ERROR = "9"; // (Другие ошибки, выявленные в КС) - если получены другие ошибки (т.е. account НЕ обновлена по sDf52).
final protected Logger log = LoggerFactory.getLogger(getClass());
protected final Producer<String, Object> kafkaResponseQueue;
protected final KafkaSender kafkaSender;
protected final ImdgProvider imdgProvider;
protected final ImdgId idGenerator;
private final Imdg<SDf52> sdf52Imdg;
public SDFProcessService(
Producer<String, Object> kafkaResponseQueue,
KafkaSender kafkaSender,
ImdgProvider imdgProvider) {
this.kafkaResponseQueue = kafkaResponseQueue;
this.kafkaSender = kafkaSender;
this.imdgProvider = imdgProvider;
this.idGenerator = imdgProvider.getImdgIdGenerator();
this.sdf52Imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf52, SDf52.class);
}
public Collection<SDf52> sdfsByGroupId(Long generationId) {
Collection<SDf52> sdfs = sdf52Imdg.getCollectionObjectsByFieldValues(Map.of(
"generationId", generationId
));
return sdfs;
}
public void process(StatementRequest req, List<Triple<SDf52, Account, String>> toProcessSDF53) {
final Long generationId = req.getGroupId();
log.info("Create S_DF53 generationId={} by {} S_DF_52",
generationId, toProcessSDF53.size());
ImdgTransaction imdgTransaction = imdgProvider.newTransaction();
boolean txOk = false;
imdgTransaction.beginTransaction();
try { // 2. обновление данных, в транзакции
Imdg<SDf53> sdf53Imdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_SDf53, SDf53.class);
Instant now = Instant.now();
for (Triple<SDf52, Account, String> item : toProcessSDF53) {
SDf53 sDf53 = createBy(item.getLeft(), item.getRight(), now, generationId);
sdf53Imdg.insert(sDf53);
}
txOk = true;
} finally {
if (txOk) {
imdgTransaction.commitTransaction();
} else {
log.debug("failed create new SDF53, rollback transaction. sdf52 generationId={}", generationId);
imdgTransaction.rollbackTransaction();
}
}
log.info("Send export command for SDF53 generationId={}", generationId);
messageStatementToExport53(generationId);
}
SDf53 createBy(SDf52 sdf52, String result, Instant now, Long newGenerationId) {
SDf53 newSdf = new SDf53();
newSdf.setId(idGenerator.nextId());
newSdf.setAccName(sdf52.getAcc_name());
newSdf.setAccount(sdf52.getAccount());
newSdf.setDeal(sdf52.getDeal());
newSdf.setStatus(sdf52.getStatus());
newSdf.setResult(result);
newSdf.setGenerationTime(now);
newSdf.setGenerationId(sdf52.getGenerationId());
if (!Objects.equals(newGenerationId, newSdf.getGenerationId())) { // never
log.warn("Different GenerationId={} for sdf53[{}] and GenerationId={} for group of sdf52",
newSdf.getGenerationId(), newSdf.getId(), newGenerationId
);
}
newSdf.setInSDfId(sdf52.getId());
return newSdf;
}
/**
* @param status sdf.getStatus()
* @return AccountStatus или null
*/
public AccountStatus parseSdf52Status(Long status) {
if (Long.valueOf(1L).equals(status)) {
return AccountStatus.ACTIVE;
} else if (Long.valueOf(0L).equals(status)) {
return AccountStatus.BLOCKED;
}
if (Long.valueOf(2L).equals(status)) {
return AccountStatus.CLOSE;
}
return null;
}
void messageStatementToExport53(Long groupId) {
final String destination = Consts.EXPORT_PROCESS; // dbf-exporter
ExportToFileRequest request = new ExportToFileRequest();
request.setNameOfTable("DF-53"); // SdfTable.SDF_53
request.setSdfGroupId(groupId);
request.setFileName(null);
Long msgId = kafkaSender.sendRequestToQueue(destination, request);
log.debug("Send ExportToFileRequest({}, {}) message id={} to kafka \"{}\"",
groupId, request.getNameOfTable(), msgId, destination);
}
}

9
clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/ClearingAccountServiceTest.java
View file

@ -225,15 +225,6 @@ class ClearingAccountServiceTest {
ACCOUNT_MATCHER.assertMatch(accountResult, predictableAccount);
}
@Test
void parseSdf52Status() {
Assertions.assertEquals(AccountStatus.BLOCKED, clearingAccountService.parseSdf52Status(0L));
Assertions.assertEquals(AccountStatus.ACTIVE, clearingAccountService.parseSdf52Status(1L));
Assertions.assertEquals(AccountStatus.CLOSE, clearingAccountService.parseSdf52Status(2L));
Assertions.assertEquals(null, clearingAccountService.parseSdf52Status(3L));
Assertions.assertEquals(null, clearingAccountService.parseSdf52Status(null));
}
// /**
// * {@link ClearingAccountService#accountNewSdf01(BaseRequest)}<br>
// * Тест проверяет создание сущности {@link BaseRequest} в Hazelcast при передаче из Apache Kafka.<br>

110
clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/SDFProcessServiceTest.java Normal file
View file

@ -0,0 +1,110 @@
package ru.spcex.clearing.account.service;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import ru.clearing.classes.statics.data.account.Account;
import ru.clearing.classes.statics.data.account.ClearingAccount;
import ru.clearing.classes.statics.data.company.ClearingMemberCategory;
import ru.clearing.classes.statics.data.company.Company;
import ru.clearing.classes.statics.data.company.relation.Relation;
import ru.clearing.platform.dictionary.AccountTypeDictionary;
import ru.clearing.platform.dictionary.ClearingAccountTypeDictionary;
import ru.spcex.clearing.account.config.BeanConfiguration;
import ru.spcex.clearing.account.config.validation.ValidationConfig;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.test.TestObjectCreator;
import ru.spcex.clearing.test.config.ImdgTestConfig;
import ru.spcex.clearing.test.config.KafkaTestConfig;
import ru.spcex.platform.enumeration.*;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService;
import javax.annotation.PostConstruct;
import java.util.Map;
import static ru.spcex.clearing.account.utils.MatcherFactory.usingIgnoringFieldsComparator;
import static ru.spcex.clearing.test.TestUtils.*;
@ExtendWith(SpringExtension.class)
@ContextConfiguration(classes = {
BeanConfiguration.class,
ValidationConfig.class,
SDFProcessService.class,
ImdgTestConfig.class,
KafkaTestConfig.class})
class SDFProcessServiceTest {
@Autowired
SDFProcessService sdfProcessService;
@Autowired
@Qualifier("hazelcastServiceTest")
private HazelcastService hazelcastServiceTest;
@Captor
private ArgumentCaptor<ProducerRecord> producerRecord;
@SpyBean
private MockProducer<String, Object> producer;
@Autowired
@Qualifier("mockProducer")
protected Producer<String, Object> mockProducer;
private Imdg<ClearingAccount> clearingAccountImdg;
private Imdg<Account> accountImdg;
private Imdg<Company> companyImdg;
private Imdg<ClearingAccountTypeDictionary> clearingAccountTypeDictionaryImdg;
private Imdg<AccountTypeDictionary> accountTypeDictionaryImdg;
private Imdg<ClearingMemberCategory> clearingMemberCategoryImdg;
private Imdg<Relation> relationImdg;
@PostConstruct
private void init() {
hazelcastServiceTest.waitAvailable();
clearingAccountImdg = hazelcastServiceTest.getImdg(
IMDGDistributedNames.Map_ClearingAccount, ClearingAccount.class
);
accountImdg = hazelcastServiceTest.getImdg(
IMDGDistributedNames.Map_Account, Account.class
);
companyImdg = hazelcastServiceTest.getImdg(
IMDGDistributedNames.Map_Company, Company.class
);
accountTypeDictionaryImdg = hazelcastServiceTest.getImdg(
IMDGDistributedNames.Map_AccountTypeDictionary, AccountTypeDictionary.class
);
clearingAccountTypeDictionaryImdg = hazelcastServiceTest.getImdg(
IMDGDistributedNames.Map_ClearingAccountTypeDictionary, ClearingAccountTypeDictionary.class
);
clearingMemberCategoryImdg = hazelcastServiceTest.getImdg(
IMDGDistributedNames.Map_ClearingMemberCategory, ClearingMemberCategory.class
);
relationImdg = hazelcastServiceTest.getImdg(
IMDGDistributedNames.Map_Relation, Relation.class
);
new TestObjectCreator(hazelcastServiceTest).createUserAdmin(1000L);
}
@Test
void parseSdf52Status() {
Assertions.assertEquals(AccountStatus.BLOCKED, sdfProcessService.parseSdf52Status(0L));
Assertions.assertEquals(AccountStatus.ACTIVE, sdfProcessService.parseSdf52Status(1L));
Assertions.assertEquals(AccountStatus.CLOSE, sdfProcessService.parseSdf52Status(2L));
Assertions.assertEquals(null, sdfProcessService.parseSdf52Status(3L));
Assertions.assertEquals(null, sdfProcessService.parseSdf52Status(null));
}
}