Added hits inserting
This commit is contained in:
parent
1813647170
commit
4bd58624ed
4 changed files with 90 additions and 11 deletions
|
|
@ -1,12 +1,14 @@
|
|||
package ru.nbch.credit_tracker.entities.app;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import lombok.Builder;
|
||||
import lombok.Getter;
|
||||
import lombok.Setter;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
//@Builder
|
||||
@Builder
|
||||
public class Hit {
|
||||
private Long id;
|
||||
private Long phone;
|
||||
|
|
@ -15,4 +17,13 @@ public class Hit {
|
|||
private Integer signalType;
|
||||
private LocalDateTime responseSentAt;
|
||||
private String reportFileName;
|
||||
private Long inn;
|
||||
private Long regNum;
|
||||
private String busCategory;
|
||||
private LocalDateTime inqDt;
|
||||
private String inqP;
|
||||
private Integer inqT;
|
||||
private Long inqA;
|
||||
private Integer segment;
|
||||
private LocalDateTime recordedAt;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import ru.nbch.credit_tracker.model.SignalKafkaEvent;
|
|||
import ru.nbch.credit_tracker.model.UserPackageLegalData;
|
||||
import ru.nbch.credit_tracker.model.UserPackagePhoneData;
|
||||
import ru.nbch.credit_tracker.service.nbch.SourceService;
|
||||
import ru.nbch.credit_tracker.service.signal.HitsService;
|
||||
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
|
||||
import ru.nbch.credit_tracker.service.signal.SegmentMapService;
|
||||
import ru.nbch.credit_tracker.utils.CTUtil;
|
||||
|
|
@ -45,17 +46,20 @@ public class KafkaMonitoringService {
|
|||
private final PackageRecordsService packageRecordsService;
|
||||
private final SourceService sourceService;
|
||||
private final SegmentMapService segmentMapService;
|
||||
private final HitsService hitsService;
|
||||
|
||||
public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
|
||||
PackageRecordsService packageRecordsService,
|
||||
SourceService sourceService,
|
||||
SegmentMapService segmentMapService,
|
||||
CreditTrackerSettings settings) {
|
||||
CreditTrackerSettings settings,
|
||||
HitsService hitsService) {
|
||||
this.signalEventsQueue = signalEventsQueue;
|
||||
this.packageRecordsService = packageRecordsService;
|
||||
this.sourceService = sourceService;
|
||||
this.segmentMapService = segmentMapService;
|
||||
this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера
|
||||
this.hitsService = hitsService;
|
||||
this.processingPermits = new Semaphore(permits);
|
||||
}
|
||||
|
||||
|
|
@ -106,7 +110,8 @@ public class KafkaMonitoringService {
|
|||
List<Hit> lpHits = processLpMessages(lpMessages, segmentMap);
|
||||
hits.addAll(lpHits);
|
||||
}
|
||||
// TODO insert hits
|
||||
|
||||
hitsService.insertHits(hits);
|
||||
}
|
||||
|
||||
private List<Hit> processNpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
|
||||
|
|
@ -186,6 +191,21 @@ public class KafkaMonitoringService {
|
|||
event.setSegment(segment != null ? segment : -1);
|
||||
|
||||
putToQueue(event);
|
||||
Hit hit = Hit.builder()
|
||||
.packageId(userPackagePhoneData.getPackageId())
|
||||
.subjectId(subjectId)
|
||||
.signalType(signalType)
|
||||
.reportFileName("") // TODO
|
||||
.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();
|
||||
hits.add(hit);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -251,15 +271,15 @@ public class KafkaMonitoringService {
|
|||
continue;
|
||||
}
|
||||
|
||||
for (UserPackageLegalData userPackagePhoneData : userPackageLegalDataList) {
|
||||
String subjectId = userPackagePhoneData.getSubjectId();
|
||||
String signalTypes = userPackagePhoneData.getSignalTypes();
|
||||
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(userPackagePhoneData);
|
||||
event.setUserPackageLegalData(userPackageLegalData);
|
||||
event.setPackageMessage(message);
|
||||
event.setSubjectId(subjectId);
|
||||
event.setFlag90days(flag90days);
|
||||
|
|
@ -271,6 +291,22 @@ public class KafkaMonitoringService {
|
|||
event.setSegment(segment != null ? segment : -1);
|
||||
|
||||
putToQueue(event);
|
||||
Hit hit = Hit.builder()
|
||||
.packageId(userPackageLegalData.getPackageId())
|
||||
.subjectId(subjectId)
|
||||
.signalType(signalType)
|
||||
.reportFileName("") // TODO
|
||||
.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();
|
||||
hits.add(hit);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,10 +6,42 @@
|
|||
<mapper namespace="ru.nbch.credit_tracker.mapper.signal.HitsMapper">
|
||||
|
||||
<insert id="insertHits" parameterType="java.util.List">
|
||||
INSERT INTO signals_hits (package_id, subject_id, phone, signal_type, response_sent_at, report_file_name)
|
||||
INSERT INTO signals_hits (
|
||||
package_id,
|
||||
subject_id,
|
||||
phone,
|
||||
signal_type,
|
||||
response_sent_at,
|
||||
report_file_name,
|
||||
inn,
|
||||
reg_num,
|
||||
bus_category,
|
||||
inq_dt,
|
||||
inq_p,
|
||||
inq_t,
|
||||
inq_a,
|
||||
segment,
|
||||
recorded_at
|
||||
)
|
||||
values
|
||||
<foreach collection="hits" item="hit" separator=",">
|
||||
(#{hit.packageId}, #{hit.subjectId}, #{hit.phone}, #{hit.signalType}, #{hit.responseSentAt}, #{hit.reportFileName})
|
||||
(
|
||||
#{hit.packageId},
|
||||
#{hit.subjectId},
|
||||
#{hit.phone},
|
||||
#{hit.signalType},
|
||||
#{hit.responseSentAt},
|
||||
#{hit.reportFileName},
|
||||
#{hit.inn},
|
||||
#{hit.regNum},
|
||||
#{hit.busCategory},
|
||||
#{hit.inqDt},
|
||||
#{hit.inqP},
|
||||
#{hit.inqT},
|
||||
#{hit.inqA},
|
||||
#{hit.segment},
|
||||
#{hit.recordedAt}
|
||||
)
|
||||
</foreach>
|
||||
</insert>
|
||||
|
||||
|
|
|
|||
|
|
@ -70,7 +70,7 @@ alter table signals_hits
|
|||
alter table signals_hits
|
||||
add inq_t smallint;
|
||||
alter table signals_hits
|
||||
add inq_a float;
|
||||
add inq_a bigint;
|
||||
alter table signals_hits
|
||||
add segment smallint;
|
||||
alter table signals_hits
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue