New message notification

crypto-messenger-desktop/pom.xml

@@ -28,6 +28,11 @@
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-openfeign</artifactId>

crypto-messenger-desktop/src/main/java/cryptomessenger/Ws.java

@@ -0,0 +1,92 @@
+package cryptomessenger;
+
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.event.EventListener;
+import org.springframework.messaging.converter.StringMessageConverter;
+import org.springframework.messaging.simp.stomp.*;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.WebSocketHttpHeaders;
+import org.springframework.web.socket.client.standard.StandardWebSocketClient;
+import org.springframework.web.socket.messaging.WebSocketStompClient;
+
+import java.lang.reflect.Type;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class Ws {
+
+    private final SessionWsHandler sessionWsHandler;
+    private final NewMessageWsHandler newMessageWsHandler;
+    private final CallWsHandler callWsHandler;
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void run() {
+        var client = new WebSocketStompClient(new StandardWebSocketClient());
+        client.setMessageConverter(new StringMessageConverter());
+
+        var headers = new WebSocketHttpHeaders();
+        headers.add("X-Auth", "It's me!");
+
+        client.connectAsync("ws://localhost:8080/ws", headers, sessionWsHandler).thenAccept(session -> {
+            log.info("Session: {}", session.getSessionId());
+            session.subscribe("/user/alex/calls/new", newMessageWsHandler);
+            session.subscribe("/user/kirill/messages/new", callWsHandler);
+        });
+    }
+
+    @Slf4j
+    @Component
+    public static class NewMessageWsHandler implements StompFrameHandler {
+        @Override
+        public Type getPayloadType(StompHeaders headers) {
+            return String.class;
+        }
+
+        @Override
+        public void handleFrame(StompHeaders headers, Object payload) {
+            log.info(">> [New Message]: {}", payload);
+        }
+    }
+
+    @Slf4j
+    @Component
+    public static class CallWsHandler implements StompFrameHandler {
+        @Override
+        public Type getPayloadType(StompHeaders headers) {
+            return String.class;
+        }
+
+        @Override
+        public void handleFrame(StompHeaders headers, Object payload) {
+            log.info(">> [Call]: {}", payload);
+        }
+    }
+
+    @Slf4j
+    @Component
+    public static class SessionWsHandler extends StompSessionHandlerAdapter {
+
+        @Lazy
+        @Autowired
+        private Ws ws;
+
+        @Override
+        public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, Throwable exception) {
+            log.error("Exception", exception);
+        }
+
+        @Override
+        @SneakyThrows
+        public void handleTransportError(StompSession session, Throwable exception) {
+            log.error("Error", exception);
+            Thread.sleep(5000);
+            ws.run();
+        }
+    }
+}

crypto-messenger-server/pom.xml

@@ -18,6 +18,11 @@
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-data-mongodb</artifactId>

crypto-messenger-server/src/main/java/cryptomessenger/server/infrastructure/websocket/NewMessageNotifier.java

@@ -0,0 +1,20 @@
+package cryptomessenger.server.infrastructure.websocket;
+
+import cryptomessenger.server.service.message.Message;
+import cryptomessenger.server.service.message.NewMessageListener;
+import lombok.RequiredArgsConstructor;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class NewMessageNotifier implements NewMessageListener {
+
+    private final SimpMessagingTemplate simpMessagingTemplate;
+
+    @Override
+    public void onNewMessage(Message message) {
+        var receiver = message.getReceiverId().toString();
+        simpMessagingTemplate.convertAndSendToUser(receiver, "/messages/new", "New message");
+    }
+}

crypto-messenger-server/src/main/java/cryptomessenger/server/infrastructure/websocket/WebSocketConfiguration.java

@@ -0,0 +1,16 @@
+package cryptomessenger.server.infrastructure.websocket;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
+
+@Configuration
+@EnableWebSocketMessageBroker
+public class WebSocketConfiguration implements WebSocketMessageBrokerConfigurer {
+
+    @Override
+    public void registerStompEndpoints(StompEndpointRegistry registry) {
+        registry.addEndpoint("/ws");
+    }
+}

crypto-messenger-server/src/main/java/cryptomessenger/server/service/message/MessageServiceImpl.java

@@ -6,6 +6,7 @@
 import org.springframework.stereotype.Service;
 
 import java.time.LocalDateTime;
+import java.util.Set;
 import java.util.UUID;
 
 import static java.util.UUID.randomUUID;
@@ -15,6 +16,7 @@
 public class MessageServiceImpl implements MessageService {
 
     private final MessageRepository messageRepository;
+    private final Set<NewMessageListener> newMessageListeners;
 
     @Override
     public Message send(MessageSending sending) {
@@ -26,7 +28,9 @@ public Message send(MessageSending sending) {
                 .contentForSender(sending.getContentForSender())
                 .contentForReceiver(sending.getContentForReceiver())
                 .build();
-        return messageRepository.save(message);
+        messageRepository.save(message);
+        newMessageListeners.forEach(listener -> listener.onNewMessage(message));
+        return message;
     }
 
     @Override

crypto-messenger-server/src/main/java/cryptomessenger/server/service/message/NewMessageListener.java

@@ -0,0 +1,6 @@
+package cryptomessenger.server.service.message;
+
+public interface NewMessageListener {
+
+    void onNewMessage(Message message);
+}