Optimize memory usage

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

@@ -11,7 +11,6 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.dto.UserPackage;
 import ru.nbch.credit_tracker.entities.user_request.active_packages.MonitoringActivePackagesRequest;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
-import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject;
 import ru.nbch.credit_tracker.entities.user_response.active_packages.MonitoringActivePackagesResponse;
 import ru.nbch.credit_tracker.entities.user_response.active_packages.Package;
 import ru.nbch.credit_tracker.entities.user_response.active_packages.Packages;
@@ -31,11 +30,16 @@ import ru.nbch.credit_tracker.utils.ValidationUtil;
 import ru.nbch.credit_tracker.utils.ZipUtils;
 
 import javax.xml.stream.XMLStreamException;
+import java.io.BufferedInputStream;
 import java.io.File;
 import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipInputStream;
@@ -81,6 +85,7 @@ public class SignalFacade {
         MonitoringRequest monitoringRequest;
         Integer packageId;
         ZipInputStream zipInputStream = null;
+        String filePath;
         try {
             zipInputStream = new ZipInputStream(file.getInputStream());
             Optional<ZipEntry> xmlFileEntry = ZipUtils.extractValidEntry(zipInputStream);
@@ -95,11 +100,11 @@ public class SignalFacade {
             zipInputStream = new ZipInputStream(file.getInputStream());
             zipInputStream.getNextEntry();
 
-            monitoringRequest = staxXmlProcessor.read(zipInputStream, MonitoringRequest.class);
+            filePath = writeTmpFile(zipInputStream);
-        } catch (Exception e) {
+            if (filePath == null) {
-            if (e.getCause() instanceof XMLStreamException) {
+                throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError);
-                throw clientExcp(SignalError.ERROR_011, XmlErrorFormat.MonitoringError);
             }
+        } catch (Exception e) {
             if (e instanceof SignalClientException) {
                 throw (SignalClientException) e;
             }
@@ -118,12 +123,17 @@ public class SignalFacade {
 
         // валидация xml
         try {
-            ValidationUtil.validateMonitoringRequest(monitoringRequest);
+            monitoringRequest = readAndValidateMonitoringRequest(filePath);
-        } catch (SignalClientException e) {
+        } catch (Exception e) {
-            if (monitoringRequest != null) {
+            CTUtil.removeFile(filePath);
-                signalService.rejectPackage(monitoringRequest);
+            if (e.getCause() instanceof XMLStreamException) {
+                throw clientExcp(SignalError.ERROR_011, XmlErrorFormat.MonitoringError);
             }
-            throw e;
+            if (e instanceof SignalClientException) {
+                throw (SignalClientException) e;
+            }
+            log.error(e.getMessage(), e);
+            throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError);
         }
 
         try {
@@ -151,16 +161,58 @@ public class SignalFacade {
         }
 
         // в отдельном потоке запускаем регистрацию субъектов, поиск фидов и расчет флагов просрочек
-        registerFidsAndDebts(packageId, monitoringRequest.getPackageName(), monitoringRequest.getSubjects().getSubject());
+        registerFidsAndDebts(packageId, monitoringRequest.getPackageName(), filePath);
 
         return ResponseEntity.ok().body(new MonitoringSetupResponse());
     }
 
-    private void registerFidsAndDebts(Integer packageId, String packageName, List<Subject> subjects) {
+    private MonitoringRequest readAndValidateMonitoringRequest(String filePath) {
+        MonitoringRequest monitoringRequest;
+        try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(filePath)), 8 * 1024 * 1024)) {
+            monitoringRequest = staxXmlProcessor.read(in, MonitoringRequest.class,
+                    request -> {
+                        try {
+                            ValidationUtil.validateMonitoringRequest(request);
+                        } catch (SignalClientException e) {
+                            if (request != null) {
+                                signalService.rejectPackage(request);
+                            }
+                            request = null;
+                            throw e;
+                        }
+                        request.getSubjects().getSubject().clear();
+                    }, 40000, "/MonitoringRequest/Subjects/Subject");
+        } catch (IOException e) {
+            throw new RuntimeException("Error while parsing and validating monitoring request", e);
+        }
+        return monitoringRequest;
+    }
+
+    private String writeTmpFile(ZipInputStream zipInputStream) {
+        String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator;
+        String filePath = folderPath + UUID.randomUUID() + "monitoring_request.xml";
+        CTUtil.createFolder(folderPath);
+
+        try (OutputStream out = Files.newOutputStream(Path.of(filePath))) {
+
+            byte[] buffer = new byte[8 * 1024 * 1024];
+            int bytesRead;
+            while ((bytesRead = zipInputStream.read(buffer)) > 0) {
+                out.write(buffer, 0, bytesRead);
+            }
+
+            return filePath;
+        } catch (Exception e) {
+            log.error("Ошибка при записи файла запроса постановки на мониторинг на диск", e);
+            return null;
+        }
+    }
+
+    private void registerFidsAndDebts(Integer packageId, String packageName, String filePath) {
         executor.execute(() -> {
             try {
-                signalService.registerSubjects(subjects, packageId);
+                signalService.registerSubjects(filePath, packageId);
-                subjects.clear();
+                CTUtil.removeFile(filePath);
                 signalService.updateSubjects(packageId, null);
                 if (signalRestService.setupMonitoring(packageId, null)) {
                     log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName);
@@ -170,6 +222,7 @@ public class SignalFacade {
                     signalService.updatePackageStatus(packageId, Status.ERROR);
                 }
             } catch (Exception e) {
+                CTUtil.removeFile(filePath);
                 signalService.updatePackageStatus(packageId, Status.ERROR);
                 log.error("Ошибка при поиске фидов или расчете флагов просрочек. Package_id {}", packageId, e);
             }

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

@@ -141,7 +141,7 @@ public class MonitoringService {
         MonitoringSignalResponseStaxXmlWriter staxXmlWriter = new MonitoringSignalResponseStaxXmlWriter(config.getEncoding());
         staxXmlWriter.writeWithStax(responsePath, response,
                 writer -> {
-                    try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(reportPath)), 4 * 1024 * 1024)) {
+                    try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(reportPath)), 8 * 1024 * 1024)) {
                         staxXmlProcessor.read(in, Persons.class,
                                 personsData -> { // как только считали пачку Persons сразу записали ее в результирующий файл
                                     List<Subject> subjectsBatch = processPersons(personsData.getPersons());
@@ -211,13 +211,13 @@ public class MonitoringService {
                 folder.mkdir();
             }
 
-            try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 4 * 1024 * 1024);
+            try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 8 * 1024 * 1024);
                  BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath + fileName + ".zip"));
                  ZipOutputStream zos = new ZipOutputStream(bos)) {
 
                 ZipEntry zipEntry = new ZipEntry(fileName + ".xml");
                 zos.putNextEntry(zipEntry);
-                byte[] buffer = new byte[65536];
+                byte[] buffer = new byte[8 * 1024 * 1024];
                 int bytesRead;
                 while ((bytesRead = in.read(buffer)) > 0) {
                     zos.write(buffer, 0, bytesRead);
@@ -243,7 +243,7 @@ public class MonitoringService {
         log.info("Updating signal hits for {}", userPackage.getPackageName());
         LocalDateTime responseSentAt = LocalDateTime.now();
         // Считываем результирующий файл пачками и обновляем таблицу hits
-        try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 4 * 1024 * 1024)) {
+        try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 8 * 1024 * 1024)) {
             staxXmlProcessor.read(in, MonitoringSignalResponse.class,
                     response -> { // как только считали пачку Subjects сразу обновлям таблицу hits
                         List<Hit> hits = new ArrayList<>();

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

@@ -11,7 +11,6 @@ import ru.nbch.credit_tracker.entities.external_request.signals.Auth;
 import ru.nbch.credit_tracker.entities.external_request.signals.Person;
 import ru.nbch.credit_tracker.entities.external_request.signals.Signals;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
-import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject;
 import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.service.indic.FlagsService;
 import ru.nbch.credit_tracker.service.indic.PersonService;
@@ -20,9 +19,14 @@ 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 ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 import ru.nbch.credit_tracker.utils.CTUtil;
 
+import java.io.BufferedInputStream;
 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;
@@ -39,6 +43,7 @@ public class SignalService {
     private final PhoneService phoneService;
     private final FlagsService flagsService;
     private final PersonService personService;
+    private final StaxXmlProcessor staxXmlProcessor;
 
     private final int batchSize = 10000;
 
@@ -48,7 +53,8 @@ public class SignalService {
                          PhoneFidMapService phoneFidMapService,
                          PhoneService phoneService,
                          FlagsService flagsService,
-                         PersonService personService) {
+                         PersonService personService,
+                         StaxXmlProcessor staxXmlProcessor) {
         this.config = config;
         this.hitsService = hitsService;
         this.userPackageService = userPackageService;
@@ -56,6 +62,7 @@ public class SignalService {
         this.phoneService = phoneService;
         this.flagsService = flagsService;
         this.personService = personService;
+        this.staxXmlProcessor = staxXmlProcessor;
     }
 
     public boolean isPackageNotUnique(MonitoringRequest request) {
@@ -72,12 +79,17 @@ public class SignalService {
         return userPackageService.insertPackage(userPackage);
     }
 
-    public void registerSubjects(List<Subject> subjects, Integer packageId) {
+    public void registerSubjects(String filePath, Integer packageId) {
         log.info("Start registering subjects. PackageId: {}", packageId);
-        for (int i = 0; i < subjects.size(); i += batchSize) {
+        try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(filePath)), 8 * 1024 * 1024)) {
-            int end = Math.min(i + batchSize, subjects.size());
+            staxXmlProcessor.read(in, MonitoringRequest.class,
-            phoneFidMapService.registerSubjects(subjects.subList(i, end), packageId);
+                    request -> {
-            log.trace("Registered subject's batch from: {} to: {}. Total count: {}. PackageId: {}", i, end, subjects.size(), packageId);
+                        phoneFidMapService.registerSubjects(request.getSubjects().getSubject(), packageId);
+                        log.trace("Registered subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
+                        request.getSubjects().getSubject().clear();
+                    }, batchSize, "/MonitoringRequest/Subjects/Subject");
+        } catch (IOException e) {
+            throw new RuntimeException("Error while parsing monitoring request", e);
         }
         log.info("Subjects were registered. PackageId: {}", packageId);
     }

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

@@ -197,7 +197,7 @@ public class SignalRestService {
             try (InputStream in = new GZIPInputStream(new ByteArrayInputStream(response.getBody()));
                  OutputStream out = Files.newOutputStream(Path.of(filePath))) {
 
-                byte[] buffer = new byte[65536];
+                byte[] buffer = new byte[8 * 1024 * 1024];
                 int bytesRead;
                 while ((bytesRead = in.read(buffer)) > 0) {
                     out.write(buffer, 0, bytesRead);
@@ -229,7 +229,7 @@ public class SignalRestService {
              OutputStream out = Files.newOutputStream(Path.of(path + ".gz"));
              GZIPOutputStream gzipOut = new GZIPOutputStream(out)) {
 
-            byte[] buffer = new byte[65536];
+            byte[] buffer = new byte[8 * 1024 * 1024];
             int bytesRead;
             while ((bytesRead = in.read(buffer)) > 0) {
                 gzipOut.write(buffer, 0, bytesRead);

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

@@ -22,7 +22,7 @@ public class MonitoringSignalResponseStaxXmlWriter extends StaxXmlWriter<Monitor
 
     @Override
     public void writeWithStax(String filePath, MonitoringSignalResponse obj, LoaderWriter<Subject> loader) {
-        try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath), 4 * 1024 * 1024)) {
+        try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath), 8 * 1024 * 1024)) {
             XMLOutputFactory output = XMLOutputFactory.newInstance();
             XMLEventFactory xmlEventFactory = XMLEventFactory.newInstance();
             XMLEventWriter writer = output.createXMLEventWriter(bos, charsetName);

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

@@ -20,7 +20,7 @@ public class SignalsStaxXmlWriter extends StaxXmlWriter<Signals, Person> {
     }
 
     public void writeWithStax(String filePath, Signals signals, LoaderWriter<Person> loader) {
-        try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath), 4 * 1024 * 1024)) {
+        try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(filePath), 8 * 1024 * 1024)) {
             XMLOutputFactory output = XMLOutputFactory.newInstance();
             XMLEventFactory xmlEventFactory = XMLEventFactory.newInstance();
             XMLEventWriter writer = output.createXMLEventWriter(bos, charsetName);

src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java

@@ -90,4 +90,15 @@ public class CTUtil {
             }
         }
     }
+
+    public static void removeFile(String filePath) {
+        try {
+            Path path = Path.of(filePath);
+            if (Files.exists(path)) {
+                Files.delete(path);
+            }
+        } catch (IOException e) {
+            log.warn("File {} hasn't been deleted", filePath);
+        }
+    }
 }