diff --git a/mallinkWebSocketServer/src/main/java/com/iformall/config/WebSocketConfig.java b/mallinkWebSocketServer/src/main/java/com/iformall/config/WebSocketConfig.java index 36c455607..1ef071f94 100644 --- a/mallinkWebSocketServer/src/main/java/com/iformall/config/WebSocketConfig.java +++ b/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; + } + } \ No newline at end of file diff --git a/mallinkWebSocketServer/src/main/java/com/iformall/controller/SendController.java b/mallinkWebSocketServer/src/main/java/com/iformall/controller/SendController.java index 0dfcb45fc..85b9e1324 100644 --- a/mallinkWebSocketServer/src/main/java/com/iformall/controller/SendController.java +++ b/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); } diff --git a/mallinkWebSocketServer/src/main/java/com/iformall/controller/websocket/ReceiverController.java b/mallinkWebSocketServer/src/main/java/com/iformall/controller/websocket/ReceiverController.java index 5a25f2641..1f4118288 100644 --- a/mallinkWebSocketServer/src/main/java/com/iformall/controller/websocket/ReceiverController.java +++ b/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 users = userService.listSessionIds(); + for(String user:users){ + System.out.println(Constance.queue_pre+user); + amqpTemplate.convertAndSend("", Constance.queue_pre+user, message.getMessage()); + } + } + } \ No newline at end of file diff --git a/mallinkWebSocketServer/src/main/java/com/iformall/model/Message.java b/mallinkWebSocketServer/src/main/java/com/iformall/model/Message.java new file mode 100644 index 000000000..fc654018e --- /dev/null +++ b/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; +} diff --git a/mallinkWebSocketServer/src/main/java/com/iformall/model/User.java b/mallinkWebSocketServer/src/main/java/com/iformall/model/User.java new file mode 100644 index 000000000..c55dfd834 --- /dev/null +++ b/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; + } +} diff --git a/mallinkWebSocketServer/src/main/java/com/iformall/service/UserService.java b/mallinkWebSocketServer/src/main/java/com/iformall/service/UserService.java new file mode 100644 index 000000000..9c6d09313 --- /dev/null +++ b/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 boundHashOperations = redisTemplate.boundHashOps(WSUSERS); + boundHashOperations.put(name, sessionId); + } + + public void deleteUser(String name){ + BoundHashOperations boundHashOperations = redisTemplate.boundHashOps(WSUSERS); + boundHashOperations.delete(name); + } + + public String getSessionId(String name){ + BoundHashOperations boundHashOperations = redisTemplate.boundHashOps(WSUSERS); + return boundHashOperations.get(name); + } + + public List listSessionIds(){ + BoundHashOperations boundHashOperations = redisTemplate.boundHashOps(WSUSERS); + List values = new ArrayList(); + for(String key:boundHashOperations.keys()){ + System.out.println(key); + values.add(boundHashOperations.get(key)); + } + return values; + } + +} diff --git a/mallinkWebSocketServer/src/main/java/com/iformall/util/Constance.java b/mallinkWebSocketServer/src/main/java/com/iformall/util/Constance.java new file mode 100644 index 000000000..80011c9cf --- /dev/null +++ b/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"; +}