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

build.gradle

@@ -23,6 +23,7 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
     implementation 'org.springframework.boot:spring-boot-starter-web'
     implementation "org.springframework.boot:spring-boot-starter-security"
+    implementation "org.springframework.boot:spring-boot-starter-websocket"
     implementation 'org.springframework.session:spring-session-core'
     compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.7'
     compile 'org.pcap4j:pcap4j-core:1.+'

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

@@ -10,10 +10,12 @@ import org.springframework.security.config.annotation.web.configuration.EnableWe
 import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter;
 import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
 import org.springframework.security.crypto.password.PasswordEncoder;
+import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
+import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
 
 @Configuration
 @EnableWebSecurity
-public class SecurityConfiguration extends WebSecurityConfigurerAdapter {
+public class ApplicationConfiguration extends WebSecurityConfigurerAdapter implements WebSocketConfigurer {
 
     @Value("${account-login}")
     private String login;
@@ -21,6 +23,13 @@ public class SecurityConfiguration extends WebSecurityConfigurerAdapter {
     @Value("${account-password}")
     private String password;
 
+    private final WebSocketHandler webSocketHandler;
+
+    @Autowired
+    public ApplicationConfiguration(WebSocketHandler webSocketHandler) {
+        this.webSocketHandler = webSocketHandler;
+    }
+
     @Autowired
     public void configureGlobal(AuthenticationManagerBuilder auth) throws Exception {
         auth.inMemoryAuthentication()
@@ -30,7 +39,9 @@ public class SecurityConfiguration extends WebSecurityConfigurerAdapter {
 
     @Override
     protected void configure(HttpSecurity http) throws Exception {
-        http.authorizeRequests()
+        http.csrf()
+                .disable()
+                .authorizeRequests()
                 .anyRequest().authenticated()
                 .and()
                 .httpBasic();
@@ -41,4 +52,8 @@ public class SecurityConfiguration extends WebSecurityConfigurerAdapter {
         return new BCryptPasswordEncoder();
     }
 
+    @Override
+    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
+        registry.addHandler(webSocketHandler, "/ws").withSockJS();
+    }
 }

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

@@ -16,10 +16,7 @@ import org.springframework.stereotype.Component;
 import ru.serega6531.packmate.model.CtfService;
 import ru.serega6531.packmate.model.Stream;
 import ru.serega6531.packmate.model.UnfinishedStream;
-import ru.serega6531.packmate.service.PacketService;
+import ru.serega6531.packmate.service.*;
-import ru.serega6531.packmate.service.PatternService;
-import ru.serega6531.packmate.service.ServicesService;
-import ru.serega6531.packmate.service.StreamService;
 
 import javax.annotation.PreDestroy;
 import java.net.Inet4Address;
@@ -36,6 +33,7 @@ public class PcapWorker {
     private final StreamService streamService;
     private final PacketService packetService;
     private final PatternService patternService;
+    private final PacketsSubscriptionService subscriptionService;
 
     private final PcapNetworkInterface device;
     private PcapHandle pcap = null;
@@ -56,12 +54,14 @@ public class PcapWorker {
                       StreamService streamService,
                       PacketService packetService,
                       PatternService patternService,
+                      PacketsSubscriptionService subscriptionService,
                       @Value("${interface-name}") String interfaceName,
                       @Value("${local-ip}") String localIp) throws PcapNativeException {
         this.servicesService = servicesService;
         this.streamService = streamService;
         this.packetService = packetService;
         this.patternService = patternService;
+        this.subscriptionService = subscriptionService;
 
         this.localIp = localIp;
 
@@ -195,7 +195,7 @@ public class PcapWorker {
                     if(rst || (acksForStream.contains(sourceIpAndPort) && acksForStream.contains(destIpAndPort))) {
                         final Stream finishedStream = saveStream(stream);
                         log.info("Конец стрима");
-                        //TODO send to ws
+                        subscriptionService.broadcastNewStream(finishedStream);
                     }
                 }
             }

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

@@ -0,0 +1,29 @@
+package ru.serega6531.packmate;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+import ru.serega6531.packmate.service.PacketsSubscriptionService;
+
+@Component
+public class WebSocketHandler extends TextWebSocketHandler {
+
+    private final PacketsSubscriptionService subscriptionService;
+
+    @Autowired
+    public WebSocketHandler(PacketsSubscriptionService subscriptionService) {
+        this.subscriptionService = subscriptionService;
+    }
+
+    @Override
+    public void afterConnectionEstablished(WebSocketSession session) {
+        subscriptionService.addSubscriber(session);
+    }
+
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
+        subscriptionService.removeSubscriber(session);
+    }
+}

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

@@ -0,0 +1,53 @@
+package ru.serega6531.packmate.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import ru.serega6531.packmate.model.Stream;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+@Service
+@Slf4j
+public class PacketsSubscriptionService {
+
+    private List<WebSocketSession> subscribers = new ArrayList<>();
+
+    private final ObjectMapper mapper;
+
+    @Autowired
+    public PacketsSubscriptionService(ObjectMapper mapper) {
+        this.mapper = mapper;
+    }
+
+    public void addSubscriber(WebSocketSession session) {
+        log.info("Подписан пользователь {}", session.getRemoteAddress().getHostName());
+        subscribers.add(session);
+    }
+
+    public void removeSubscriber(WebSocketSession session) {
+        log.info("Отписан пользователь {}", session.getRemoteAddress().getHostName());
+        subscribers.remove(session);
+    }
+
+    public void broadcastNewStream(Stream stream) {
+        subscribers.forEach(s -> {
+            try {
+                s.sendMessage(objectToTextMessage(stream));
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        });
+    }
+
+    private TextMessage objectToTextMessage(Object object) throws JsonProcessingException {
+        return new TextMessage(mapper.writeValueAsString(object));
+    }
+
+}

src/main/resources/static/index.html

@@ -0,0 +1 @@
+<html>test</html>