Minor refactoring
This commit is contained in:
Mikhail Trofimov
parent
59b074b2da
commit
77f217e0f8
2 changed files with 117 additions and 78 deletions
|
|
@ -198,10 +198,6 @@ public class KafkaMonitoringService {
|
|||
}
|
||||
|
||||
for (PhonesMap phonesMap : phonesMapList) {
|
||||
// флаги просрочек
|
||||
Integer flagDebt5000 = phonesMap.getFlagDebt5000();
|
||||
Integer flag90days = phonesMap.getFlag90days();
|
||||
|
||||
// в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
|
||||
List<UserPackagePhoneData> userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone());
|
||||
|
||||
|
|
@ -210,44 +206,30 @@ public class KafkaMonitoringService {
|
|||
}
|
||||
|
||||
for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) {
|
||||
String subjectId = userPackagePhoneData.getSubjectId();
|
||||
String signalTypes = userPackagePhoneData.getSignalTypes();
|
||||
|
||||
if (signalTypes.contains(Integer.toString(signalType))) {
|
||||
SignalKafkaEvent event = new SignalKafkaEvent();
|
||||
event.setEventType(EventType.WRITE);
|
||||
event.setNatural(true);
|
||||
event.setUserPackagePhoneData(userPackagePhoneData);
|
||||
event.setPackageMessage(message);
|
||||
event.setSubjectId(subjectId);
|
||||
event.setFlagDebt5000(flagDebt5000);
|
||||
event.setFlag90days(flag90days);
|
||||
event.setSignalType(Integer.toString(signalType));
|
||||
event.setBusCategory(signalType == 3 ? busCategory : null);
|
||||
|
||||
Integer segment = segmentMap.get(memberCode.substring(0, 6));
|
||||
event.setSegment(segment != null ? segment : -1);
|
||||
SignalKafkaEvent event = buildSignalKafkaEvent(
|
||||
true, message,
|
||||
phonesMap, userPackagePhoneData,
|
||||
null, null, // legal person fields
|
||||
signalType, busCategory, segment
|
||||
);
|
||||
|
||||
String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
|
||||
+ "_" + userPackagePhoneData.getPackageName();
|
||||
String reportFileName = resolveNameAndEnqueue(reportKey, event);
|
||||
String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
|
||||
if (reportFileName == null) {
|
||||
continue; // событие не встало в очередь — hit не пишем
|
||||
}
|
||||
Hit hit = Hit.builder()
|
||||
.packageId(userPackagePhoneData.getPackageId())
|
||||
.subjectId(subjectId)
|
||||
.signalType(signalType)
|
||||
.reportFileName(reportFileName.replace(".xml", ".zip"))
|
||||
.inqT(message.getInquiryType())
|
||||
.busCategory(busCategory)
|
||||
.inqDt(message.getInquiryDate())
|
||||
.inqP(message.getInquiryPurpose())
|
||||
.inqA(message.getInquiryAmount())
|
||||
.segment(segment != null ? segment : -1)
|
||||
.recordedAt(LocalDateTime.now())
|
||||
.phone(userPackagePhoneData.getPhone())
|
||||
.build();
|
||||
Hit hit = buildHit(
|
||||
true, message,
|
||||
userPackagePhoneData,
|
||||
null,
|
||||
signalType, busCategory,
|
||||
segment, reportFileName
|
||||
);
|
||||
hits.add(hit);
|
||||
}
|
||||
}
|
||||
|
|
@ -303,10 +285,6 @@ public class KafkaMonitoringService {
|
|||
}
|
||||
|
||||
for (LegalsMap legalsMap : legalsMapList) {
|
||||
// флаги просрочек
|
||||
Integer flag90days = legalsMap.getFlag90days();
|
||||
LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate();
|
||||
|
||||
// в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются
|
||||
List<UserPackageLegalData> userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum()));
|
||||
|
||||
|
|
@ -315,45 +293,32 @@ public class KafkaMonitoringService {
|
|||
}
|
||||
|
||||
for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) {
|
||||
String subjectId = userPackageLegalData.getSubjectId();
|
||||
String signalTypes = userPackageLegalData.getSignalTypes();
|
||||
|
||||
if (signalTypes.contains(Integer.toString(signalType))) {
|
||||
SignalKafkaEvent event = new SignalKafkaEvent();
|
||||
event.setEventType(EventType.WRITE);
|
||||
event.setNatural(false);
|
||||
event.setUserPackageLegalData(userPackageLegalData);
|
||||
event.setPackageMessage(message);
|
||||
event.setSubjectId(subjectId);
|
||||
event.setFlag90days(flag90days);
|
||||
event.setSignalType(Integer.toString(signalType));
|
||||
event.setBusCategory(signalType == 3 ? busCategory : null);
|
||||
event.setLastRejectedDate(lastRejectedDate);
|
||||
|
||||
Integer segment = segmentMap.get(memberCode.substring(0, 6));
|
||||
event.setSegment(segment != null ? segment : -1);
|
||||
|
||||
SignalKafkaEvent event = buildSignalKafkaEvent(
|
||||
false, message,
|
||||
null, null, // np fields
|
||||
legalsMap,
|
||||
userPackageLegalData,
|
||||
signalType, busCategory, segment
|
||||
);
|
||||
|
||||
String reportKey = userPackageLegalData.getUserId().substring(0, 6)
|
||||
+ "_" + userPackageLegalData.getPackageName();
|
||||
String reportFileName = resolveNameAndEnqueue(reportKey, event);
|
||||
String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
|
||||
if (reportFileName == null) {
|
||||
continue; // событие не встало в очередь — hit не пишем
|
||||
}
|
||||
Hit hit = Hit.builder()
|
||||
.packageId(userPackageLegalData.getPackageId())
|
||||
.subjectId(subjectId)
|
||||
.signalType(signalType)
|
||||
.reportFileName(reportFileName.replace(".xml", ".zip"))
|
||||
.inqT(message.getInquiryType())
|
||||
.busCategory(busCategory)
|
||||
.inqDt(message.getInquiryDate())
|
||||
.inqP(message.getInquiryPurpose())
|
||||
.inqA(message.getInquiryAmount())
|
||||
.segment(segment != null ? segment : -1)
|
||||
.recordedAt(LocalDateTime.now())
|
||||
.inn(userPackageLegalData.getInn())
|
||||
.regNum(userPackageLegalData.getRegNum())
|
||||
.build();
|
||||
Hit hit = buildHit(
|
||||
false, message,
|
||||
null,
|
||||
userPackageLegalData,
|
||||
signalType, busCategory,
|
||||
segment, reportFileName
|
||||
);
|
||||
hits.add(hit);
|
||||
}
|
||||
}
|
||||
|
|
@ -370,18 +335,86 @@ public class KafkaMonitoringService {
|
|||
(inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Резолвит имя результирующего файла и ставит событие в очередь.
|
||||
*
|
||||
* @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
|
||||
*/
|
||||
private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
|
||||
try {
|
||||
return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
|
||||
} catch (InterruptedException e) {
|
||||
log.error("Error while putting signal event", e);
|
||||
Thread.currentThread().interrupt();
|
||||
return null;
|
||||
private SignalKafkaEvent buildSignalKafkaEvent(boolean isNatural, PackageMessage message,
|
||||
PhonesMap phonesMap, UserPackagePhoneData userPackagePhoneData,
|
||||
LegalsMap legalsMap, UserPackageLegalData userPackageLegalData,
|
||||
int signalType, String busCategory, Integer segment) {
|
||||
if (isNatural) {
|
||||
// флаги просрочек
|
||||
Integer flagDebt5000 = phonesMap.getFlagDebt5000();
|
||||
Integer flag90days = phonesMap.getFlag90days();
|
||||
|
||||
String subjectId = userPackagePhoneData.getSubjectId();
|
||||
|
||||
return SignalKafkaEvent.builder()
|
||||
.eventType(EventType.WRITE)
|
||||
.natural(true)
|
||||
.userPackagePhoneData(userPackagePhoneData)
|
||||
.packageMessage(message)
|
||||
.subjectId(subjectId)
|
||||
.flagDebt5000(flagDebt5000)
|
||||
.flag90days(flag90days)
|
||||
.signalType(Integer.toString(signalType))
|
||||
.busCategory(signalType == 3 ? busCategory : null)
|
||||
.segment(segment != null ? segment : -1)
|
||||
.build();
|
||||
} else {
|
||||
// флаги просрочек
|
||||
Integer flag90days = legalsMap.getFlag90days();
|
||||
LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate();
|
||||
|
||||
String subjectId = userPackageLegalData.getSubjectId();
|
||||
return SignalKafkaEvent.builder()
|
||||
.eventType(EventType.WRITE)
|
||||
.natural(false)
|
||||
.userPackageLegalData(userPackageLegalData)
|
||||
.packageMessage(message)
|
||||
.subjectId(subjectId)
|
||||
.flag90days(flag90days)
|
||||
.signalType(Integer.toString(signalType))
|
||||
.busCategory(signalType == 3 ? busCategory : null)
|
||||
.lastRejectedDate(lastRejectedDate)
|
||||
.segment(segment != null ? segment : -1)
|
||||
.build();
|
||||
}
|
||||
}
|
||||
|
||||
private Hit buildHit(boolean isNatural, PackageMessage message,
|
||||
UserPackagePhoneData userPackagePhoneData,
|
||||
UserPackageLegalData userPackageLegalData,
|
||||
int signalType, String busCategory,
|
||||
Integer segment, String reportFileName) {
|
||||
if (isNatural) {
|
||||
return Hit.builder()
|
||||
.packageId(userPackagePhoneData.getPackageId())
|
||||
.subjectId(userPackagePhoneData.getSubjectId())
|
||||
.signalType(signalType)
|
||||
.reportFileName(reportFileName.replace(".xml", ".zip"))
|
||||
.inqT(message.getInquiryType())
|
||||
.busCategory(busCategory)
|
||||
.inqDt(message.getInquiryDate())
|
||||
.inqP(message.getInquiryPurpose())
|
||||
.inqA(message.getInquiryAmount())
|
||||
.segment(segment != null ? segment : -1)
|
||||
.recordedAt(LocalDateTime.now())
|
||||
.phone(userPackagePhoneData.getPhone())
|
||||
.build();
|
||||
} else {
|
||||
return Hit.builder()
|
||||
.packageId(userPackageLegalData.getPackageId())
|
||||
.subjectId(userPackageLegalData.getSubjectId())
|
||||
.signalType(signalType)
|
||||
.reportFileName(reportFileName.replace(".xml", ".zip"))
|
||||
.inqT(message.getInquiryType())
|
||||
.busCategory(busCategory)
|
||||
.inqDt(message.getInquiryDate())
|
||||
.inqP(message.getInquiryPurpose())
|
||||
.inqA(message.getInquiryAmount())
|
||||
.segment(segment != null ? segment : -1)
|
||||
.recordedAt(LocalDateTime.now())
|
||||
.inn(userPackageLegalData.getInn())
|
||||
.regNum(userPackageLegalData.getRegNum())
|
||||
.build();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
package ru.nbch.credit_tracker.service.mode.monitoring;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.nbch.credit_tracker.enums.EventType;
|
||||
|
|
@ -21,6 +22,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
|
|||
* закрывающий эпоху N. Поэтому WRITE-события кладутся в очередь только через
|
||||
* {@link #resolveNameAndEnqueue}, а ROTATE — только через {@link #enqueueRotate}.
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class ReportNameRegistry {
|
||||
|
||||
|
|
@ -40,7 +42,7 @@ public class ReportNameRegistry {
|
|||
* @param key membercode_packageName
|
||||
* @return имя xml-файла, в который попадёт сигнал
|
||||
*/
|
||||
public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) throws InterruptedException {
|
||||
public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) {
|
||||
lock.readLock().lock();
|
||||
try {
|
||||
String name = names.computeIfAbsent(key, k ->
|
||||
|
|
@ -48,6 +50,10 @@ public class ReportNameRegistry {
|
|||
LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter)));
|
||||
signalEventsQueue.put(event);
|
||||
return name;
|
||||
} catch (InterruptedException e) {
|
||||
log.error("Error while putting signal event", e);
|
||||
Thread.currentThread().interrupt();
|
||||
return null;
|
||||
} finally {
|
||||
lock.readLock().unlock();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue