Добавлен поиск паттернов в прошлых стримах

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

@@ -1,8 +1,10 @@
 package ru.serega6531.packmate.service;
 
+import com.google.common.collect.Iterables;
 import lombok.extern.slf4j.Slf4j;
 import org.modelmapper.ModelMapper;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import ru.serega6531.packmate.model.FoundPattern;
 import ru.serega6531.packmate.model.Pattern;
@@ -14,7 +16,9 @@ import ru.serega6531.packmate.model.pojo.PatternDto;
 import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
 import ru.serega6531.packmate.repository.PatternRepository;
 
+import java.time.Instant;
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.PatternSyntaxException;
 import java.util.stream.Collectors;
 
@@ -23,6 +27,7 @@ import java.util.stream.Collectors;
 public class PatternService {
 
     private final PatternRepository repository;
+    private final StreamService streamService;
     private final SubscriptionService subscriptionService;
 
     private final Map<Integer, Pattern> patterns = new HashMap<>();
@@ -30,8 +35,10 @@ public class PatternService {
 
     @Autowired
     public PatternService(PatternRepository repository,
+                          @Lazy StreamService streamService,
                           SubscriptionService subscriptionService) {
         this.repository = repository;
+        this.streamService = streamService;
         this.subscriptionService = subscriptionService;
 
         repository.findAll().forEach(p -> patterns.put(p.getId(), p));
@@ -55,6 +62,11 @@ public class PatternService {
         return new PatternMatcher(bytes, list).findMatches();
     }
 
+    public Optional<FoundPattern> tryMatch(byte[] bytes, Pattern pattern) {
+        Set<FoundPattern> matches = new PatternMatcher(bytes, List.of(pattern)).findMatches();
+        return Optional.ofNullable(Iterables.getOnlyElement(matches, null));
+    }
+
     public void enable(int id, boolean enabled) {
         final Pattern pattern = find(id);
         if (pattern != null) {
@@ -81,13 +93,32 @@ public class PatternService {
             }
         }
 
+        pattern.setSearchStartTimestamp(System.currentTimeMillis());
+
         final Pattern saved = repository.save(pattern);
         patterns.put(saved.getId(), saved);
+
         log.info("Added new pattern '{}' with value '{}'", pattern.getName(), pattern.getValue());
         subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.SAVE_PATTERN, toDto(saved)));
+
         return saved;
     }
 
+    public void lookBack(int id, int minutes) {
+        final Pattern pattern = find(id);
+        if (pattern != null && pattern.getActionType() == PatternActionType.FIND) {
+            long end = pattern.getSearchStartTimestamp();
+            long start = end - TimeUnit.MINUTES.toMillis(minutes);
+
+            pattern.setSearchStartTimestamp(start);
+            repository.save(pattern);
+
+            log.info("Scanning for pattern '{}' between {} and {}", pattern.getName(),
+                    Instant.ofEpochMilli(start), Instant.ofEpochMilli(end));
+            streamService.processLookbackPattern(pattern, start, end);
+        }
+    }
+
     public Pattern fromDto(PatternDto dto) {
         return modelMapper.map(dto, Pattern.class);
     }

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

@@ -7,6 +7,7 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.jpa.domain.Specification;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -123,6 +124,21 @@ public class StreamService {
         return true;
     }
 
+    @Async
+    public void processLookbackPattern(Pattern pattern, long start, long end) {
+        List<Stream> streams = findAllBetweenTimestamps(start, end);
+
+        for (Stream stream : streams) {
+            boolean found = matchPattern(stream.getPackets(), pattern);
+            if (found) {
+                stream.getFoundPatterns().add(pattern);
+                repository.save(stream);
+            }
+        }
+
+        subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.FINISH_LOOKBACK, pattern.getId()));
+    }
+
     private void processUserAgent(List<Packet> packets, Stream stream) {
         String ua = null;
         for (Packet packet : packets) {
@@ -169,6 +185,30 @@ public class StreamService {
         return foundPatterns;
     }
 
+    private boolean matchPattern(List<Packet> packets, Pattern pattern) {
+        boolean matched = false;
+
+        for (Packet packet : packets) {
+            PatternDirectionType direction = packet.isIncoming() ? PatternDirectionType.INPUT : PatternDirectionType.OUTPUT;
+
+            if (pattern.getDirectionType() != PatternDirectionType.BOTH && pattern.getDirectionType() != direction) {
+                continue;
+            }
+
+            final Optional<FoundPattern> matchOpt = patternService.tryMatch(packet.getContent(), pattern);
+
+            if (matchOpt.isPresent()) {
+                FoundPattern match = matchOpt.get();
+                packet.getMatches().add(match);
+                match.setPacket(packet);
+
+                matched = true;
+            }
+        }
+
+        return matched;
+    }
+
     private boolean isStreamIgnored(List<Packet> packets) {
         for (Packet packet : packets) {
             PatternDirectionType direction = packet.isIncoming() ? PatternDirectionType.INPUT : PatternDirectionType.OUTPUT;
@@ -203,7 +243,6 @@ public class StreamService {
         repository.setFavorite(id, favorite);
     }
 
-    @SuppressWarnings("ConstantConditions")
     public List<Stream> findAll(Pagination pagination, Optional<Integer> service, boolean onlyFavorites) {
         PageRequest page = PageRequest.of(0, pagination.getPageSize(), pagination.getDirection(), "id");
 
@@ -229,6 +268,11 @@ public class StreamService {
         return repository.findAll(spec, page).getContent();
     }
 
+     public List<Stream> findAllBetweenTimestamps(long start, long end) {
+         Specification<Stream> spec = streamTimestampBetween(start, end);
+         return repository.findAll(spec);
+     }
+
     public StreamDto streamToDto(Stream stream) {
         return modelMapper.map(stream, StreamDto.class);
     }
@@ -253,6 +297,10 @@ public class StreamService {
         return (root, query, cb) -> cb.lessThan(root.get("id"), id);
     }
 
+    private Specification<Stream> streamTimestampBetween(long start, long end) {
+        return (root, query, cb) -> cb.between(root.get("startTimestamp"), start, end);
+    }
+
     private Specification<Stream> streamPatternsContains(Pattern pattern) {
         return (root, query, cb) -> cb.isMember(pattern, root.get("foundPatterns"));
     }