clearing-service sdf08 обработка автосоздания счетов. Оказалось что DEPO ужно делать (част 2/2). Поправил транзакции.

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/BankAccountService.java

@@ -55,11 +55,11 @@ public class BankAccountService extends QueueConsumer implements InitializingBea
                               ValidationHelper validationHelper,
                               AccountService accountService,
                               @Qualifier("bankAccountNewRequestValidator")
-                                          Function<BankAccountNewRequest, IValidator> bankAccountNewRequestValidator,
+                                      Function<BankAccountNewRequest, IValidator> bankAccountNewRequestValidator,
                               @Qualifier("bankAccountUpdateRequestValidator")
-                                          Function<BankAccountUpdateRequest, IValidator> bankAccountUpdateRequestValidator,
+                                      Function<BankAccountUpdateRequest, IValidator> bankAccountUpdateRequestValidator,
                               @Qualifier("bankAccountBlockRequestValidator")
-                                          Function<CommonIdRequest, IValidator> bankAccountBlockRequestValidator) {
+                                      Function<CommonIdRequest, IValidator> bankAccountBlockRequestValidator) {
         super(kafkaQueue, kafkaProducer);
         this.bankAccountMap = imdgProvider.getImdg(IMDGDistributedNames.Map_BankAccount, BankAccount.class);
         this.accountMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Account, Account.class);
@@ -115,6 +115,8 @@ public class BankAccountService extends QueueConsumer implements InitializingBea
         boolean txOk = false;
         imdgTransaction.beginTransaction();
         try {
+            Imdg<BankAccount> bankAccountMap = imdgTransaction.getImdg(IMDGDistributedNames.Map_BankAccount, BankAccount.class);
+            Imdg<Account> accountMap = imdgTransaction.getImdg(IMDGDistributedNames.Map_Account, Account.class);
             accountId = accountMap.insert(account);
 
             BankAccount bankAccount = new BankAccount();
@@ -175,6 +177,8 @@ public class BankAccountService extends QueueConsumer implements InitializingBea
         boolean txOk = false;
         imdgTransaction.beginTransaction();
         try {
+            Imdg<BankAccount> bankAccountMap = imdgTransaction.getImdg(IMDGDistributedNames.Map_BankAccount, BankAccount.class);
+            Imdg<Account> accountMap = imdgTransaction.getImdg(IMDGDistributedNames.Map_Account, Account.class);
             accountMap.update(account);
             bankAccountMap.update(bankAccount);
             txOk = true;
@@ -189,7 +193,7 @@ public class BankAccountService extends QueueConsumer implements InitializingBea
                         bankAccount.getId(),
                         account.getId());
                 imdgTransaction.rollbackTransaction();
-            } 
+            }
         }
         log.debug("successfully update, existing bankAccount with id {}", bankAccount.getId());
         return null;
@@ -210,27 +214,10 @@ public class BankAccountService extends QueueConsumer implements InitializingBea
         Account account = accountMap.getSingleObjectByID(bankAccount.getAccountId());
         account.setStatus(ServiceStatus.Blocked.getKey());
         account.setUpdated(Instant.now());
-        
+
-        ImdgTransaction imdgTransaction = imdgProvider.newTransaction();
+        accountMap.update(account);
-        boolean txOk = false;
+        //без изменений bankAccountMap.update(bankAccount);
-        imdgTransaction.beginTransaction();
+
-        try {
-            accountMap.update(account);
-            //без изменений bankAccountMap.update(bankAccount);
-            txOk = true;
-        } finally {
-            if (txOk) {
-                imdgTransaction.commitTransaction();
-                log.debug("successfully processed, new bank account id {}, account id {}",
-                        bankAccount.getId(),
-                        account.getId());
-            } else {
-                log.debug("failed block, bank account id {}, new account id {}",
-                        bankAccount.getId(),
-                        account.getId());
-                imdgTransaction.rollbackTransaction();
-            }
-        }
         log.debug("successfully block, existing bankAccount with id {}", bankAccount.getId());
         return null;
     }

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

@@ -52,8 +52,6 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
     private final AccountService accountService;
     private final ValidationHelper validationHelper;
     private final ImdgProvider imdgProvider;
-    private final Imdg<Account> accountImdg;
-    private final Imdg<ClearingAccount> clearingAccountImdg;
     private final IMessageResolver messageResolver;
     private final RequestHelper requestHelper;
     private final Function<ClearingAccountNewRequest, IValidator> clearingAccountNewRequestValidator;
@@ -77,8 +75,6 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
         this.accountService = accountService;
         this.validationHelper = validationHelper;
         this.imdgProvider = imdgProvider;
-        this.accountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Account, Account.class);
-        this.clearingAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ClearingAccount, ClearingAccount.class);
         this.messageResolver = messageResolver;
         this.requestHelper = requestHelper;
         this.clearingAccountNewRequestValidator = clearingAccountNewRequestValidator;
@@ -126,6 +122,8 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
         imdgTransaction.beginTransaction();
         ClearingAccount clearingAccount = null;
         try {
+            Imdg<Account> accountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_Account, Account.class);
+            Imdg<ClearingAccount> clearingAccountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_ClearingAccount, ClearingAccount.class);
             accountId = accountImdg.insert(account);
 
             clearingAccount = new ClearingAccount();
@@ -161,28 +159,41 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
         if (requestInfoUpdate != null) return requestInfoUpdate;
 
         ClearingAccountUpdateRequest req = userRequest.getRequestPayload();
-        ImdgPredicateBuilder pb = accountImdg.predicateBuilder();
-        ImdgPredicate accountValuePredicate = pb.equals("account", req.getAccount());
-        ImdgPredicate accountStatusPredicate = pb.equals("status", ServiceStatus.Active.getKey());
-        ImdgPredicate accountTypePredicate = pb.equals("accountType", AccountType.Clrn.getKey());
-        ImdgPredicate finalPredicate = pb.and(accountValuePredicate,
-                accountStatusPredicate,
-                accountTypePredicate);
 
-        Account account = accountImdg.getCollectionObjectsByPredicate(finalPredicate).iterator().next();
+        ImdgTransaction imdgTransaction = imdgProvider.newTransaction();
-        Long accountId = account.getId();
+        boolean txOk = false;
+        imdgTransaction.beginTransaction();
+        try {
+            Imdg<Account> accountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_Account, Account.class);
+            Imdg<ClearingAccount> clearingAccountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_ClearingAccount, ClearingAccount.class);
 
-        ClearingAccount clearingAccount = clearingAccountImdg.getSingleObjectByFieldValues(Map.of("accountId", accountId));
+            ImdgPredicateBuilder pb = accountImdg.predicateBuilder();
-        if (clearingAccount == null)
+            ImdgPredicate accountValuePredicate = pb.equals("account", req.getAccount());
-            return requestHelper.makeErrorResponse(userRequest, AccountError.AccountNotFound, account.getAccount());
+            ImdgPredicate accountStatusPredicate = pb.equals("status", ServiceStatus.Active.getKey());
+            ImdgPredicate accountTypePredicate = pb.equals("accountType", AccountType.Clrn.getKey());
+            ImdgPredicate finalPredicate = pb.and(accountValuePredicate,
+                    accountStatusPredicate,
+                    accountTypePredicate);
 
-        Integer statusValue = req.getStatus();
+            Account account = accountImdg.getCollectionObjectsByPredicate(finalPredicate).iterator().next();
-        if (statusValue == 0) account.setStatus(ServiceStatus.Blocked.getKey());
+            Long accountId = account.getId();
-        else if (statusValue == 1) account.setStatus(ServiceStatus.Active.getKey());
-        else if (statusValue == 2) account.setStatus(ServiceStatus.Closed.getKey());
-        account.setUpdated(Instant.now());
 
-        accountImdg.update(account);
+            ClearingAccount clearingAccount = clearingAccountImdg.getSingleObjectByFieldValues(Map.of("accountId", accountId));
+            if (clearingAccount == null)
+                return requestHelper.makeErrorResponse(userRequest, AccountError.AccountNotFound, account.getAccount());
+
+            Integer statusValue = req.getStatus();
+            if (statusValue == 0) account.setStatus(ServiceStatus.Blocked.getKey());
+            else if (statusValue == 1) account.setStatus(ServiceStatus.Active.getKey());
+            else if (statusValue == 2) account.setStatus(ServiceStatus.Closed.getKey());
+            account.setUpdated(Instant.now());
+
+            accountImdg.update(account);
+            txOk = true;
+        } finally {
+            if (txOk) imdgTransaction.commitTransaction();
+            else imdgTransaction.rollbackTransaction();
+        }
 
         return null;
     }
@@ -199,6 +210,9 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
         boolean txOk = false;
         imdgTransaction.beginTransaction();
         try {
+            Imdg<Account> accountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_Account, Account.class);
+            Imdg<ClearingAccount> clearingAccountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_ClearingAccount, ClearingAccount.class);
+
             accountsLoop:
             for (AccountSdfRequestPart accountReq : req.getAccounts()) {
                 Instant now = Instant.now();

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/DepoAccountService.java

@@ -9,10 +9,15 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import ru.clearing.classes.statics.data.account.Account;
 import ru.clearing.classes.statics.data.account.DepoAccount;
+import ru.spcex.clearing.account.errors.AccountError;
 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.account.DepoAccountNewRequest;
+import ru.spcex.clearing.platform.messaging.domain.cud.account.sdf01.AccountSdf01Request;
+import ru.spcex.clearing.platform.messaging.domain.cud.account.sdf01.AccountSdfRequestPart;
+import ru.spcex.clearing.platform.messaging.domain.cud.balance.AccountSdfToStatementRequestPart;
+import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.registry.TradingClearingRegistryNewRequest;
 import ru.spcex.clearing.platform.messaging.serialization.LogFormatter;
 import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
@@ -20,6 +25,7 @@ import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
 import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
 import ru.spcex.clearing.validation.common.ValidationHelper;
 import ru.spcex.platform.enumeration.AccountType;
+import ru.spcex.platform.enumeration.SdfTable;
 import ru.spcex.platform.enumeration.ServiceStatus;
 import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgProvider;
@@ -27,6 +33,8 @@ import ru.spcex.platform.imdg.api.ImdgTransaction;
 import ru.spcex.platform.utils.validation.IValidator;
 
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.function.Function;
 
 @Service
@@ -38,8 +46,6 @@ public class DepoAccountService extends QueueConsumer implements InitializingBea
     private final ImdgProvider imdgProvider;
     private final AccountService accountService;
     private final Function<DepoAccountNewRequest, IValidator> depoAccountNewRequestValidator;
-    private final Imdg<Account> accountImdg;
-    private final Imdg<DepoAccount> depoAccountImdg;
 
     public DepoAccountService(Consumer<String, Object> kafkaQueue,
                               Producer<String, Object> kafkaProducer,
@@ -55,8 +61,6 @@ public class DepoAccountService extends QueueConsumer implements InitializingBea
         this.imdgProvider = imdgProvider;
         this.accountService = accountService;
         this.depoAccountNewRequestValidator = depoAccountNewRequestValidator;
-        this.accountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Account, Account.class);
-        this.depoAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_DepoAccount, DepoAccount.class);
     }
 
     @Override
@@ -64,6 +68,10 @@ public class DepoAccountService extends QueueConsumer implements InitializingBea
         callback(DepoAccountNewRequest.class)
                 .setFunction(this::depoAccountNew)
                 .forDestination(Consts.DESTINATION_DEPO_ACCOUNT_NEW, callbacks::put);
+        callback(AccountSdf01Request.class)
+                .setFunction(this::accountNewSdf08)
+                .forDestination(Consts.ACCOUNT_NEW_SDF08, callbacks::put);
+
         init();
     }
 
@@ -94,6 +102,8 @@ public class DepoAccountService extends QueueConsumer implements InitializingBea
         imdgTransaction.beginTransaction();
         DepoAccount depoAccount = null;
         try {
+            Imdg<Account> accountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Account, Account.class);
+            Imdg<DepoAccount> depoAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_DepoAccount, DepoAccount.class);
             accountId = accountImdg.insert(account);
 
             depoAccount = new DepoAccount();
@@ -120,4 +130,109 @@ public class DepoAccountService extends QueueConsumer implements InitializingBea
         }
         return null;
     }
+
+    public RequestInfoUpdate accountNewSdf08(BaseRequest<AccountSdf01Request> userRequest) {
+        log.debug("AccountSdf08Request received, id={}", userRequest.getId());
+
+        AccountSdf01Request req = userRequest.getRequestPayload();
+        List<AccountSdfToStatementRequestPart> accountToStatement = new ArrayList<>();
+
+        ImdgTransaction imdgTransaction = imdgProvider.newTransaction();
+        List<TradingClearingRegistryNewRequest> toTCRRequests = new ArrayList<>();
+        boolean txOk = false;
+        imdgTransaction.beginTransaction();
+        try {
+            Imdg<Account> accountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Account, Account.class);
+            Imdg<DepoAccount> depoAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_DepoAccount, DepoAccount.class);
+            accountsLoop:
+            for (AccountSdfRequestPart accountReq : req.getAccounts()) {
+
+//                DepoAccountNewRequest req = userRequest.getRequestPayload();
+
+                Instant now = Instant.now();
+                Account account = new Account();
+                account.setAccount(accountReq.getAccount());
+                account.setAccountType(AccountType.Depo.getKey());
+                account.setStatus(ServiceStatus.Active.getKey());
+                account.setCompanyId(accountReq.getCompanyId());
+                account.setCreated(now);
+                account.setUpdated(now);
+                RequestInfoUpdate requestInfoUpdate = accountService.fillAccountFromRelation(account, userRequest.getId(), true);
+                if (requestInfoUpdate != null) {
+                    log.warn("Error fill new account from relation. {}", /*account.getId(),*/ requestInfoUpdate.getMessage());
+                    {
+                        AccountSdfToStatementRequestPart responsePart = new AccountSdfToStatementRequestPart();
+                        responsePart.setSdfId(accountReq.getSdfId());
+                        responsePart.setErrorCode(AccountError.ClearingCategoryNotFound.getId()); // see accountService.fillAccountFromRelation
+                        responsePart.setErrorText(requestInfoUpdate.getMessage());
+                        accountToStatement.add(responsePart);
+                        continue accountsLoop;
+                    }
+                }
+
+                Long depoAccountId = -1L;
+                Long accountId = -1L;
+
+                    accountId = accountImdg.insert(account);
+
+                DepoAccount depoAccount = new DepoAccount();
+                depoAccount.setCompanyId(accountReq.getCompanyId());
+                depoAccount.setAccountId(accountId);
+                depoAccount.setDepoAccountType(accountReq.getAccountType());
+                depoAccountId = depoAccountImdg.insert(depoAccount);
+
+
+
+                log.debug("New account {}, depoAccount {} was created.", accountId, depoAccountId);
+
+                {
+                    TradingClearingRegistryNewRequest request = new TradingClearingRegistryNewRequest();
+                    request.setDepoAccountId(accountId);
+                    request.setCompanyId(depoAccount.getCompanyId());
+//                    request.setTradingClearingRegistryType(cl);
+                    toTCRRequests.add(request);
+                }
+                {
+                    AccountSdfToStatementRequestPart responsePart = new AccountSdfToStatementRequestPart();
+                    responsePart.setSdfId(accountReq.getSdfId());
+                    responsePart.setErrorCode(null);
+                    responsePart.setErrorText(null);
+                    accountToStatement.add(responsePart);
+                }
+            }
+            txOk = true;
+        } finally {
+            if (txOk) {
+                imdgTransaction.commitTransaction();
+            } else {
+                log.debug("failed insert, new clearing accounts. Request id={}", userRequest.getId());
+                imdgTransaction.rollbackTransaction();
+            }
+        }
+
+        if (txOk) {
+            log.debug("Sending {} messages of TradingClearingRegistryNewRequest", toTCRRequests.size());
+            for (TradingClearingRegistryNewRequest tcrReq : toTCRRequests) {
+                log.debug("Send message to kafka \"{}\": {}", Consts.DESTINATION_TRADING_CLEARING_REGISTRY_AUTO_NEW,
+                        LogFormatter.toStringWrapper(tcrReq));
+                Long kafkaId = kafkaSender.sendRequestToQueue(Consts.DESTINATION_TRADING_CLEARING_REGISTRY_AUTO_NEW, tcrReq);
+                log.trace("successfully send request {} to kafka: new clearing account MoneyAccountId {}, DepoAccountId {}",
+                        kafkaId, tcrReq.getMoneyAccountId(), tcrReq.getDepoAccountId());
+            }
+        }
+
+        sendStatementRequestBack(req.getGroupingSdf01Id(), accountToStatement);
+        log.debug("successfully processed, grouping id={}, processed number={}", req.getGroupingSdf01Id(), accountToStatement.size());
+
+        return null;
+    }
+
+    public void sendStatementRequestBack(Long groupingSdf01Id, List<AccountSdfToStatementRequestPart> results) {
+        StatementRequest request = new StatementRequest();
+        request.setGroupId(groupingSdf01Id);
+        request.setAccountCreationResults(results);
+        request.setTable(SdfTable.SDF_08); // по нему запрос получили
+        log.debug("Send message to kafka \"{}\": {}", Consts.STATEMENT_PROCESS, LogFormatter.toStringWrapper(request));
+        kafkaSender.sendRequestToQueue(Consts.STATEMENT_PROCESS, request);
+    }
 }

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/InformationAccountService.java

@@ -17,7 +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.account.InformationAccountNewRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.registry.TradingClearingRegistryNewRequest;
-import ru.spcex.clearing.platform.messaging.domain.cud.reports.ReportRequest;
+import ru.spcex.clearing.platform.messaging.domain.cud.reports.NotificationRequest;
 import ru.spcex.clearing.platform.messaging.serialization.LogFormatter;
 import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
 import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
@@ -25,7 +25,6 @@ import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
 import ru.spcex.clearing.util.services.RequestHelper;
 import ru.spcex.clearing.validation.common.ValidationHelper;
 import ru.spcex.platform.enumeration.AccountType;
-import ru.spcex.platform.enumeration.ReportType;
 import ru.spcex.platform.enumeration.ServiceStatus;
 import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgProvider;
@@ -142,6 +141,9 @@ public class InformationAccountService extends QueueConsumer implements Initiali
         Long accountId = -1L;
         InformationAccount informationAccount = null;
         try {
+            Imdg<InformationAccount> informationAccountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_InformationAccount, InformationAccount.class);
+            Imdg<Account> accountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_Account, Account.class);
+
             accountId = accountImdg.insert(account);
 
             informationAccount = new InformationAccount();
@@ -239,6 +241,8 @@ public class InformationAccountService extends QueueConsumer implements Initiali
         Long accountId = -1L;
         InformationAccount informationAccount = null;
         try {
+            Imdg<InformationAccount> informationAccountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_InformationAccount, InformationAccount.class);
+            Imdg<Account> accountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_Account, Account.class);
             accountId = accountImdg.insert(account);
 
             informationAccount = new InformationAccount();
@@ -284,13 +288,10 @@ public class InformationAccountService extends QueueConsumer implements Initiali
      * Формирование уведолмения о регистрации УК
      */
     protected void sendNotificationToReport(InformationAccount informationAccount, Account account) {
-        //todo актуализировать ТЗ или ReportRequest
+        NotificationRequest request = new NotificationRequest();
-//        ReportRequest request = new ReportRequest();
+        request.setConsumerId(account.getCompanyId());
-//        request.setReportId(ReportType.REGISTRACTION_UK/NEW_INFO_ACCOUNT);
+        log.debug("Send message to kafka \"{}\": {}", Consts.CREATE_NOTIFICATION_NTCR, LogFormatter.toStringWrapper(request));
-//        request.setCompanyId(informationAccount.getCompanyId());
+        kafkaSender.sendRequestToQueue(Consts.CREATE_NOTIFICATION_NCMP, request);
-//        log.debug("Send message to kafka \"{}\": {}", Consts.CREATE_REPORT_WITH_COMPANY_ID, LogFormatter.toStringWrapper(request));
-//        kafkaSender.sendRequestToQueue(Consts.CREATE_REPORT_WITH_COMPANY_ID, request);
-//        // см. в report-service: ROOT_ACTV_NotificationBuilder, ru.spcex.clearing.reports.services.ReportService
     }
 
     public String generateInfoAccount(Long id) {

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/account/sdf01/AccountSdf01Request.java

@@ -4,6 +4,9 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 
 import java.util.List;
 
+/**
+ * SDF01 / SDF08
+ */
 public class AccountSdf01Request {
 
     private Long groupingSdf01Id;