WebSocketServer.java
6.26 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
package com.infoloop.tianting.server;
import com.infoloop.tianting.constant.CommonConstants;
import com.infoloop.tianting.server.message.SocketMessage;
import com.infoloop.tianting.server.session.UserSessionData;
import com.infoloop.tianting.server.session.UserSessionKey;
import com.infoloop.tianting.server.session.UserTypeEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@Slf4j
public class WebSocketServer extends TextWebSocketHandler {
private static final ConcurrentHashMap<UserSessionKey, UserSessionData> userSessions = new ConcurrentHashMap<>();
private static final long INACTIVITY_TIMEOUT = 60 * 60 * 1000;
private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
static {
scheduler.scheduleAtFixedRate(WebSocketServer::cleanInactiveSessions, 1, 1, TimeUnit.HOURS);
}
private static void cleanInactiveSessions() {
final var currentTime = System.currentTimeMillis();
final var iterator = userSessions.entrySet().iterator();
log.info("cleanInactiveSessions; sessions:{}", userSessions.keySet());
while (iterator.hasNext()) {
final var entry = iterator.next();
final var key = entry.getKey();
final var userData = entry.getValue();
final var lastActiveTime = userData.getLastActiveTime();
if (currentTime - lastActiveTime > INACTIVITY_TIMEOUT) {
try {
userData.closeSession();
log.info("closeSession，userId : {}, userType : {}", key.getUserId(), key.getUserType());
} catch (IOException e) {
log.error("closeSession error，userId : {}, userType : {}", key.getUserId(), key.getUserType(), e);
}
iterator.remove();
}
}
}
public static <T> void sendMessageToUser(UserSessionKey key, SocketMessage<T> socketMessage) throws IOException {
sendMessageToUsers(List.of(key), socketMessage);
}
public static <T> void sendMessageToUsers(List<UserSessionKey> keys, SocketMessage<T> socketMessage) throws IOException {
for (final var key : keys) {
final var userData = userSessions.get(key);
if (userData != null && userData.getSession().isOpen()) {
userData.getSession().sendMessage(new TextMessage(socketMessage.toJsonString()));
userData.updateLastActiveTime();
log.info("send message to userId : {}, userType : {}", key.getUserId(), key.getUserType());
} else {
log.info("userId : {}, userType : {} WebSocket connect closed; ", key.getUserId(), key.getUserType());
}
}
}
public static <T> void sendMessageByUserType(UserTypeEnum userType, SocketMessage<T> socketMessage) throws IOException {
sendMessageByUserTypes(List.of(userType), socketMessage);
}
public static <T> void sendMessageByUserTypes(List<UserTypeEnum> userTypes, SocketMessage<T> socketMessage) throws IOException {
final var keys = userSessions.keySet().stream()
.filter(userSessionData -> userTypes.contains(userSessionData.getUserType()))
.collect(Collectors.toList());
sendMessageToUsers(keys, socketMessage);
}
public static void closeConnection(UserSessionKey key) {
final var userData = userSessions.get(key);
if (userData != null) {
try {
userData.closeSession();
userSessions.remove(key);
log.info("WebSocket close; userId = {}, userType = {}", key.getUserId(), key.getUserType());
} catch (IOException e) {
log.error("WebSocket close failed: userId = {}, userType = {}", key.getUserId(), key.getUserType(), e);
}
} else {
log.warn("Not Found userId : {}, userType : {} WebSocket Connect ", key.getUserId(), key.getUserType());
}
}
private UserSessionKey getUserSessionKey(WebSocketSession session) {
final var userId = (String) session.getAttributes().get(CommonConstants.USER_ID);
final var userTypeStr = (String) session.getAttributes().get(CommonConstants.USER_TYPE);
final var userType = UserTypeEnum.fromString(userTypeStr);
if (userId != null && userType != null) {
return UserSessionKey.builder().userId(userId).userType(userType).build();
}
return null;
}
@Override
public void afterConnectionEstablished(WebSocketSession session) {
final var key = getUserSessionKey(session);
if (key != null) {
final var existingSession = userSessions.remove(key);
if (existingSession != null) {
try {
existingSession.closeSession();
log.info("WebSocket old connect closed，userId : {}, userType : {}", key.getUserId(), key.getUserType());
} catch (IOException e) {
log.error("old WebSocket close failed，userId : {}, userType : {}", key.getUserId(), key.getUserType(), e);
}
}
userSessions.put(key, new UserSessionData(session));
log.info("WebSocket connect success, userId : {}, userType : {}", key.getUserId(), key.getUserType());
} else {
log.warn("WebSocket connect failed，unable to get valid user information");
}
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
final var key = getUserSessionKey(session);
if (key != null) {
userSessions.remove(key);
log.info("WebSocket closed: userId : {}, userType : {}", key.getUserId(), key.getUserType());
} else {
log.warn("WebSocket closed failed，unable to get valid user information");
}
}
}