[WebSocket][新增]:支持websocket

mallinkWebSocketServer/src/main/java/com/iformall/config/WebSocketConfig.java

@@ -1,11 +1,27 @@
 package com.iformall.config;
 
 import com.iformall.interceptor.AuthHandshakeInterceptor;
+import com.iformall.model.User;
+import com.iformall.service.UserService;
+import com.iformall.util.Constance;
 import lombok.Data;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.server.ServerHttpRequest;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
+import org.springframework.messaging.simp.config.ChannelRegistration;
 import org.springframework.messaging.simp.config.MessageBrokerRegistry;
+import org.springframework.messaging.simp.stomp.StompCommand;
+import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.MessageHeaderAccessor;
 import org.springframework.scheduling.concurrent.DefaultManagedTaskScheduler;
 import org.springframework.web.socket.WebSocketHandler;
 import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
@@ -14,6 +30,7 @@ import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerCo
 import org.springframework.web.socket.server.support.DefaultHandshakeHandler;
 
 import java.security.Principal;
+import java.util.LinkedList;
 import java.util.Map;
 
 /**
@@ -29,12 +46,20 @@ import java.util.Map;
 @EnableWebSocketMessageBroker
 public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
 
+    private Logger logger = LoggerFactory.getLogger(this.getClass());
+
     @Autowired
     private RabbitMqProperty property;
 
     @Autowired
     private AuthHandshakeInterceptor authHandshakeInterceptor;
 
+    @Autowired
+    private UserService userService;
+
+    @Autowired
+    private RabbitAdmin rabbitAdmin;
+
     /**
      * 配置了一个简单的消息代理，
      *    如果不重载，默认情况下回自动配置一个简单的内存消息代理，用来处理以"/topic"为前缀的消息。这里重载configureMessageBroker()方法，
@@ -69,8 +94,9 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
          * 在网页上可以通过"/gs-guide-websocket"来和服务器的WebSocket连接
          * 这个和客户端创建连接时的url有关，其中setAllowedOrigins()方法表示允许连接的域名，withSockJS()方法表示支持以SockJS方式连接服务器。
          */
-        registry.addEndpoint("/ws").setAllowedOrigins("*")
-                .addInterceptors(authHandshakeInterceptor);
+        registry.addEndpoint("/ws")
+                .setAllowedOrigins("*")
+                ;//.addInterceptors(authHandshakeInterceptor);
     }
 
     /**
@@ -78,15 +104,39 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
      *
      * @param registration
      */
-    /*
     @Override
     public void configureClientInboundChannel(ChannelRegistration registration) {
-        registration.interceptors(inboundChannelInterceptor);
-        registration.taskExecutor()    // 线程信息
-                .corePoolSize(400)     // 核心线程池
-                .maxPoolSize(800)      // 最多线程池数
-                .keepAliveSeconds(60); // 超过核心线程数后，空闲线程超时60秒则杀死
-    }*/
+        registration.interceptors(new ChannelInterceptor() {
+            @Override
+            public Message<?> preSend(Message<?> message, MessageChannel channel) {
+
+                StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
+
+                logger.info("interceptors-> "+accessor.getCommand()+"-"+accessor);
+
+                if (StompCommand.CONNECT.equals(accessor.getCommand())) {// 创建连接
+                    Object raw = message.getHeaders().get(SimpMessageHeaderAccessor.NATIVE_HEADERS);
+                    if (raw instanceof Map) {
+                        Object name = ((Map) raw).get("login");
+                        if (name instanceof LinkedList) {
+                            // 设置当前访问器的认证用户
+                            String userName = ((LinkedList) name).get(0).toString();
+                            System.out.println("regist userName:"+userName);
+                            accessor.setUser(new User(userName));
+                            userService.addUser(userName, accessor.getSessionId());
+                        }
+                    }
+                }
+
+                if (StompCommand.DISCONNECT.equals(accessor.getCommand())) {// 断开连接
+                    String userName = accessor.getUser().getName();
+                    rabbitAdmin.deleteQueue(Constance.queue_pre+userService.getSessionId(userName));
+                    userService.deleteUser(userName);
+                }
+                return message;
+            }
+        });
+    }
 
     /**
      * 消息传输参数配置
@@ -102,4 +152,10 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
     }
     */
 
+    @Bean
+    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
+        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
+        return rabbitAdmin;
+    }
+
 }

mallinkWebSocketServer/src/main/java/com/iformall/controller/SendController.java

@@ -2,12 +2,13 @@
 package com.iformall.controller;
 
 import com.alibaba.fastjson.JSON;
+import com.iformall.service.UserService;
+import com.iformall.util.Constance;
+import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.messaging.simp.SimpMessagingTemplate;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.*;
+
 
 /**
  * @author Stormeye
@@ -16,42 +17,50 @@ import org.springframework.web.bind.annotation.RestController;
 @RequestMapping("/websocket")
 public class SendController {
 
+    @Autowired
+    private AmqpTemplate amqpTemplate;
+
     @Autowired
     private SimpMessagingTemplate messagingTemplate;
 
+    @Autowired
+    private UserService userService;
+
 
     /**
-     * 通知消息
+     * ADMIN通知消息
      */
-    @GetMapping("/noticea")
-    public void noticea() {
-        messagingTemplate.convertAndSend("/topic/1001A", JSON.toJSONString("这是A通知消息！！"));
+    @GetMapping("/noticeA/{tenantId}")
+    public void noticea(@PathVariable("tenantId") String tenantId, @RequestParam String message) {
+        messagingTemplate.convertAndSend("/topic/"+tenantId+"A", JSON.toJSONString(message));
     }
 
     /**
-     * 通知消息
+     * B端通知消息
      */
-    @GetMapping("/noticeb")
-    public void noticeb() {
-        messagingTemplate.convertAndSend("/topic/1001B", JSON.toJSONString("这是B通知消息！！"));
+    @GetMapping("/noticeB/{tenantId}")
+    public void noticeb(@PathVariable("tenantId") String tenantId, @RequestParam String message) {
+        messagingTemplate.convertAndSend("/topic/"+tenantId+"B", JSON.toJSONString(message));
     }
 
     /**
      * 通知消息
      */
-    @GetMapping("/noticec")
-    public void noticec() {
-        messagingTemplate.convertAndSend("/topic/1001C", JSON.toJSONString("这是B通知消息！！"));
+    @GetMapping("/noticeC/{tenantId}")
+    public void noticec(@PathVariable("tenantId") String tenantId, @RequestParam String message) {
+        messagingTemplate.convertAndSend("/topic/"+tenantId+"C", JSON.toJSONString(message));
     }
 
     /**
      * 具体用户消息
      */
     @GetMapping("/user/{name}")
-    public void user(@PathVariable("name") String name) {
-        messagingTemplate.convertAndSendToUser(name, "/message", JSON.toJSONString("这是发送给sa;fa;sfj" + name + "用户的消息！！"));
-        ///messagingTemplate.convertAndSendToUser(name, "/user/"+name+"/message", JSON.toJSONString("这是发送给" + name + "用户的消息！！"));
-        //messagingTemplate.convertAndSend("/user/"+name+"/message", JSON.toJSONString("用户的消息！！"));
+    public void user(@PathVariable("name") String name, @RequestParam String message) {
+        // 方式1
+        messagingTemplate.convertAndSendToUser(name, "/message", JSON.toJSONString(message));
+        // 方式2
+        //String userId = userService.getSessionId(name);
+        //amqpTemplate.convertAndSend("", Constance.queue_pre+userId, message);
     }
 
 

mallinkWebSocketServer/src/main/java/com/iformall/controller/websocket/ReceiverController.java

@@ -2,19 +2,24 @@ package com.iformall.controller.websocket;
 
 import com.alibaba.fastjson.JSONObject;
 import com.iformall.interceptor.AuthHandshakeInterceptor;
+import com.iformall.model.Message;
+import com.iformall.service.UserService;
+import com.iformall.util.Constance;
 import lombok.Data;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.messaging.handler.annotation.DestinationVariable;
 import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
 import org.springframework.messaging.simp.SimpMessageSendingOperations;
 import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
 import org.springframework.web.bind.annotation.RestController;
 
-import javax.websocket.Session;
 import java.security.Principal;
+import java.util.List;
 
 
 /**
@@ -30,12 +35,17 @@ public class ReceiverController {
     @Autowired
     SimpMessageSendingOperations simpMessageSendingOperations;
 
+    @Autowired
+    private AmqpTemplate amqpTemplate;
+
+    @Autowired
+    private UserService userService;
+
     /**
      * 接收客户端发来的消息
      */
     @MessageMapping("/message")
     public void handleSubscribe(String msg, StompHeaderAccessor stompHeaderAccessor) {
-        //System.out.println("测试发送消息：随机消息" +session.getId());
         Principal user = stompHeaderAccessor.getUser();
         String tenantId = (String) stompHeaderAccessor.getSessionAttributes().get(AuthHandshakeInterceptor.TENANT_ID);
         Long userId = (Long) stompHeaderAccessor.getSessionAttributes().get(AuthHandshakeInterceptor.LOGIN_USER_KEY);
@@ -58,5 +68,31 @@ public class ReceiverController {
         simpMessageSendingOperations.convertAndSend("/topic/chat-group/" + roomId, msg);
     }
 
+    /**
+     * 个人聊天消息
+     * @param message
+     * @param headerAccessor
+     */
+    @MessageMapping("/p2p")
+    public void p2p(Message message, SimpMessageHeaderAccessor headerAccessor) {
+        String userId = userService.getSessionId(message.getTo());
+        System.out.println(Constance.queue_pre+userId);
+        amqpTemplate.convertAndSend("", Constance.queue_pre+userId, message.getMessage());
+    }
+
+    /**
+     * 群发消息
+     * @param message
+     * @throws Exception
+     */
+    @MessageMapping("/sendall")
+    public void group(Message message) throws Exception {
+        List<String> users = userService.listSessionIds();
+        for(String user:users){
+            System.out.println(Constance.queue_pre+user);
+            amqpTemplate.convertAndSend("", Constance.queue_pre+user, message.getMessage());
+        }
+    }
+
 
 }

mallinkWebSocketServer/src/main/java/com/iformall/model/Message.java

@@ -0,0 +1,10 @@
+package com.iformall.model;
+
+import lombok.Data;
+
+@Data
+public class Message {
+    private String from;
+    private String to;
+    private String message;
+}

mallinkWebSocketServer/src/main/java/com/iformall/model/User.java

@@ -0,0 +1,16 @@
+package com.iformall.model;
+
+import java.security.Principal;
+
+public class User implements Principal {
+    private String name;
+
+    public User(String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+}

mallinkWebSocketServer/src/main/java/com/iformall/service/UserService.java

@@ -0,0 +1,43 @@
+package com.iformall.service;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.BoundHashOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Service
+public class UserService {
+    private static final String WSUSERS = "wsuser";
+
+    @Autowired
+    private RedisTemplate redisTemplate;
+
+    public void addUser(String name,String sessionId){
+        BoundHashOperations<String, String, String> boundHashOperations = redisTemplate.boundHashOps(WSUSERS);
+        boundHashOperations.put(name, sessionId);
+    }
+
+    public void deleteUser(String name){
+        BoundHashOperations<String, String, String> boundHashOperations = redisTemplate.boundHashOps(WSUSERS);
+        boundHashOperations.delete(name);
+    }
+
+    public String getSessionId(String name){
+        BoundHashOperations<String, String, String> boundHashOperations = redisTemplate.boundHashOps(WSUSERS);
+        return boundHashOperations.get(name);
+    }
+
+    public List<String> listSessionIds(){
+        BoundHashOperations<String, String, String> boundHashOperations = redisTemplate.boundHashOps(WSUSERS);
+        List<String> values = new ArrayList<String>();
+        for(String key:boundHashOperations.keys()){
+            System.out.println(key);
+            values.add(boundHashOperations.get(key));
+        }
+        return values;
+    }
+
+}

mallinkWebSocketServer/src/main/java/com/iformall/util/Constance.java

@@ -0,0 +1,5 @@
+package com.iformall.util;
+
+public class Constance {
+    public static final String queue_pre = "p2p-user";
+}