From d2800aa2bfd423a26a8717a447d8a123b7c764cf Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Mon, 22 Jun 2026 15:31:50 +0300 Subject: [PATCH] Added writing logic --- .../model/SignalKafkaEvent.java | 3 +- .../model/UserPackagePhoneData.java | 2 + .../monitoring/KafkaMonitoringService.java | 3 +- .../SignalKafkaEventWriterService.java | 49 +++++++++++++++++-- 4 files changed, 48 insertions(+), 9 deletions(-) diff --git a/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java b/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java index 5f49721..354851b 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java +++ b/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java @@ -11,6 +11,5 @@ import ru.nbch.credit_tracker.enums.EventType; public class SignalKafkaEvent { private EventType eventType; - private Long packageId; - // other data for signal + private UserPackagePhoneData userPackagePhoneData; } diff --git a/src/main/java/ru/nbch/credit_tracker/model/UserPackagePhoneData.java b/src/main/java/ru/nbch/credit_tracker/model/UserPackagePhoneData.java index 3d62951..a9970a3 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/UserPackagePhoneData.java +++ b/src/main/java/ru/nbch/credit_tracker/model/UserPackagePhoneData.java @@ -12,4 +12,6 @@ public class UserPackagePhoneData { private String signalTypes; private Long phone; private String subjectId; + private String userId; + private String packageName; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java index 690980f..d9b9d7a 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java @@ -132,14 +132,13 @@ public class KafkaMonitoringService { } for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) { - Long packageId = userPackagePhoneData.getPackageId(); String subjectId = userPackagePhoneData.getSubjectId(); String signalTypes = userPackagePhoneData.getSignalTypes(); if (signalTypes.contains(signalType.toString())) { SignalKafkaEvent event = new SignalKafkaEvent(); event.setEventType(EventType.WRITE); - event.setPackageId(packageId); + event.setUserPackagePhoneData(userPackagePhoneData); putToQueue(event); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java index 272b9ef..9f75b0e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java @@ -4,9 +4,17 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.enums.EventType; +import ru.nbch.credit_tracker.enums.Status; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.SignalKafkaEvent; +import ru.nbch.credit_tracker.model.UserPackagePhoneData; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; import java.util.concurrent.BlockingQueue; @Slf4j @@ -14,18 +22,23 @@ import java.util.concurrent.BlockingQueue; public class SignalKafkaEventWriterService { private final BlockingQueue signalEventsQueue; + private final UserPackageMapper userPackageMapper; private final CreditTrackerSettings config; public SignalKafkaEventWriterService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, + UserPackageMapper userPackageMapper, CreditTrackerSettings config) { this.signalEventsQueue = signalEventsQueue; + this.userPackageMapper = userPackageMapper; this.config = config; } - // TODO вызвать в main public void startWriting() { while (true) { SignalKafkaEvent event = take(); + if (event == null) { + continue; + } if (event.getEventType() == EventType.STOP) { break; } @@ -44,15 +57,41 @@ public class SignalKafkaEventWriterService { * Записывает сигнал в xml для конкретного пакета */ private void writeSignal(SignalKafkaEvent event) { - // TODO записать сигнал в существующий xml/создать новый и записать - // TODO добавить пакет в сет + UserPackagePhoneData packagePhoneData = event.getUserPackagePhoneData(); + try { + Path folder = Path.of(config.getLoadDir(), + "tmp", packagePhoneData.getUserId(), packagePhoneData.getPackageName()); + Files.createDirectories(folder); + Path targetFilePathWithExt = folder.resolve(packagePhoneData.getUserId() + "_" + packagePhoneData.getPackageName() + ".xml"); + + boolean exists = targetFilePathWithExt.toFile().exists(); + + if (!exists) { + Files.createFile(targetFilePathWithExt); + // TODO начать запись + } + + // TODO записать сигнал + } catch (Exception e) { + log.error("Report was not created for PackageId = {}, PackageName = {}. {}", packagePhoneData.getPackageId(), packagePhoneData.getPackageName(), ExceptionUtils.getStackTrace(e)); + } } /** * Упаковывает все результирующие xml в архив */ private void rotate() { - // TODO пройтись по сету пакетов и финализировать запись. Очистить сет + List userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name()); // TODO + PROCESSING_REPEATED? + for (UserPackage userPackage : userPackages) { + Path targetFilePath = Path.of(config.getLoadDir(), + "tmp", userPackage.getUserId(), userPackage.getPackageName()) + .resolve(userPackage.getUserId() + "_" + userPackage.getPackageName() + ".xml"); + + if (targetFilePath.toFile().exists()) { + // TODO дописать, переименовать с timestamp, зазиповать и выложить в папку + } + } + // TODO update signals_user_packages } private SignalKafkaEvent take() { @@ -61,7 +100,7 @@ public class SignalKafkaEventWriterService { } catch (InterruptedException e) { log.error("Error while taking signal event", e); Thread.currentThread().interrupt(); - throw new RuntimeException(e); + return null; } } }