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 9a2aff484..5eb1f20b0 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 @@ -368,7 +368,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin log.debug("Selected to update {} account's", toUpdate.size()); // 4. to notification - List notificationIds = new ArrayList<>(); + List notificationAccountIds = new ArrayList<>(); boolean needWait = false; { for (Pair item : toUpdate) { @@ -383,17 +383,19 @@ public class ClearingAccountService extends QueueConsumer implements Initializin continue; } if (AccountStatus.BLOCKED == newStatus || AccountStatus.CLOSE == newStatus) { // статус 0/2 - notificationIds.add(sendNotificationRequest(ObjectType.account_block, account, newStatus)); + sendNotificationRequest(ObjectType.account_block, account, newStatus); + notificationAccountIds.add(account.getId()); needWait = true; } if (AccountStatus.ACTIVE == newStatus) { // статус 1/3 - notificationIds.add(sendNotificationRequest(ObjectType.account_active, account, newStatus)); + sendNotificationRequest(ObjectType.account_active, account, newStatus); + notificationAccountIds.add(account.getId()); needWait = true; } } } if (needWait) { - SDFProcessService.SDF52WaitingData data = new SDFProcessService.SDF52WaitingData(systemRequest, toProcessSDF53, toUpdate, notificationIds); + 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 { @@ -532,24 +534,12 @@ public class ClearingAccountService extends QueueConsumer implements Initializin kafkaSender.sendRequestToQueue(destination, request); } - public void sendTCRUpdateRequest(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 \"{}\": {}", destination, LogFormatter.toStringWrapper(request)); - kafkaSender.sendRequestToQueue(destination, request); - } - public Long sendNotificationRequest(ObjectType objectType, Account account, AccountStatus requestToStatus) { String statusText = AccountStatus.BLOCKED == requestToStatus ? "Заблокирован" : AccountStatus.CLOSE == requestToStatus ? "Закрыт" : AccountStatus.ACTIVE == requestToStatus ? "Разблокирован" : requestToStatus.getKey(); - String message = String.format("Для счета «%s» будет изменен статус на «%s»", account.getAccount(), statusText); + String message = String.format("Для счета %s будет изменен статус на «%s»", account.getAccount(), statusText); final String destination = Consts.NOTIFICATION_NEW; NotificationNewRequest request = new NotificationNewRequest(); request.setObjectType(objectType.getKey()); 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 3321cd9ac..cb4dd9d00 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 @@ -6,6 +6,7 @@ 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.misc.Notification; import ru.clearing.classes.statics.data.sdf.SDf52; import ru.clearing.classes.statics.data.sdf.SDf53; import ru.spcex.clearing.imdg.IMDGDistributedNames; @@ -17,6 +18,7 @@ import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationFee import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.platform.enumeration.AccountStatus; import ru.spcex.platform.enumeration.NotificationStatus; +import ru.spcex.platform.enumeration.ObjectType; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgProvider; @@ -45,6 +47,7 @@ public class SDFProcessService { protected final ImdgId idGenerator; private final Imdg sdf52Imdg; + private final Imdg notificationImdg; public SDFProcessService( Producer kafkaResponseQueue, @@ -56,6 +59,7 @@ 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) { @@ -155,15 +159,15 @@ public class SDFProcessService { public BaseRequest systemRequest; public List> toProcessSDF53; public List> toUpdate; - public HashSet notAnsweredNotifications = new HashSet<>(); + public HashSet notAnsweredAccounts = new HashSet<>(); public String globalStatus; public SDF52WaitingData(BaseRequest systemRequest, List> toProcessSDF53, List> toUpdate, - Collection notAnsweredNotifications) { + Collection notAnsweredAccounts) { this.systemRequest = systemRequest; this.toProcessSDF53 = toProcessSDF53; this.toUpdate = toUpdate; - this.notAnsweredNotifications = new HashSet<>(notAnsweredNotifications); + this.notAnsweredAccounts = new HashSet<>(notAnsweredAccounts); } public Long getGroupId() { @@ -181,6 +185,8 @@ public class SDFProcessService { 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; @@ -190,42 +196,58 @@ public class SDFProcessService { * @return triggered SDF52WaitingData or null */ public SDF52WaitingData onNotificationResponse(NotificationFeedbackRequest onNotification) { - final Long nId = onNotification.getNotificationId(); - if (nId == null) + 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.notAnsweredNotifications.contains(nId)) { + if (i.notAnsweredAccounts.contains(accountId)) { inWaitingLst = i; break; } } if (inWaitingLst == null) { - log.debug("SDF52 waiting list not found for notificationId={}", nId); + log.debug("SDF52 waiting list (count {}) not found for accountId={}", waitingList.size(), accountId); return null; } boolean returnTrigger = false; synchronized (inWaitingLst) { - boolean removedId = inWaitingLst.notAnsweredNotifications.remove(nId); + boolean removedId = inWaitingLst.notAnsweredAccounts.remove(accountId); if (!removedId) { // never - log.warn("Can not remove notificationId={} from waiting list", nId); + log.warn("Can not remove accountId={} from waiting list {}", accountId, inWaitingLst); } if (NotificationStatus.ACPT.equalsByKey(onNotification.getNotificationStatus())) { - if (inWaitingLst.notAnsweredNotifications.isEmpty()) { - log.trace("Waiting list is empty, by notificationId={}", nId); + 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 notificationId={}", nId); + 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.notAnsweredNotifications.isEmpty()) { - log.debug("Remove waiting list by notificationId={} (list groupId={})", nId, inWaitingLst.getGroupId()); + if (returnTrigger || inWaitingLst.notAnsweredAccounts.isEmpty()) { + log.debug("Remove waiting list by accountId={} (list groupId={})", accountId, inWaitingLst.getGroupId()); waitingList.remove(inWaitingLst); } }