Рефакторинг

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

@@ -3,6 +3,7 @@ package ru.serega6531.packmate;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.pcap4j.core.*;
 import org.pcap4j.packet.IpV4Packet;
 import org.pcap4j.packet.Packet;
@@ -42,9 +43,9 @@ public class PcapWorker implements PacketListener {
 
     private final Map<UnfinishedStream, List<ru.serega6531.packmate.model.Packet>> unfinishedStreams = new HashMap<>();
 
-    // в следующих мапах в сетах srcIp соответствующего пакета
+    // в следующих мапах в Set находится srcIp соответствующего пакета
-    private final Map<UnfinishedStream, Set<String>> fins = new HashMap<>();
+    private final Map<UnfinishedStream, Set<ImmutablePair<Inet4Address, Integer>>> fins = new HashMap<>();
-    private final Map<UnfinishedStream, Set<String>> acks = new HashMap<>();
+    private final Map<UnfinishedStream, Set<ImmutablePair<Inet4Address, Integer>>> acks = new HashMap<>();
 
     @Autowired
     public PcapWorker(ServicesService servicesService,
@@ -55,7 +56,7 @@ public class PcapWorker implements PacketListener {
         this.streamService = streamService;
 
         this.localIp = InetAddress.getByName(localIpString);
-        if(!(this.localIp instanceof Inet4Address)) {
+        if (!(this.localIp instanceof Inet4Address)) {
             throw new IllegalArgumentException("Only ipv4 local ips are supported");
         }
 
@@ -143,9 +144,6 @@ public class PcapWorker implements PacketListener {
                 servicesService.findService(sourceIp, sourcePort, destIp, destPort);
 
         if (serviceOptional.isPresent()) {
-            String sourceIpAndPort = sourceIpString + ":" + sourcePort;
-            String destIpAndPort = destIpString + ":" + destPort;
-
             UnfinishedStream stream = addNewPacket(sourceIp, destIp, sourcePort, destPort, ttl, content, protocol);
 
             if (log.isDebugEnabled()) {
@@ -155,7 +153,7 @@ public class PcapWorker implements PacketListener {
             }
 
             if (protocol == Protocol.TCP) {  // udp не имеет фазы закрытия, поэтому закрываем по таймауту
-                checkTcpTermination(ack, fin, rst, sourceIpAndPort, destIpAndPort, stream);
+                checkTcpTermination(ack, fin, rst, new ImmutablePair<>(sourceIp, sourcePort), new ImmutablePair<>(destIp, destPort), stream);
             }
         } else { // сервис не найден
             if (log.isTraceEnabled()) {
@@ -189,7 +187,9 @@ public class PcapWorker implements PacketListener {
         return stream;
     }
 
-    private void checkTcpTermination(boolean ack, boolean fin, boolean rst, String sourceIpAndPort, String destIpAndPort, UnfinishedStream stream) {
+    private void checkTcpTermination(boolean ack, boolean fin, boolean rst,
+                                     ImmutablePair<Inet4Address, Integer> sourceIpAndPort, ImmutablePair<Inet4Address, Integer> destIpAndPort,
+                                     UnfinishedStream stream) {
         if (!fins.containsKey(stream)) {
             fins.put(stream, new HashSet<>());
         }
@@ -198,8 +198,8 @@ public class PcapWorker implements PacketListener {
             acks.put(stream, new HashSet<>());
         }
 
-        final Set<String> finsForStream = fins.get(stream);
+        final Set<ImmutablePair<Inet4Address, Integer>> finsForStream = fins.get(stream);
-        final Set<String> acksForStream = acks.get(stream);
+        final Set<ImmutablePair<Inet4Address, Integer>> acksForStream = acks.get(stream);
 
         if (fin) {
             finsForStream.add(sourceIpAndPort);
@@ -209,7 +209,7 @@ public class PcapWorker implements PacketListener {
             acksForStream.add(sourceIpAndPort);
         }
 
-        // если соединение разорвано или закрыто с помощью fin-ack-fin-ack
+        // если соединение разорвано с помощью rst или закрыто с помощью fin-ack-fin-ack
         if (rst || (acksForStream.contains(sourceIpAndPort) && acksForStream.contains(destIpAndPort))) {
             streamService.saveNewStream(stream, unfinishedStreams.get(stream));
 

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

@@ -40,7 +40,6 @@ public class StreamService {
     private final PacketService packetService;
     private final StreamSubscriptionService subscriptionService;
 
-    private final String localIp;
     private final boolean ignoreEmptyPackets;
 
     private static final byte[] GZIP_HEADER = {0x1f, (byte) 0x8b, 0x08};
@@ -59,7 +58,6 @@ public class StreamService {
         this.servicesService = servicesService;
         this.packetService = packetService;
         this.subscriptionService = subscriptionService;
-        this.localIp = localIp;
         this.ignoreEmptyPackets = ignoreEmptyPackets;
     }
 
@@ -69,10 +67,9 @@ public class StreamService {
     @Transactional
     public boolean saveNewStream(UnfinishedStream unfinishedStream, List<Packet> packets) {
         final Optional<CtfService> serviceOptional = servicesService.findService(
-                localIp,
+                unfinishedStream.getFirstIp(),
-                unfinishedStream.getFirstIp().getHostAddress(),
                 unfinishedStream.getFirstPort(),
-                unfinishedStream.getSecondIp().getHostAddress(),
+                unfinishedStream.getSecondIp(),
                 unfinishedStream.getSecondPort()
         );
 
@@ -175,7 +172,7 @@ public class StreamService {
                     final List<Packet> cut = packets.subList(start, i);
                     compress(packets, cut, incoming);
 
-                    i++;
+                    i++;  // продвигаем указатель на следующий после склеенного блок
                 }
                 start = i;
                 packetsInRow = 1;
@@ -246,7 +243,7 @@ public class StreamService {
                 gzipEndPacket = i - 1;
                 if(extractGzip(packets, gzipStartPacket, gzipEndPacket)) {
                     gzipStarted = false;
-                    i = gzipStartPacket + 1;
+                    i = gzipStartPacket + 1;  // продвигаем указатель на следующий после склеенного блок
                 }
             } else if (!packet.isIncoming()) {
                 String content = new String(packet.getContent());
@@ -258,7 +255,7 @@ public class StreamService {
                     gzipEndPacket = i - 1;
                     if(extractGzip(packets, gzipStartPacket, gzipEndPacket)) {
                         gzipStarted = false;
-                        i = gzipStartPacket + 1;
+                        i = gzipStartPacket + 1;  // продвигаем указатель на следующий после склеенного блок
                     }
                 }
 

src/main/java/ru/serega6531/packmate/utils/Bytes.java

@@ -1,5 +1,8 @@
 package ru.serega6531.packmate.utils;
 
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
 public class Bytes {
 
     public static int indexOf(byte[] array, byte[] target) {