Исправлено заполнение pcap фильтра при запуске, добавлен вывод состояния очереди

src/main/java/ru/serega6531/packmate/PackmateApplication.java

@@ -1,33 +1,13 @@
 package ru.serega6531.packmate;
 
-import org.pcap4j.core.PcapNativeException;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.context.event.ApplicationReadyEvent;
-import org.springframework.context.event.EventListener;
-import ru.serega6531.packmate.model.enums.CaptureMode;
-import ru.serega6531.packmate.service.PcapService;
 
 @SpringBootApplication
 public class PackmateApplication {
 
-    @Value("${enable-capture}")
-    private boolean enableCapture;
-
-    @Value("${capture-mode}")
-    private CaptureMode captureMode;
-
     public static void main(String[] args) {
         SpringApplication.run(PackmateApplication.class, args);
     }
 
-    @EventListener(ApplicationReadyEvent.class)
-    public void afterStartup(ApplicationReadyEvent event) throws PcapNativeException {
-        if (enableCapture && captureMode == CaptureMode.LIVE) {
-            final PcapService pcapService = event.getApplicationContext().getBean(PcapService.class);
-            pcapService.start();
-        }
-    }
-
 }

src/main/java/ru/serega6531/packmate/pcap/AbstractPcapWorker.java

@@ -257,7 +257,6 @@ public abstract class AbstractPcapWorker implements PcapWorker, PacketListener {
     }
 
     @Override
-    @SneakyThrows
     public void setFilter(String filter) {
         this.filter = filter;
         applyFilter();

src/main/java/ru/serega6531/packmate/pcap/FilePcapWorker.java

@@ -79,4 +79,9 @@ public class FilePcapWorker extends AbstractPcapWorker {
 
         subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.PCAP_STOPPED, null));
     }
+
+    @Override
+    public String getExecutorState() {
+        return "inline";
+    }
 }

src/main/java/ru/serega6531/packmate/pcap/LivePcapWorker.java

@@ -10,7 +10,9 @@ import ru.serega6531.packmate.service.ServicesService;
 import ru.serega6531.packmate.service.StreamService;
 
 import java.net.UnknownHostException;
-import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class LivePcapWorker extends AbstractPcapWorker {
@@ -30,7 +32,7 @@ public class LivePcapWorker extends AbstractPcapWorker {
 
         BasicThreadFactory factory = new BasicThreadFactory.Builder()
                 .namingPattern("pcap-processor").build();
-        processorExecutorService = Executors.newSingleThreadExecutor(factory);
+        processorExecutorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), factory);
     }
 
     public void start() throws PcapNativeException {
@@ -63,4 +65,8 @@ public class LivePcapWorker extends AbstractPcapWorker {
         log.info("Intercept stopped");
     }
 
+    @Override
+    public String getExecutorState() {
+        return processorExecutorService.toString();
+    }
 }

src/main/java/ru/serega6531/packmate/pcap/NoOpPcapWorker.java

@@ -24,4 +24,9 @@ public class NoOpPcapWorker implements PcapWorker {
     @Override
     public void setFilter(String filter) {
     }
+
+    @Override
+    public String getExecutorState() {
+        return "none";
+    }
 }

src/main/java/ru/serega6531/packmate/pcap/PcapWorker.java

@@ -20,4 +20,5 @@ public interface PcapWorker {
 
     void setFilter(String filter);
 
+    String getExecutorState();
 }

src/main/java/ru/serega6531/packmate/service/PatternService.java

@@ -15,6 +15,7 @@ import ru.serega6531.packmate.model.pojo.PatternDto;
 import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
 import ru.serega6531.packmate.repository.PatternRepository;
 
+import javax.annotation.PostConstruct;
 import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
@@ -40,7 +41,10 @@ public class PatternService {
         this.streamService = streamService;
         this.subscriptionService = subscriptionService;
         this.modelMapper = modelMapper;
+    }
 
+    @PostConstruct
+    public void init() {
         repository.findAll().forEach(p -> patterns.put(p.getId(), p));
         log.info("Loaded {} patterns", patterns.size());
     }

src/main/java/ru/serega6531/packmate/service/PcapService.java

@@ -54,4 +54,8 @@ public class PcapService {
         worker.setFilter(filter);
     }
 
+    public String getExecutorState() {
+        return worker.getExecutorState();
+    }
+
 }

src/main/java/ru/serega6531/packmate/service/ServicesService.java

@@ -12,12 +12,10 @@ import ru.serega6531.packmate.model.pojo.ServiceDto;
 import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
 import ru.serega6531.packmate.repository.ServiceRepository;
 
+import javax.annotation.PostConstruct;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
-import java.util.Collection;
+import java.util.*;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
 
 @Service
 @Slf4j
@@ -43,7 +41,10 @@ public class ServicesService {
         this.pcapService = pcapService;
         this.modelMapper = modelMapper;
         this.localIp = InetAddress.getByName(localIpString);
+    }
 
+    @PostConstruct
+    public void init() {
         repository.findAll().forEach(s -> services.put(s.getPort(), s));
         log.info("Loaded {} services", services.size());
     }
@@ -78,7 +79,7 @@ public class ServicesService {
 
         subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.DELETE_SERVICE, port));
 
-        pcapService.updateFilter(findAll());
+        updateFilter();
     }
 
     public CtfService save(CtfService service) {
@@ -89,11 +90,15 @@ public class ServicesService {
 
         subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.SAVE_SERVICE, toDto(saved)));
 
-        pcapService.updateFilter(findAll());
+        updateFilter();
 
         return saved;
     }
 
+    public void updateFilter() {
+        pcapService.updateFilter(findAll());
+    }
+
     public ServiceDto toDto(CtfService service) {
         return modelMapper.map(service, ServiceDto.class);
     }

src/main/java/ru/serega6531/packmate/tasks/ExecutorStateLoggerTask.java

@@ -0,0 +1,23 @@
+package ru.serega6531.packmate.tasks;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import ru.serega6531.packmate.service.PcapService;
+
+@Component
+@Slf4j
+public class ExecutorStateLoggerTask {
+
+    private final PcapService service;
+
+    public ExecutorStateLoggerTask(PcapService service) {
+        this.service = service;
+    }
+
+    @Scheduled(fixedDelayString = "PT1M", initialDelayString = "PT1M")
+    public void cleanup() {
+        log.info("Executor state: {}", service.getExecutorState());
+    }
+
+}

src/main/java/ru/serega6531/packmate/tasks/OldStreamsCleanupTask.java

@@ -1,4 +1,4 @@
-package ru.serega6531.packmate;
+package ru.serega6531.packmate.tasks;
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;

src/main/java/ru/serega6531/packmate/tasks/StartupListener.java

@@ -0,0 +1,40 @@
+package ru.serega6531.packmate.tasks;
+
+import org.pcap4j.core.PcapNativeException;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+import ru.serega6531.packmate.model.enums.CaptureMode;
+import ru.serega6531.packmate.service.PcapService;
+import ru.serega6531.packmate.service.ServicesService;
+
+@Component
+public class StartupListener {
+
+    @Value("${enable-capture}")
+    private boolean enableCapture;
+
+    @Value("${capture-mode}")
+    private CaptureMode captureMode;
+
+    private final PcapService pcapService;
+    private final ServicesService servicesService;
+
+    public StartupListener(PcapService pcapService, ServicesService servicesService) {
+        this.pcapService = pcapService;
+        this.servicesService = servicesService;
+    }
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void afterStartup() throws PcapNativeException {
+        if (enableCapture) {
+            servicesService.updateFilter();
+
+            if (captureMode == CaptureMode.LIVE) {
+                pcapService.start();
+            }
+        }
+    }
+
+}

src/main/java/ru/serega6531/packmate/tasks/TimeoutStreamsSaver.java

@@ -1,4 +1,4 @@
-package ru.serega6531.packmate;
+package ru.serega6531.packmate.tasks;
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;