Decrease memory usage for loading and creating reports

src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java

@@ -18,7 +18,6 @@ import ru.nbch.credit_tracker.entities.response.active_packages.Packages;
 import ru.nbch.credit_tracker.entities.response.success.MonitoringPackageDeleteResponse;
 import ru.nbch.credit_tracker.entities.response.success.MonitoringReportListResponse;
 import ru.nbch.credit_tracker.entities.response.success.MonitoringResponse;
-import ru.nbch.credit_tracker.entities.signal_package.Signals;
 import ru.nbch.credit_tracker.enums.SignalError;
 import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.enums.XmlErrorFormat;
@@ -163,8 +162,7 @@ public class SignalFacade {
                 signalService.registerSubjects(subjects, packageId);
                 subjects.clear();
                 signalService.updateSubjects(packageId, null);
-                Signals signals = signalService.buildSignals(packageId, null);
+                if (signalRestService.update(packageId, null)) {
-                if (signalRestService.update(signals)) {
                     log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName);
                     signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
                 } else {

src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java

@@ -1,6 +1,7 @@
 package ru.nbch.credit_tracker.service;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@@ -82,14 +83,14 @@ public class MonitoringService {
 
             LocalDateTime lastDownloadedReportDate = CTUtil.parseDateFromReportFileName(userPackage.getLastDownloadedReport());
 
-            if (userPackage.getLastDownloadedReport() != null) {
+            if (StringUtils.isNotBlank(userPackage.getLastDownloadedReport())) {
                 log.info("Last downloaded report found for package {}. Last Downloaded Report Name: {}", userPackage.getPackageName(), userPackage.getLastDownloadedReport());
             } else {
                 log.info("Last Downloaded Report Name is empty");
             }
 
             OnlineDownloadReport report = null;
-            if (userPackage.getLastDownloadedReport() == null && newReport.isPresent()) {
+            if (StringUtils.isBlank(userPackage.getLastDownloadedReport()) && newReport.isPresent()) {
                 report = loadReport(newReport.get()); // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным
             } else if (lastDownloadedReportDate != null && newReport.isPresent()) {
                 LocalDateTime newReportDate = CTUtil.parseDateFromReportFileName(newReport.get());

src/main/java/ru/nbch/credit_tracker/service/SignalService.java

@@ -19,9 +19,15 @@ import ru.nbch.credit_tracker.service.indic.PhoneService;
 import ru.nbch.credit_tracker.service.signal.HitsService;
 import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
 import ru.nbch.credit_tracker.service.signal.UserPackageService;
+import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter;
 
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.time.LocalDateTime;
 import java.util.*;
+import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 @Slf4j
@@ -165,7 +171,32 @@ public class SignalService {
         userPackageService.updatePackageStatus(packageId, status);
     }
 
-    public Signals buildSignals(Integer packageId, Predicate<PhoneFidMap> filter) {
+    public String writeSignals(Integer packageId, Predicate<PhoneFidMap> filter) {
+        String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator + packageId + File.separator;
+        String filePath = folderPath + UUID.randomUUID() + "_output.xml";
+
+        File folder = new File(folderPath);
+        if (!folder.exists()) {
+            try {
+                Files.createDirectories(Path.of(folderPath));
+            } catch (IOException e) {
+                log.error("Could not create folder: {}", folderPath, e);
+                throw new RuntimeException(e);
+            }
+        }
+
+        Signals signals = buildSignals(packageId);
+
+        SignalsStaxXmlWriter signalsStaxXmlWriter = new SignalsStaxXmlWriter("windows-1251");
+
+        signalsStaxXmlWriter.writeWithStax(filePath,
+                signals,
+                writer -> buildPersons(packageId, filter, writer));
+
+        return filePath;
+    }
+
+    private Signals buildSignals(Integer packageId) {
         UserPackage userPackage = userPackageService.findPackage(packageId);
         Signals signals = new Signals();
         signals.setOnline("2"); // Всегда=2 (только онлайн-сигналы)
@@ -174,13 +205,11 @@ public class SignalService {
         auth.setUser(config.getSignalsuser());
         signals.setAuth(auth);
 
-        signals.getConditions().getPersons().getPerson().addAll(buildPersons(packageId, null));
 
         return signals;
     }
 
-    private List<Person> buildPersons(Integer packageId, Predicate<PhoneFidMap> filter) {
+    private void buildPersons(Integer packageId, Predicate<PhoneFidMap> filter, Consumer<List<Person>> writer) {
-        List<Person> result = new ArrayList<>();
         int offset = 0;
         List<PhoneFidMap> batch;
         do {
@@ -192,12 +221,11 @@ public class SignalService {
                 persons = findPersons(batch);
             }
             if (!persons.isEmpty()) {
-                result.addAll(persons);
+                writer.accept(persons);
             }
             offset += batchSize;
         } while (!batch.isEmpty());
 
-        return result;
     }
 
     private List<Person> findPersons(List<PhoneFidMap> subjects) {

src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java

@@ -6,7 +6,6 @@ import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.dto.UserPackage;
-import ru.nbch.credit_tracker.entities.signal_package.Signals;
 import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
 
@@ -57,8 +56,7 @@ public class UpdatingFlagsService {
             Predicate<PhoneFidMap> filter = subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt());
             boolean updated = signalService.updateSubjects(packageId, filter);
             if (updated) {
-                Signals signals = signalService.buildSignals(packageId, filter);
+                if (signalRestService.update(packageId, filter)) {
-                if (signalRestService.update(signals)) {
                     log.info("Flags were updated successfully. PackageId: {} PackageName: {}. Package status changed to ON_MONITORING", packageId, packageName);
                     signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
                 } else {

src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java

@@ -3,7 +3,7 @@ package ru.nbch.credit_tracker.service.signal;
 import jakarta.xml.bind.DatatypeConverter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.FileUrlResource;
 import org.springframework.core.io.Resource;
 import org.springframework.http.*;
 import org.springframework.stereotype.Service;
@@ -11,17 +11,22 @@ import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.client.RestTemplate;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.response.external_errors.SgnlReport;
 import ru.nbch.credit_tracker.entities.response.online_report.SgnlOnlineReport;
-import ru.nbch.credit_tracker.entities.signal_package.Signals;
+import ru.nbch.credit_tracker.service.SignalService;
 import ru.nbch.credit_tracker.service.xml.jaxb.IXmlProcessor;
-import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 
-import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.security.MessageDigest;
 import java.util.Base64;
+import java.util.function.Predicate;
 import java.util.zip.GZIPOutputStream;
 
 @Slf4j
@@ -36,20 +41,24 @@ public class SignalRestService {
     private final RestTemplate restTemplate;
     private final IXmlProcessor xmlProcessor;
     private final StaxXmlProcessor staxProcessor;
+    private final SignalService signalService;
     private final String authToken;
 
     public SignalRestService(@Qualifier(value = "restTemplate") RestTemplate restTemplate,
                              IXmlProcessor xmlProcessor,
                              StaxXmlProcessor staxProcessor,
+                             SignalService signalService,
                              CreditTrackerSettings config) throws Exception {
         this.restTemplate = restTemplate;
         this.xmlProcessor = xmlProcessor;
         this.staxProcessor = staxProcessor;
+        this.signalService = signalService;
         this.authToken = buildToken(config.getSignalsToken(), config.getSignalsuser());
     }
 
-    public boolean update(Signals signals) throws Exception {
+    public boolean update(Integer packageId, Predicate<PhoneFidMap> filter) throws Exception {
-        Resource request = compressFile(signals);
+        String path = signalService.writeSignals(packageId, filter);
+        Resource request = compressFile(path);
 
         MultiValueMap<String, Resource> body = new LinkedMultiValueMap<>();
         body.add("file", request);
@@ -64,6 +73,12 @@ public class SignalRestService {
 
         if (response.getStatusCode().value() == 302) {
             // good response
+            try {
+                Files.delete(Path.of(path));
+                Files.delete(Path.of(path + ".gz"));
+            } catch (IOException e) {
+                log.warn("File {} hasn't been deleted", path);
+            }
             return true;
         } else if (response.getStatusCode().value() == 200 && response.getBody() != null) {
             SgnlReport error = staxProcessor.read(response.getBody(), SgnlReport.class);
@@ -72,6 +87,7 @@ public class SignalRestService {
             } else {
                 log.error("неизвестная ошибка при постановке пакета на мониторинг в официальный сервис сигналов: {}", response.getBody());
             }
+            log.info("Files will not be deleted: {}, {}", path, path + ".gz");
             return false;
         }
         return false;
@@ -171,20 +187,20 @@ public class SignalRestService {
         }
     }
 
-    private Resource compressFile(Signals signals) throws Exception {
+    public Resource compressFile(String path) throws Exception {
-        SignalsStaxXmlWriter signalsStaxXmlWriter = new SignalsStaxXmlWriter("windows-1251");
+        try (InputStream in = Files.newInputStream(Path.of(path));
+             OutputStream out = Files.newOutputStream(Path.of(path + ".gz"));
+             GZIPOutputStream gzipOut = new GZIPOutputStream(out)) {
 
-        ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
+            byte[] buffer = new byte[65536];
-        try (GZIPOutputStream gzipStream = new GZIPOutputStream(byteStream)) {
+            int bytesRead;
-            signalsStaxXmlWriter.writeWithStax(gzipStream, signals, false);
+            while ((bytesRead = in.read(buffer)) > 0) {
+                gzipOut.write(buffer, 0, bytesRead);
             }
+            gzipOut.finish();
 
-        return new ByteArrayResource(byteStream.toByteArray()) {
+            return new FileUrlResource(path + ".gz");
-            @Override
-            public String getFilename() {
-                return "arhive.gz";
         }
-        };
     }
 
     private boolean isXmlError(byte[] data) {

src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java

@@ -7,7 +7,10 @@ import javax.xml.stream.XMLEventFactory;
 import javax.xml.stream.XMLEventWriter;
 import javax.xml.stream.XMLOutputFactory;
 import java.io.BufferedOutputStream;
+import java.io.FileOutputStream;
 import java.io.OutputStream;
+import java.util.List;
+import java.util.function.Consumer;
 
 @Slf4j
 public class SignalsStaxXmlWriter extends StaxXmlWriter<Signals> {
@@ -16,6 +19,51 @@ public class SignalsStaxXmlWriter extends StaxXmlWriter<Signals>{
         super(charsetName);
     }
 
+    public void writeWithStax(String filePath, Signals signals, LoaderWriter loader) {
+        try (FileOutputStream out = new FileOutputStream(filePath)) {
+            XMLOutputFactory output = XMLOutputFactory.newInstance();
+            XMLEventFactory xmlEventFactory = XMLEventFactory.newInstance();
+            XMLEventWriter writer = output.createXMLEventWriter(out, charsetName);
+            writer = new IndentingXMLEventWriter(writer);
+
+            writerWrapper = new WriterWrapper(xmlEventFactory, writer);
+            writerWrapper.addStartDocument(charsetName);
+
+            writerWrapper.addStartElement("Signals");
+            writerWrapper.addAttribute("online", signals.getOnline());
+            writerWrapper.addSimplelement("Id", signals.getId());
+            writeAuth(signals.getAuth());
+
+            writerWrapper.addStartElement("Conditions");
+            writerWrapper.addStartElement("Persons");
+
+
+            loader.accept(persons -> {
+                try {
+                    writePersons(persons);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            });
+
+            writerWrapper.addEndElement("Persons");
+            writerWrapper.addEndElement("Conditions");
+
+            writerWrapper.addEndElement("Signals");
+
+            writerWrapper.addEndDocument();
+            writerWrapper.flush();
+            System.gc();
+        } catch (Throwable e) {
+            log.error("Error serializing object to XML string: {}", e.getMessage(), e);
+            throw new IllegalStateException("cannot serialize object to XML");
+        }
+    }
+
+    public void writePersons(List<Person> persons) throws Exception{
+        writerWrapper.writeCollection("p", persons, this::writePerson);
+    }
+
     public void writeWithStax(OutputStream outputStream, Signals signals, boolean useStandalone) {
         try (BufferedOutputStream bos = new BufferedOutputStream(outputStream, 4 * 1024 * 1024)) {
 
@@ -110,4 +158,7 @@ public class SignalsStaxXmlWriter extends StaxXmlWriter<Signals>{
     }
 
 
+    public interface LoaderWriter {
+        void accept(Consumer<List<Person>> writer);
+    }
 }