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 5eb1f20b0..7b3bc5045 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 @@ -15,6 +15,7 @@ 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.classes.statics.data.misc.Notification; import ru.clearing.classes.statics.data.registry.TradingClearingRegistry; import ru.clearing.classes.statics.data.sdf.SDf52; import ru.clearing.platform.dictionary.ClearingCategoryDictionary; @@ -51,8 +52,10 @@ import ru.spcex.platform.utils.validation.IValidator; import java.time.Instant; import java.util.*; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; + @Service public class ClearingAccountService extends QueueConsumer implements InitializingBean { private final Logger log = LoggerFactory.getLogger(getClass()); @@ -72,6 +75,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin private final Imdg clearingMemberCategoryImdg; private final Imdg clearingCategoryImdg; private final Imdg tradingClearingRegistryImdg; + private final Imdg notificationImdg; @Autowired public ClearingAccountService(Consumer kafkaQueue, @@ -104,6 +108,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin this.clearingMemberCategoryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ClearingMemberCategory, ClearingMemberCategory.class); this.clearingCategoryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ClearingCategoryDictionary, ClearingCategoryDictionary.class); this.tradingClearingRegistryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class); + this.notificationImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Notification, Notification.class); } @Override @@ -369,39 +374,31 @@ public class ClearingAccountService extends QueueConsumer implements Initializin // 4. to notification List notificationAccountIds = new ArrayList<>(); - boolean needWait = false; - { - for (Pair item : toUpdate) { - SDf52 sdf = item.getFirst(); - Account account = item.getSecond(); - AccountStatus newStatus = sdfProcessService.parseSdf52Status(sdf.getStatus()); - if (newStatus == null) { - throw new IllegalArgumentException("Can not parse sdf status " + sdf.getStatus()); - } - if (newStatus.equalsByKey(account.getStatus())) { - // одинаковых обычно не бывает. - continue; - } - if (AccountStatus.BLOCKED == newStatus || AccountStatus.CLOSE == newStatus) { // статус 0/2 - sendNotificationRequest(ObjectType.account_block, account, newStatus); - notificationAccountIds.add(account.getId()); - needWait = true; - } - if (AccountStatus.ACTIVE == newStatus) { // статус 1/3 - sendNotificationRequest(ObjectType.account_active, account, newStatus); - notificationAccountIds.add(account.getId()); - needWait = true; - } + for (Pair item : toUpdate) { + SDf52 sdf = item.getFirst(); + Account account = item.getSecond(); + AccountStatus newStatus = sdfProcessService.parseSdf52Status(sdf.getStatus()); + if (newStatus == null) { // never + throw new IllegalArgumentException("Can not parse sdf status " + sdf.getStatus()); + } + if (newStatus.equalsByKey(account.getStatus())) { + // одинаковых обычно не бывает. + continue; + } + if (AccountStatus.BLOCKED == newStatus || AccountStatus.CLOSE == newStatus) { // статус 0/2 + sendNotificationRequest(ObjectType.account_block, account, newStatus); + notificationAccountIds.add(account.getId()); + putNotificationWaiting(systemRequest, item.getFirst(), item.getSecond()); + } + if (AccountStatus.ACTIVE == newStatus) { // статус 1/3 + sendNotificationRequest(ObjectType.account_active, account, newStatus); + notificationAccountIds.add(account.getId()); + putNotificationWaiting(systemRequest, item.getFirst(), item.getSecond()); } } - if (needWait) { - SDFProcessService.SDF52WaitingData data = new SDFProcessService.SDF52WaitingData(systemRequest, toProcessSDF53, toUpdate, notificationAccountIds); - sdfProcessService.putNotificationWaiting(data); - log.info("Account's need wait accept over notification. Waiting user. GroupId={}", groupId); - } else { - log.info("No one account for wait notification reply. Do immediatly stage 2. groupId={}", groupId); - accountUpdateSdf52_part2(systemRequest, toProcessSDF53, toUpdate, null); - } + sdfProcessService.process(req, toProcessSDF53); + + log.info("{} account's need wait accept over notification. GroupId={}", notificationAccountIds.size(), groupId); log.debug("successfully processed, grouping id={} with {} accounts.", groupId, toUpdate.size()); @@ -410,7 +407,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin protected RequestInfoUpdate accountUpdateSdf52_part2Notification(BaseRequest secondSystemRequest2) { log.debug("accountUpdateSdf52 NotificationFeedbackRequest received, id={}", secondSystemRequest2.getId()); - SDFProcessService.SDF52WaitingData trigger = sdfProcessService.onNotificationResponse(secondSystemRequest2.getRequestPayload()); + SDF52WaitingData trigger = onNotificationResponse(secondSystemRequest2.getRequestPayload()); if (trigger == null) { log.trace("Not triggered, continue waiting"); return null; @@ -418,7 +415,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin log.debug("Triggered, groupId={}, notification's status: {}, last notification user id={}", trigger.getGroupId(), trigger.globalStatus, secondSystemRequest2.getUserId()); if (NotificationStatus.ACPT.equalsByKey(trigger.globalStatus)) { - return accountUpdateSdf52_part2(trigger.systemRequest, trigger.toProcessSDF53, trigger.toUpdate, secondSystemRequest2); + return accountUpdateSdf52_part2(trigger.systemRequest, trigger.sdf, trigger.account, secondSystemRequest2); } else if (NotificationStatus.CNCL.equalsByKey(trigger.globalStatus)) { log.debug("groupId={} rejected by user {}.", trigger.getGroupId(), secondSystemRequest2.getUserId()); return null; @@ -430,8 +427,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin } protected RequestInfoUpdate accountUpdateSdf52_part2(BaseRequest systemRequest1, - List> toProcessSDF53, - List> toUpdate, + SDf52 sdf, Account account, BaseRequest secondSystemRequest2) { // 5. from notification: StatementRequest req = systemRequest1.getRequestPayload(); @@ -440,64 +436,35 @@ public class ClearingAccountService extends QueueConsumer implements Initializin groupId, systemRequest1 == null ? null : systemRequest1.getId(), secondSystemRequest2 == null ? null : secondSystemRequest2.getId()); int countOfUpdated = 0; - List toTCRRequests = new ArrayList<>(); - ImdgTransaction imdgTransaction = imdgProvider.newTransaction(); - boolean txOk = false; - imdgTransaction.beginTransaction(); - try { // 2. обновление данных, в транзакции - Imdg accountImdg = imdgTransaction.getImdg(IMDGDistributedNames.Map_Account, Account.class); - Instant now = Instant.now(); - accountsLoop: - for (Pair item : toUpdate) { - SDf52 sdf = item.getFirst(); - Account account = item.getSecond(); - AccountStatus newStatus = sdfProcessService.parseSdf52Status(sdf.getStatus()); - if (newStatus == null) { - throw new IllegalArgumentException("Can not parse sdf status " + sdf.getStatus()); - } - // За долгое время ожидания пользователя счёт мог быть обновлён, прочитать его ещё раз - account = accountImdg.getSingleObjectByID(account.getId()); - item.setSecond(account); // В toProcessSDF53 есть тоже account, но он на следующих этапах не используется - 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); - log.trace("transaction {}, S_DF52[{}] do update status to {} for account[{}]", - imdgTransaction, sdf.getId(), newStatus.getKey(), account.getId()); - if (AccountStatus.ACTIVE == newStatus) { - TradingClearingRegistryUpdateRequest tcrReq = new TradingClearingRegistryUpdateRequest(); - tcrReq.setId(account.getId()); - tcrReq.setCompanyId(account.getCompanyId()); - // tcrReq.setId(); // select TCR; - выполнить поиск вне этой транзакции, чтобы - tcrReq.setStatus(account.getStatus()); - toTCRRequests.add(tcrReq); - } - countOfUpdated++; - } else { - log.info("Account[{}] \"{}\" do not updated - same status \"{}\"", account.getId(), account.getAccount(), newStatus); - } - } - txOk = true; - } finally { - if (txOk) { - imdgTransaction.commitTransaction(); - } else { - log.debug("failed update, clearing accounts, rollback transaction {}. Request id={}; sdf52 groupId={}", - imdgTransaction, systemRequest1 == null ? null : systemRequest1.getId(), groupId); - imdgTransaction.rollbackTransaction(); - } - } - if (txOk) { - for (TradingClearingRegistryUpdateRequest tcrReq : toTCRRequests) { - TradingClearingRegistry tcr = tradingClearingRegistryImdg.getFirstObjectByFieldValues(Map.of( - "moneyAccountId", tcrReq.getMoneyAccountId(), - "companyId", tcrReq.getCompanyId())); + // 2. обновление данных + Instant now = Instant.now(); + AccountStatus newStatus = sdfProcessService.parseSdf52Status(sdf.getStatus()); + if (newStatus == null) { + throw new IllegalArgumentException("Can not parse sdf status " + sdf.getStatus()); + } + // За долгое время ожидания пользователя счёт мог быть обновлён, прочитать его ещё раз + account = accountImdg.getSingleObjectByID(account.getId()); + if (!newStatus.equalsByKey(account.getStatus())) { + // Обновление счёта + account.setStatus(newStatus.getKey()); + account.setUpdated(now); + accountImdg.update(account); + log.trace("S_DF52[{}] do update status to {} for account[{}]", + sdf.getId(), newStatus.getKey(), account.getId()); + if (AccountStatus.ACTIVE == newStatus) { + // Отправка в ТКР + TradingClearingRegistryUpdateRequest tcrReq = new TradingClearingRegistryUpdateRequest(); + tcrReq.setMoneyAccountId(account.getId()); + tcrReq.setCompanyId(account.getCompanyId()); + tcrReq.setStatus(account.getStatus()); + Map queryTCR = new HashMap<>(); + queryTCR.put("moneyAccountId", tcrReq.getMoneyAccountId()); + if (tcrReq.getCompanyId() != null) + queryTCR.put("companyId", tcrReq.getCompanyId()); + TradingClearingRegistry tcr = tradingClearingRegistryImdg.getFirstObjectByFieldValues(queryTCR); if (tcr == null) { - log.warn("TradingClearingRegistry by moneyAccountId={} and companyId={} not found, do not send update TCR.", - tcrReq.getMoneyAccountId(), tcrReq.getCompanyId()); + log.warn("TradingClearingRegistry by {} not found, do not send update TCR.", queryTCR); } else { tcrReq.setId(tcr.getId()); String destination = Consts.DESTINATION_TRADING_CLEARING_REGISTRY_UPDATE; @@ -505,12 +472,13 @@ public class ClearingAccountService extends QueueConsumer implements Initializin kafkaSender.sendRequestToQueue(destination, tcrReq); } } - - sdfProcessService.process(req, toProcessSDF53); + countOfUpdated++; + } else { + log.info("Account[{}] \"{}\" do not updated - same status \"{}\"", account.getId(), account.getAccount(), newStatus); } - log.debug("successfully processed, grouping id={}. Updated {} of {} accounts.", - groupId, countOfUpdated, toUpdate.size()); + log.debug("successfully processed, grouping id={}. Updated {} accounts.", + groupId, countOfUpdated); return null; } @@ -551,4 +519,89 @@ public class ClearingAccountService extends QueueConsumer implements Initializin log.trace("About account.id={} send notification request id={}", account.getId(), rKey); return rKey; } + + + // --------- notification apply system ----------- + public static class SDF52WaitingData { + public BaseRequest systemRequest; + public SDf52 sdf; + public Account account; + public Long notAnsweredAccountId; + public String globalStatus; + + public SDF52WaitingData(BaseRequest systemRequest, SDf52 sdf, Account account) { + this.systemRequest = systemRequest; + this.sdf = sdf; + this.account = account; + this.notAnsweredAccountId = account.getId(); + } + + public Long getGroupId() { + if (systemRequest != null && systemRequest.getRequestPayload() != null) + return systemRequest.getRequestPayload().getGroupId(); + return null; + } + } + + protected ConcurrentHashMap waitingList = new ConcurrentHashMap<>(); + + public boolean putNotificationWaiting(BaseRequest systemRequest, SDf52 sdf, Account account) { + SDF52WaitingData data = new SDF52WaitingData(systemRequest, sdf, account); + if (waitingList.contains(data.notAnsweredAccountId)) { + log.warn("groupId={} accountId={} already waiting", data.getGroupId(), data.notAnsweredAccountId); + } + data.globalStatus = null; + waitingList.put(data.notAnsweredAccountId, data); + return true; + } + + /** + * @return triggered SDF52WaitingData or null + */ + public SDF52WaitingData onNotificationResponse(NotificationFeedbackRequest onNotification) { + if (onNotification.getNotificationId() == null) { + log.trace("notificationId was empty"); + return null; + } + Notification notification = notificationImdg.getSingleObjectByID(onNotification.getNotificationId()); + if (notification == null) { + log.warn("Notification with id={} not exist", onNotification.getNotificationId()); + return null; + } + if (!ObjectType.account_active.equalsByKey(notification.getObjectType()) && !ObjectType.account_block.equalsByKey(notification.getObjectType())) { + log.warn("Notification[{}].objectType={} not supported", notification.getId(), notification.getObjectType()); + return null; + } + if (notification.getObjectId() == null) { + log.warn("Notification[{}] with empty objectId", notification.getId()); + return null; + } + final Long accountId = notification.getObjectId(); + log.trace("Match accountId={} by notificationId={}", accountId, notification.getId()); + SDF52WaitingData inWaitingLst = waitingList.get(accountId); + if (inWaitingLst == null) { + log.debug("SDF52 waiting list (count {}) not found for accountId={}", waitingList.size(), accountId); + return null; + } + boolean returnTrigger = false; + if (NotificationStatus.ACPT.equalsByKey(onNotification.getNotificationStatus())) { + log.trace("Triggered ACPT, by accountId={}", accountId); + inWaitingLst.globalStatus = onNotification.getNotificationStatus(); + returnTrigger = true; + } else if (NotificationStatus.CNCL.equalsByKey(onNotification.getNotificationStatus())) { + log.trace("Triggered CNCL, by accountId={}", accountId); + inWaitingLst.globalStatus = onNotification.getNotificationStatus(); + returnTrigger = true; + } else { + log.warn("Unknown notification[{}] status={}", onNotification.getNotificationId(), onNotification.getNotificationStatus()); + } + if (returnTrigger) { + log.debug("Remove from waiting list accountId={} (list groupId={})", accountId, inWaitingLst.getGroupId()); + waitingList.remove(inWaitingLst); + } + if (returnTrigger) + return inWaitingLst; + else + return null; + } } 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 index cb4dd9d00..612107078 100644 --- 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 @@ -47,7 +47,6 @@ public class SDFProcessService { protected final ImdgId idGenerator; private final Imdg sdf52Imdg; - private final Imdg notificationImdg; public SDFProcessService( Producer kafkaResponseQueue, @@ -59,7 +58,6 @@ public class SDFProcessService { this.idGenerator = imdgProvider.getImdgIdGenerator(); this.sdf52Imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf52, SDf52.class); - this.notificationImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Notification, Notification.class); } public Collection sdfsByGroupId(Long generationId) { @@ -154,106 +152,4 @@ public class SDFProcessService { groupId, request.getNameOfTable(), msgId, destination); } - // --------- notification apply system ----------- - public static class SDF52WaitingData { - public BaseRequest systemRequest; - public List> toProcessSDF53; - public List> toUpdate; - public HashSet notAnsweredAccounts = new HashSet<>(); - public String globalStatus; - - public SDF52WaitingData(BaseRequest systemRequest, List> toProcessSDF53, List> toUpdate, - Collection notAnsweredAccounts) { - this.systemRequest = systemRequest; - this.toProcessSDF53 = toProcessSDF53; - this.toUpdate = toUpdate; - this.notAnsweredAccounts = new HashSet<>(notAnsweredAccounts); - } - - public Long getGroupId() { - if (systemRequest != null && systemRequest.getRequestPayload() != null) - return systemRequest.getRequestPayload().getGroupId(); - return null; - } - } - - protected CopyOnWriteArrayList waitingList = new CopyOnWriteArrayList<>(); - - public boolean putNotificationWaiting(SDF52WaitingData data) { - for (SDF52WaitingData i : waitingList) { - if (Objects.equals(i.getGroupId(), data.getGroupId())) { - log.warn("groupId={} already waiting", data.getGroupId()); - } - } - if (data.notAnsweredAccounts.isEmpty()) - log.warn("Add to wait notification list with empty notAnsweredAccounts"); - data.globalStatus = null; - waitingList.add(data); - return true; - } - - /** - * @return triggered SDF52WaitingData or null - */ - public SDF52WaitingData onNotificationResponse(NotificationFeedbackRequest onNotification) { - if (onNotification.getNotificationId() == null) { - log.trace("notificationId was empty"); - return null; - } - Notification notification = notificationImdg.getSingleObjectByID(onNotification.getNotificationId()); - if (notification == null) { - log.warn("Notification with id={} not exist", onNotification.getNotificationId()); - return null; - } - if (!ObjectType.account_active.equalsByKey(notification.getObjectType()) && !ObjectType.account_block.equalsByKey(notification.getObjectType())) { - log.warn("Notification[{}].objectType={} not supported", notification.getId(), notification.getObjectType()); - return null; - } - if (notification.getObjectId() == null) { - log.warn("Notification[{}] with empty objectId", notification.getId()); - return null; - } - final Long accountId = notification.getObjectId(); - log.trace("Match accountId={} by notificationId={}", accountId, notification.getId()); - SDF52WaitingData inWaitingLst = null; - for (SDF52WaitingData i : waitingList) - synchronized (i) { - if (i.notAnsweredAccounts.contains(accountId)) { - inWaitingLst = i; - break; - } - } - if (inWaitingLst == null) { - log.debug("SDF52 waiting list (count {}) not found for accountId={}", waitingList.size(), accountId); - return null; - } - boolean returnTrigger = false; - synchronized (inWaitingLst) { - boolean removedId = inWaitingLst.notAnsweredAccounts.remove(accountId); - if (!removedId) { // never - log.warn("Can not remove accountId={} from waiting list {}", accountId, inWaitingLst); - } - if (NotificationStatus.ACPT.equalsByKey(onNotification.getNotificationStatus())) { - if (inWaitingLst.notAnsweredAccounts.isEmpty()) { - log.trace("Waiting list is empty, by accountId={}", accountId); - inWaitingLst.globalStatus = onNotification.getNotificationStatus(); - returnTrigger = true; - } - } else if (NotificationStatus.CNCL.equalsByKey(onNotification.getNotificationStatus())) { - log.trace("Waiting list cancel, by accountId={}", accountId); - inWaitingLst.globalStatus = onNotification.getNotificationStatus(); - returnTrigger = true; - } else { - log.warn("Unknown notification[{}] status={}", onNotification.getNotificationId(), onNotification.getNotificationStatus()); - } - if (returnTrigger || inWaitingLst.notAnsweredAccounts.isEmpty()) { - log.debug("Remove waiting list by accountId={} (list groupId={})", accountId, inWaitingLst.getGroupId()); - waitingList.remove(inWaitingLst); - } - } - if (returnTrigger) - return inWaitingLst; - else - return null; - } }