Compare commits
No commits in common. "8ebf8175dad5fa4df81b809efc9a0d4f36dd9712" and "50c346018b401e417074940e36fda641dd26735f" have entirely different histories.
8ebf8175da
...
50c346018b
2 changed files with 0 additions and 136 deletions
|
|
@ -1,132 +0,0 @@
|
|||
package ru.nbch.credit_tracker.service.mode.monitoring;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.nbch.credit_tracker.entities.app.Hit;
|
||||
import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
|
||||
import ru.nbch.credit_tracker.model.IUserPackageData;
|
||||
import ru.nbch.credit_tracker.model.SignalKafkaEvent;
|
||||
import ru.nbch.credit_tracker.service.nbch.SourceService;
|
||||
import ru.nbch.credit_tracker.utils.CTUtil;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class KafkaSignalProcessorSupport {
|
||||
|
||||
private final SourceService sourceService;
|
||||
private final SignalKafkaEventPublisher eventPublisher;
|
||||
|
||||
public KafkaSignalProcessorSupport(SourceService sourceService,
|
||||
SignalKafkaEventPublisher eventPublisher) {
|
||||
this.sourceService = sourceService;
|
||||
this.eventPublisher = eventPublisher;
|
||||
}
|
||||
|
||||
public <M, D extends IUserPackageData, K> List<Hit> process(List<PackageMessage> messages,
|
||||
Map<String, Integer> segmentMap,
|
||||
Function<List<Integer>, List<M>> mappingLoader,
|
||||
Function<M, Integer> fidExtractor,
|
||||
Function<M, K> keyExtractor,
|
||||
Function<List<M>, Map<K, List<D>>> packageDataLoader,
|
||||
SignalEventBuilder<M, D> eventBuilder,
|
||||
SignalHitBuilder<D> hitBuilder) {
|
||||
List<Hit> hits = new ArrayList<>();
|
||||
|
||||
List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
|
||||
List<String> memberCodes = messages.stream()
|
||||
.map(message -> message.getMemberCode().substring(0, 6))
|
||||
.toList();
|
||||
|
||||
List<M> mappings = mappingLoader.apply(fids);
|
||||
Map<Integer, List<M>> mappingsByFid = mappings.stream()
|
||||
.collect(Collectors.groupingBy(fidExtractor));
|
||||
Map<K, List<D>> packageDataByKey = packageDataLoader.apply(mappings);
|
||||
|
||||
// прелоад по memberCode категорий
|
||||
Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
|
||||
|
||||
for (PackageMessage message : messages) {
|
||||
Integer fid = message.getFid();
|
||||
String commonMemberCode = message.getMemberCode().substring(0, 6);
|
||||
|
||||
String busCategory = busCategories.get(commonMemberCode);
|
||||
if (StringUtils.isBlank(busCategory)) {
|
||||
log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", commonMemberCode);
|
||||
continue;
|
||||
}
|
||||
|
||||
int signalType = CTUtil.checkBusCategory(busCategory);
|
||||
|
||||
List<M> messageMappings = mappingsByFid.get(fid);
|
||||
if (messageMappings == null) {
|
||||
continue;
|
||||
}
|
||||
|
||||
for (M mapping : messageMappings) {
|
||||
List<D> packageDataList = packageDataByKey.get(keyExtractor.apply(mapping));
|
||||
if (packageDataList == null) {
|
||||
continue;
|
||||
}
|
||||
|
||||
for (D packageData : packageDataList) {
|
||||
if (!hasSignalType(packageData.getSignalTypes(), signalType)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
Integer segment = segmentMap.get(commonMemberCode);
|
||||
SignalKafkaEvent event = eventBuilder.build(message, mapping, packageData, signalType, busCategory, segment);
|
||||
|
||||
String reportKey = packageData.getUserId().substring(0, 6)
|
||||
+ "_" + packageData.getPackageName();
|
||||
String reportFileName = eventPublisher.resolveNameAndEnqueue(reportKey, event);
|
||||
if (reportFileName == null) {
|
||||
continue; // событие не встало в очередь — hit не пишем
|
||||
}
|
||||
|
||||
hits.add(hitBuilder.build(message, packageData, signalType, busCategory, segment, reportFileName));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return hits;
|
||||
}
|
||||
|
||||
private boolean hasSignalType(String signalTypes, int signalType) {
|
||||
if (StringUtils.isBlank(signalTypes)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
String expectedSignalType = Integer.toString(signalType);
|
||||
return Arrays.stream(signalTypes.split(","))
|
||||
.map(String::trim)
|
||||
.anyMatch(expectedSignalType::equals);
|
||||
}
|
||||
|
||||
@FunctionalInterface
|
||||
public interface SignalEventBuilder<M, D extends IUserPackageData> {
|
||||
SignalKafkaEvent build(PackageMessage message,
|
||||
M mapping,
|
||||
D packageData,
|
||||
int signalType,
|
||||
String busCategory,
|
||||
Integer segment);
|
||||
}
|
||||
|
||||
@FunctionalInterface
|
||||
public interface SignalHitBuilder<D extends IUserPackageData> {
|
||||
Hit build(PackageMessage message,
|
||||
D packageData,
|
||||
int signalType,
|
||||
String busCategory,
|
||||
Integer segment,
|
||||
String reportFileName);
|
||||
}
|
||||
}
|
||||
|
|
@ -321,12 +321,8 @@ public class SignalKafkaEventWriterService {
|
|||
private List<SignalKafkaEvent> drain() {
|
||||
try {
|
||||
List<SignalKafkaEvent> batch = new ArrayList<>(8000);
|
||||
batch.add(signalEventsQueue.take());
|
||||
signalEventsQueue.drainTo(batch, 8000);
|
||||
return batch;
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return Collections.emptyList();
|
||||
} catch (Exception e) {
|
||||
log.error("Error while drain signal events", e);
|
||||
return Collections.emptyList();
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue