Added writing logic

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;
+    private UserPackagePhoneData userPackagePhoneData;
-    // other data for signal
 }

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;
 }

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);
                     }
                 }

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<SignalKafkaEvent> signalEventsQueue;
+    private final UserPackageMapper userPackageMapper;
     private final CreditTrackerSettings config;
 
     public SignalKafkaEventWriterService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> 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/создать новый и записать
+        UserPackagePhoneData packagePhoneData = event.getUserPackagePhoneData();
-        // TODO добавить пакет в сет
+        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<UserPackage> 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;
         }
     }
 }