chatroom-dev-001-cluster-with-redis : integrate with Redis

springBootBlog/README.md

@@ -2,6 +2,8 @@
 
 <p align="center"><img src ="./doc/pic/SpringBlog.svg"></p>
 
+<p align="center"><img src ="./doc/pic/blog_ui_1.png"></p>
+
 Build a blog system via spring boot. Functionality : CRUD, Login/Logout (Spring security, OAuth), Blog post, edit, Comment, New user register, edit, Blog search
 
 

springChatRoom/README.md

@@ -5,20 +5,51 @@ Build a chat system with webSocket, run in Spring boot system.
 - UI : http://localhost:8080
 
 ## Tech
-- Java 11
+- Java 17
 - spring boot 3.0.12
 - webSocket
 - STOMP
 - FE
   - SockJS (webSocket client)
-- Extension
+- Cluster version
     - Redis
 
-## Run
+## Run (single)
+
+<details>
+<summary>Run (single)</summary>
+
 ```bash
+# compile
 mvn package -DskipTests
+# run app
 java -jar target/springChatRoom-0.0.1-SNAPSHOT.jar
 ```
+</details>
+
+## Run (cluster)
+
+<details>
+<summary>Run (single)</summary>
+
+```bash
+
+# install redis
+# https://github.com/yennanliu/utility_shell/blob/master/redis/install_redis_mac.sh
+
+# run redis
+brew services start redis
+# test (local)
+redis-cli ping
+# to the redis shell (local)
+redis-cli
+
+# compile
+mvn package -DskipTests
+# run app
+java -jar target/springChatRoom-0.0.1-SNAPSHOT.jar
+```
+</details>
 
 ## Todo
 - Feature
@@ -34,6 +65,7 @@ java -jar target/springChatRoom-0.0.1-SNAPSHOT.jar
   - offer user counts/online users (offer API)
 
 ## Knowledge
+
 - WebSocket
 	- Server can send info to client (bi-directon communicartion, TCP protocol)
 		- 3 ways handshake

springChatRoom/doc/ref.md

@@ -1,4 +1,5 @@
 ## Ref
+
 - Article
 	- https://blog.csdn.net/qqxx6661/article/details/98883166
 	- https://github.com/yennanliu/springboot-websocket-demo

springChatRoom/pom.xml

@@ -38,11 +38,11 @@
 			<scope>test</scope>
 		</dependency>
 
-<!--		&lt;!&ndash; redis &ndash;&gt;-->
-<!--		<dependency>-->
-<!--			<groupId>org.springframework.boot</groupId>-->
-<!--			<artifactId>spring-boot-starter-data-redis</artifactId>-->
-<!--		</dependency>-->
+		<!-- redis -->
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-redis</artifactId>
+		</dependency>
 
 		<!-- json -->
 		<dependency>

springChatRoom/src/main/java/com/yen/springChatRoom/ChatRoomApplication.java

@@ -2,7 +2,10 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.ComponentScan;
 
+// NOTE !!! DO NOT enable below, or will cause websocket connection error
+//@ComponentScan(basePackages = "com.yen.springChatRoom.redis.RedisListenerBean") // https://blog.csdn.net/automal/article/details/111859409
 @SpringBootApplication
 public class ChatRoomApplication {
 

springChatRoom/src/main/java/com/yen/springChatRoom/controller/ChatController.java

@@ -1,6 +1,10 @@
 package com.yen.springChatRoom.controller;
 
 import com.yen.springChatRoom.model.ChatMessage;
+import com.yen.springChatRoom.util.JsonUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.messaging.handler.annotation.MessageMapping;
 import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.handler.annotation.SendTo;
@@ -12,31 +16,79 @@
 @Controller
 public class ChatController {
 
+    @Value("${redis.channel.msgToAll}")
+    private String msgToAll;
+
+    @Value("${redis.set.onlineUsers}")
+    private String onlineUsers;
+
+    @Value("${redis.channel.userStatus}")
+    private String userStatus;
+
+    // TODO : check difference ? RedisTemplate VS RedisTemplate<String, String>
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+    //private RedisTemplate redisTemplate;
+
     private static final Logger LOGGER = LoggerFactory.getLogger(ChatController.class);
 
+    /**
+     *  single mode : read msg from FE, and send to
+     *                other users (@SendTo("/topic/public")) directly
+     */
+//    @MessageMapping("/chat.sendMessage")
+//    @SendTo("/topic/public")
+//    public ChatMessage sendMessage(@Payload ChatMessage chatMessage){
+//
+//        return chatMessage;
+//    }
+
+    /**
+     *  cluster mode : read msg from FE, but NOT send to other users,
+     *                 instead, send to Redis channel, so the other service
+     *                 on cluster can read/digest the msg
+     */
     @MessageMapping("/chat.sendMessage")
-    @SendTo("/topic/public")
-    public ChatMessage sendMessage(@Payload ChatMessage chatMessage){
-        return chatMessage;
+    public void sendMessage(@Payload ChatMessage chatMessage){
+        try{
+            //redisTemplate.convertAndSend(msgToAll, JsonUtil.parseObjToJson(chatMessage)));
+            redisTemplate.convertAndSend(msgToAll, JsonUtil.parseObjToJson(chatMessage));
+        }catch (Exception e){
+            LOGGER.error("send msg error : " + e.getMessage(), e);
+        }
     }
 
+
     @MessageMapping("/chat.addUser")
-    @SendTo("/topic/public")
-    public ChatMessage addUser(@Payload ChatMessage chatMessage, SimpMessageHeaderAccessor headerAccessor) {
+    public void addUser(@Payload ChatMessage chatMessage, SimpMessageHeaderAccessor headerAccessor) {
 
         LOGGER.info("User added in Chatroom:" + chatMessage.getSender());
-        // add username in web socket session
-        headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
-        return chatMessage;
-
-        // TODO : update with below
-//        try {
-//            headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
-////            redisTemplate.opsForSet().add(onlineUsers, chatMessage.getSender());
-////            redisTemplate.convertAndSend(userStatus, JsonUtil.parseObjToJson(chatMessage));
-//        } catch (Exception e) {
-//            LOGGER.error(e.getMessage(), e);
-//        }
+        try {
+            headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
+            redisTemplate.opsForSet().add(onlineUsers, chatMessage.getSender());
+            redisTemplate.convertAndSend(userStatus, JsonUtil.parseObjToJson(chatMessage));
+        } catch (Exception e) {
+            LOGGER.error(e.getMessage(), e);
+        }
     }
 
+//    @MessageMapping("/chat.addUser")
+//    @SendTo("/topic/public")
+//    public ChatMessage addUser(@Payload ChatMessage chatMessage, SimpMessageHeaderAccessor headerAccessor) {
+//
+//        LOGGER.info("User added in Chatroom:" + chatMessage.getSender());
+//        // add username in web socket session
+//        headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
+//        return chatMessage;
+//
+//        // TODO : update with below
+////        try {
+////            headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
+//////            redisTemplate.opsForSet().add(onlineUsers, chatMessage.getSender());
+//////            redisTemplate.convertAndSend(userStatus, JsonUtil.parseObjToJson(chatMessage));
+////        } catch (Exception e) {
+////            LOGGER.error(e.getMessage(), e);
+////        }
+//    }
+
 }

springChatRoom/src/main/java/com/yen/springChatRoom/listener/WebSocketEventListener.java

@@ -2,47 +2,98 @@
 
 import com.yen.springChatRoom.controller.ChatController;
 import com.yen.springChatRoom.model.ChatMessage;
+import com.yen.springChatRoom.util.JsonUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.event.EventListener;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.messaging.simp.SimpMessageSendingOperations;
 import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.messaging.SessionConnectedEvent;
 import org.springframework.web.socket.messaging.SessionDisconnectEvent;
 
+import java.net.Inet4Address;
+import java.net.InetAddress;
+
 @Component
 public class WebSocketEventListener {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ChatController.class);
 
+    @Value("${server.port}")
+    private String serverPort;
+
+    @Value("${redis.set.onlineUsers}")
+    private String onlineUsers;
+
+    @Value("${redis.channel.userStatus}")
+    private String userStatus;
+
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+
     @Autowired
     private SimpMessageSendingOperations messagingTemplate;
 
     // connect
-    @EventListener
-    public void handleWebSocketConnectListener(SessionConnectedEvent event){
+//    @EventListener
+//    public void handleWebSocketConnectListener(SessionConnectedEvent event){
+//
+//        LOGGER.info("Receive a new web socket connection!");
+//    }
+    public void handleWebSocketConnectListener(SessionConnectedEvent event) {
+
+        InetAddress localHost;
+        try {
+            localHost = Inet4Address.getLocalHost();
+            LOGGER.info("Received a new web socket connection from:" + localHost.getHostAddress() + ":" + serverPort);
+        } catch (Exception e) {
+            LOGGER.error(e.getMessage(), e);
+        }
 
-        LOGGER.info("Receive a new web socket connection!");
     }
 
+
     // disconnect
     @EventListener
-    public void handleWebSocketDisConnectListener(SessionDisconnectEvent event){
+    public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {
 
         StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
-        String username = (String) headerAccessor.getSessionAttributes().get("username");
 
-        if (username != null){
+        String username = (String) headerAccessor.getSessionAttributes().get("username");
 
-            LOGGER.info("User disconnected : " + username);
+        if(username != null) {
+            LOGGER.info("User Disconnected : " + username);
             ChatMessage chatMessage = new ChatMessage();
             chatMessage.setType(ChatMessage.MessageType.LEAVE);
             chatMessage.setSender(username);
-
-            messagingTemplate.convertAndSend("/topic/public", chatMessage);
+            try {
+                redisTemplate.opsForSet().remove(onlineUsers, username);
+                redisTemplate.convertAndSend(userStatus, JsonUtil.parseObjToJson(chatMessage));
+            } catch (Exception e) {
+                LOGGER.error(e.getMessage(), e);
+            }
         }
     }
 
+//    @EventListener
+//    public void handleWebSocketDisConnectListener(SessionDisconnectEvent event){
+//
+//        StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
+//        String username = (String) headerAccessor.getSessionAttributes().get("username");
+//
+//        if (username != null){
+//
+//            LOGGER.info("User disconnected : " + username);
+//            ChatMessage chatMessage = new ChatMessage();
+//            chatMessage.setType(ChatMessage.MessageType.LEAVE);
+//            chatMessage.setSender(username);
+//
+//            messagingTemplate.convertAndSend("/topic/public", chatMessage);
+//        }
+//    }
+
 }

springChatRoom/src/main/java/com/yen/springChatRoom/redis/RedisListenerBean.java

@@ -0,0 +1,52 @@
+package com.yen.springChatRoom.redis;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.listener.PatternTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.stereotype.Component;
+
+import java.net.Inet4Address;
+import java.net.InetAddress;
+import org.springframework.stereotype.Component;
+
+
+/** Class for Redis channel conn */
+@Component
+public class RedisListenerBean {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(RedisListenerBean.class);
+
+    // read setting from config
+    @Value("${server.port}")
+    private String serverPort;
+
+    @Value("${redis.channel.msgToAll}")
+    private String msgToAll;
+
+    @Value("${redis.channel.userStatus}")
+    private String userStatus;
+
+    /** Redis channel bean
+     *
+     *   1. listen Redis channel via binding (for example : container.addMessageListener)
+     *   2. can do further biz logic in method
+     */
+    @Bean
+    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter){
+
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(connectionFactory);
+
+        // listen msgToAll (Redis channel)
+        container.addMessageListener(listenerAdapter, new PatternTopic(msgToAll));
+        container.addMessageListener(listenerAdapter, new PatternTopic(userStatus));
+        LOGGER.info("Subscribe Redis channel : " + msgToAll);
+        return container;
+    }
+
+}