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-527

Browse source
This commit is contained in:
AKurakin 2023-09-19 13:28:50 +03:00
parent f09b02b693
commit e19cb6876e
2 changed files with 43 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -368,7 +368,7 @@ public class ClearingAccountService extends QueueConsumer implements Initializin
log.debug("Selected to update {} account's", toUpdate.size());
// 4. to notification
List<Long> notificationIds = new ArrayList<>();
List<Long> notificationAccountIds = new ArrayList<>();
boolean needWait = false;
{
for (Pair<SDf52, Account> 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<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 \"{}\": {}", 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());

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

@ -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<SDf52> sdf52Imdg;
private final Imdg<Notification> notificationImdg;
public SDFProcessService(
Producer<String, Object> 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<SDf52> sdfsByGroupId(Long generationId) {
@ -155,15 +159,15 @@ public class SDFProcessService {
public BaseRequest<StatementRequest> systemRequest;
public List<Triple<SDf52, Account, String>> toProcessSDF53;
public List<Pair<SDf52, Account>> toUpdate;
public HashSet<Long> notAnsweredNotifications = new HashSet<>();
public HashSet<Long> notAnsweredAccounts = new HashSet<>();
public String globalStatus;
public SDF52WaitingData(BaseRequest<StatementRequest> systemRequest, List<Triple<SDf52, Account, String>> toProcessSDF53, List<Pair<SDf52, Account>> toUpdate,
Collection<Long> notAnsweredNotifications) {
Collection<Long> 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);
}
}