diff --git a/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/ClearingAccountService.java b/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/ClearingAccountService.java index 396ab021e..33b070a87 100644 --- a/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/ClearingAccountService.java +++ b/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/ClearingAccountService.java @@ -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 clearingAccountNewRequestValidator; private final Function clearingAccountUpdateRequestValidator; - private final Imdg sdf52Imdg; private final Imdg accountImdg; private final Imdg companyImdg; private final Imdg relationImdg; @@ -69,6 +71,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin Producer 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 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 sdfs = sdf52Imdg.getCollectionObjectsByFieldValues(Map.of( - "generationId", groupId - )); + Collection 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> toUpdate = new ArrayList<>(); + List> 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 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 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); } } diff --git a/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/SDFProcessService.java b/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/SDFProcessService.java new file mode 100644 index 000000000..12213ffab --- /dev/null +++ b/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/SDFProcessService.java @@ -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 kafkaResponseQueue; + protected final KafkaSender kafkaSender; + + protected final ImdgProvider imdgProvider; + protected final ImdgId idGenerator; + + private final Imdg sdf52Imdg; + + public SDFProcessService( + Producer 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 sdfsByGroupId(Long generationId) { + Collection sdfs = sdf52Imdg.getCollectionObjectsByFieldValues(Map.of( + "generationId", generationId + )); + return sdfs; + } + + public void process(StatementRequest req, List> 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 sdf53Imdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_SDf53, SDf53.class); + + Instant now = Instant.now(); + for (Triple 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); + } +} diff --git a/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/ClearingAccountServiceTest.java b/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/ClearingAccountServiceTest.java index e416e16da..2876c4be2 100644 --- a/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/ClearingAccountServiceTest.java +++ b/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/ClearingAccountServiceTest.java @@ -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)}
// * Тест проверяет создание сущности {@link BaseRequest} в Hazelcast при передаче из Apache Kafka.
diff --git a/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/SDFProcessServiceTest.java b/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/SDFProcessServiceTest.java new file mode 100644 index 000000000..819b41379 --- /dev/null +++ b/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/SDFProcessServiceTest.java @@ -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; + @SpyBean + private MockProducer producer; + @Autowired + @Qualifier("mockProducer") + protected Producer mockProducer; + + private Imdg clearingAccountImdg; + private Imdg accountImdg; + private Imdg companyImdg; + private Imdg clearingAccountTypeDictionaryImdg; + private Imdg accountTypeDictionaryImdg; + private Imdg clearingMemberCategoryImdg; + private Imdg 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)); + } +} \ No newline at end of file