gitee
/
roses
mirror of https://gitee.com/stylefeng/roses
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

删除socket无用逻辑

pull/22/head
rays 2021-06-07 15:09:39 +08:00
parent e23fc5d24b
commit 2b715d90fe
4 changed files with 6 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
kernel-d-socket/socket-api/src/main/java/cn/stylefeng/roses/kernel/socket/api/enums/ClientMessageTypeEnum.java
View File

@ -13,11 +13,6 @@ import lombok.Getter;
@Getter
public enum ClientMessageTypeEnum {
/**
* 添加用户监听的消息类型
*/
USER_ADD_MSG_TYPE("200001", "用户添加一个监听的消息类型"),
/**
* 用户心跳消息类型
*/

19
kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/spring/WebSocketApplicationRunnerImpl.java
View File

@ -1,21 +1,14 @@
package cn.stylefeng.roses.kernel.socket.business.websocket.spring;
import cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi;
import cn.stylefeng.roses.kernel.socket.api.expander.SocketConfigExpander;
import cn.stylefeng.roses.kernel.socket.websocket.message.WebSocketMessagePOJO;
import cn.stylefeng.roses.kernel.socket.websocket.server.WebSocketServer;
import cn.stylefeng.roses.kernel.socket.websocket.session.SessionCenter;
import com.gettyio.core.channel.config.ServerConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import java.net.StandardSocketOptions;
import static cn.stylefeng.roses.kernel.socket.api.enums.ClientMessageTypeEnum.*;
/**
* Spring Boot启动完成拉起WebSocket
*
@ -26,9 +19,6 @@ import static cn.stylefeng.roses.kernel.socket.api.enums.ClientMessageTypeEnum.*
@Slf4j
public class WebSocketApplicationRunnerImpl implements ApplicationRunner {
@Autowired
private SocketOperatorApi socketOperatorApi;
@Override
public void run(ApplicationArguments args) {
// 初始化配置对象
@ -51,14 +41,5 @@ public class WebSocketApplicationRunnerImpl implements ApplicationRunner {
WebSocketServer.run(aioServerConfig);
log.info("WebSocket Server Start Success!");
// 添加用户新增消息类型的回调
socketOperatorApi.msgTypeCallback(USER_ADD_MSG_TYPE.getCode(), (msgType, msg, socketSession) -> {
// 转换对象
WebSocketMessagePOJO webSocketMessage = (WebSocketMessagePOJO)msg;
// 维护会话中心的消息类型
SessionCenter.addSocketSessionMsgType(webSocketMessage.getData().toString(), webSocketMessage.getFormUserId());
});
}
}

18
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/WebSocketOperator.java
View File

@ -42,18 +42,12 @@ public class WebSocketOperator implements SocketOperatorApi {
@Override
public void sendMsgOfAllUserSession(String msgType, Object msg) {
// 获取监听该消息类型的所有会话
List<SocketSession<GettySocketOperator>> socketSessionList = SessionCenter.getSocketSessionByMsgType(msgType);
if (ObjectUtil.isNotEmpty(socketSessionList)) {
// 给所有会话发送消息
for (SocketSession<GettySocketOperator> socketSession : socketSessionList) {
WebSocketMessagePOJO webSocketMessagePOJO = new WebSocketMessagePOJO();
webSocketMessagePOJO.setData(msg);
webSocketMessagePOJO.setServerMsgType(msgType);
// 发送内容
socketSession.getSocketOperatorApi().writeAndFlush(webSocketMessagePOJO);
}
for (SocketSession<GettySocketOperator> socketSession : SessionCenter.getSocketSessionMap().values()) {
WebSocketMessagePOJO webSocketMessagePOJO = new WebSocketMessagePOJO();
webSocketMessagePOJO.setData(msg);
webSocketMessagePOJO.setServerMsgType(msgType);
// 发送内容
socketSession.getSocketOperatorApi().writeAndFlush(webSocketMessagePOJO);
}
}

80
kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/session/SessionCenter.java
View File

@ -1,10 +1,8 @@
package cn.stylefeng.roses.kernel.socket.websocket.session;
import cn.hutool.core.util.ObjectUtil;
import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession;
import cn.stylefeng.roses.kernel.socket.websocket.operator.channel.GettySocketOperator;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -23,11 +21,6 @@ public class SessionCenter {
*/
private static ConcurrentMap<String, SocketSession<GettySocketOperator>> socketSessionMap = new ConcurrentHashMap<>();
/**
* 消息类型和用户ID关系维护
*/
private static ConcurrentMap<String, Set<String>> messageTypeSessionMap = new ConcurrentHashMap<>();
/**
* 获取维护的所有会话
*
@ -39,17 +32,6 @@ public class SessionCenter {
return socketSessionMap;
}
/**
* 获取消息和用户ID的完整映射关系
*
* @return {@link ConcurrentMap< String, Set<String>>}
* @author majianguo
* @date 2021/6/1 下午2:14
**/
public static ConcurrentMap<String, Set<String>> getMessageTypeSessionMap() {
return messageTypeSessionMap;
}
/**
* 根据用户ID获取会话详情
*
@ -70,66 +52,7 @@ public class SessionCenter {
* @date 2021/6/1 下午1:49
**/
public static void addSocketSession(SocketSession<GettySocketOperator> socketSession) {
// 维护会话
socketSessionMap.put(socketSession.getUserId(), socketSession);
// 维护会话所有的消息类型和会话的关系
if (ObjectUtil.isNotEmpty(socketSession.getMessageTypes())) {
for (String messageType : socketSession.getMessageTypes()) {
Set<String> userIds = messageTypeSessionMap.get(messageType);
if (ObjectUtil.isEmpty(userIds)) {
userIds = new HashSet<>();
messageTypeSessionMap.put(messageType, userIds);
}
userIds.add(socketSession.getUserId());
}
}
}
/**
* 根据消息类型获取所有的会话
*
* @return {@link List< SocketSession<GettySocketOperator>>}
* @author majianguo
* @date 2021/6/1 下午2:06
**/
public static List<SocketSession<GettySocketOperator>> getSocketSessionByMsgType(String msgType) {
List<SocketSession<GettySocketOperator>> res = new ArrayList<>();
// 获取监听该消息所有的会话
Set<String> userIds = messageTypeSessionMap.get(msgType);
if (ObjectUtil.isNotEmpty(userIds)) {
for (String userId : userIds) {
SocketSession<GettySocketOperator> socketSession = socketSessionMap.get(userId);
res.add(socketSession);
}
}
return res;
}
/**
* 给会话添加监听的消息类型
*
* @param msgType 消息类型
* @param userId 用户ID
* @author majianguo
* @date 2021/6/1 下午2:11
**/
public static void addSocketSessionMsgType(String msgType, String userId) {
// 维护Session信息
SocketSession<GettySocketOperator> socketSession = socketSessionMap.get(userId);
if (ObjectUtil.isNotEmpty(socketSession)) {
socketSession.getMessageTypes().add(msgType);
}
// 维护消息列表
Set<String> userIds = messageTypeSessionMap.get(msgType);
if (ObjectUtil.isEmpty(userIds)) {
userIds = new HashSet<>();
messageTypeSessionMap.put(msgType, userIds);
}
userIds.add(userId);
}
/**
@ -141,8 +64,5 @@ public class SessionCenter {
**/
public static void closed(String userId) {
socketSessionMap.remove(userId);
for (Map.Entry<String, Set<String>> stringListEntry : messageTypeSessionMap.entrySet()) {
stringListEntry.getValue().removeIf(item -> item.equals(userId));
}
}
}