package com.ard.sample.component.websocket.model;
|
|
import com.ard.sample.component.websocket.config.ConcurrentWebSocketSession;
|
import com.ard.sample.manage.model.enums.UserTypeEnum;
|
import lombok.extern.slf4j.Slf4j;
|
import java.io.IOException;
|
import java.util.*;
|
import java.util.concurrent.ConcurrentHashMap;
|
|
/**
|
* Manage all WebSocket connection objects.
|
* @author sean.zhou
|
* @date 2021/11/16
|
* @version 0.1
|
*/
|
@Slf4j
|
public class WebSocketManager {
|
|
// three layers workspace,user,id
|
private static final ConcurrentHashMap<String,
|
ConcurrentHashMap<String,
|
ConcurrentHashMap<String, ConcurrentWebSocketSession>>> MANAGER =
|
new ConcurrentHashMap<>(16);
|
|
// 存储发送命令的客户端
|
public static final ConcurrentHashMap<String, UserDeviceSession> CMD_SESSION
|
= new ConcurrentHashMap<>(16);
|
|
/**
|
* WebSocket connection from the pilot.
|
*/
|
private static final Set<ConcurrentWebSocketSession> PILOT_SESSION
|
= ConcurrentHashMap.newKeySet(16);
|
|
/**
|
* WebSocket connection from the web.
|
*/
|
private static final Set<ConcurrentWebSocketSession> WEB_SESSION
|
= ConcurrentHashMap.newKeySet(16);
|
|
public static void closeWsConnectByUserId(String userId,String workspaceId,String userType){
|
|
//首先关闭命令socket连接
|
closeCmdConnectByUserId(userId);
|
|
//然后关闭广播socket连接
|
closeDataConnectByUserId(userId,workspaceId,userType);
|
|
}
|
|
private static void closeCmdConnectByUserId(String userId){
|
//一个用户只能建立一个命令socket连接，则把之前的连接关闭，重新添加新的连接
|
if(CMD_SESSION.containsKey(userId)){
|
//关闭之前的连接
|
try {
|
CMD_SESSION.get(userId).session.close();
|
CMD_SESSION.remove(userId);
|
} catch (IOException e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
|
private static void closeDataConnectByUserId(String userId,String workspaceId,String userType){
|
//{workspaceId}/{userType}/{userId}/{userName}
|
if(MANAGER.containsKey(workspaceId)){
|
//根据部门查询
|
ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentWebSocketSession>> workspaceSessions =
|
MANAGER.get(workspaceId);
|
if(workspaceSessions.containsKey(userId)){
|
//根据用户查询
|
ConcurrentHashMap<String, ConcurrentWebSocketSession> userSessions =
|
workspaceSessions.get(userId);
|
//判断广播连接数量
|
if(userSessions.size()>0){
|
//关闭先前的连接
|
for(Map.Entry<String,ConcurrentWebSocketSession> userSession:userSessions.entrySet()){
|
ConcurrentWebSocketSession session = userSession.getValue();
|
//从用户连接中删除
|
Set<ConcurrentWebSocketSession> typeSession = getSetByUserType(Integer.valueOf(userType));
|
typeSession.remove(session);
|
userSessions.remove(userSession.getKey());
|
//关闭连接
|
try {
|
session.close();
|
} catch (IOException e) {
|
e.printStackTrace();
|
}
|
//一个用户最多有一个广播连接
|
break;
|
}
|
}
|
}
|
}
|
}
|
|
public static void put(String key, ConcurrentWebSocketSession val) {
|
|
String[] name = key.split("/");
|
if (name.length != 4) {
|
log.debug("The key is out of format. [{workspaceId}/{userType}/{userId}/{userName}]");
|
return;
|
}
|
|
// 根据连接的地址，添加到命令的session中
|
if(val.getUri().getPath().endsWith("cmd"))
|
{
|
//一个用户只能建立一个命令socket连接，则把之前的连接关闭，重新添加新的连接
|
closeCmdConnectByUserId(name[2]);
|
// 重新添加新连接,然后返回即可
|
val.setBinaryMessageSizeLimit(1024*1024);
|
val.setTextMessageSizeLimit(1024*1024);
|
CMD_SESSION.put(name[2],UserDeviceSession.builder().tyBoxSn("")
|
.session(val).userName(name[3]).build());
|
|
//更新用户缓存中的命令ws状态
|
// LoginUserManager.setCmdWsStatus(name[2],true);
|
|
return;
|
}
|
|
//同用户建立新连接，之前的连接删除
|
closeDataConnectByUserId(name[2],name[0],name[1]);
|
|
//执行新的添加操作
|
ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentWebSocketSession>> workspaceSessions =
|
MANAGER.getOrDefault(name[0], new ConcurrentHashMap<>(16));
|
|
ConcurrentHashMap<String, ConcurrentWebSocketSession> userSessions = workspaceSessions.getOrDefault(
|
name[2], new ConcurrentHashMap<>(16));
|
userSessions.put(val.getId(), val);
|
workspaceSessions.put(name[2], userSessions);
|
MANAGER.put(name[0], workspaceSessions);
|
|
//根据用户类型，区分是web还是pilot
|
getSetByUserType(Integer.valueOf(name[1])).add(val);
|
|
//更新用户缓存中的数据ws状态
|
// LoginUserManager.setDataWsStatus(name[2],true);
|
}
|
|
public static String remove(String key, String sessionId, String path) {
|
String[] name = key.split("/");
|
if (name.length != 4) {
|
log.debug("The key is out of format. [{workspaceId}/{userType}/{userId}/{userName}]");
|
return "";
|
}
|
|
// //先从登录列表中把该用户删除
|
// LoginUserManager.remove(name[2]);
|
|
//若为命令连接，直接删除后返回，无须后面的删除操作
|
if(path.endsWith("cmd")){
|
String boxSn = "";
|
if(CMD_SESSION.containsKey(name[2]))
|
{
|
//先获取当前的sn
|
boxSn = CMD_SESSION.get(name[2]).getTyBoxSn();
|
//删掉该连接
|
CMD_SESSION.remove(name[2]);
|
//删除控制的设备相关信息
|
}
|
//更新用户缓存中的数据ws状态
|
// LoginUserManager.setCmdWsStatus(name[2],false);
|
return boxSn;
|
}
|
|
ConcurrentHashMap<String, ConcurrentWebSocketSession> userSession = MANAGER.get(name[0]).get(name[2]);
|
|
Set<ConcurrentWebSocketSession> typeSession = getSetByUserType(Integer.valueOf(name[1]));
|
|
ConcurrentWebSocketSession session = userSession.get(sessionId);
|
if(session!=null){
|
typeSession.remove(session);
|
userSession.remove(sessionId);
|
}
|
//更新用户缓存中的数据ws状态
|
//LoginUserManager.setDataWsStatus(name[2],false);
|
return "";
|
}
|
|
public static ConcurrentWebSocketSession getConnectionByUser(String userId){
|
if(CMD_SESSION.containsKey(userId)){
|
return CMD_SESSION.get(userId).getSession();
|
}
|
return null;
|
}
|
|
public static int getConnectedCount() {
|
//return PILOT_SESSION.size() + WEB_SESSION.size();
|
return CMD_SESSION.size() + WEB_SESSION.size();
|
}
|
|
public static Collection<ConcurrentWebSocketSession> getValueWithWorkspace(String workspaceId) {
|
// 根据workspaceID取出每一个session
|
Set<ConcurrentWebSocketSession> sessions = ConcurrentHashMap.newKeySet();
|
|
//MANAGER有可能为0
|
MANAGER.getOrDefault(workspaceId,new ConcurrentHashMap<>())
|
.forEach((userId, userSessions) -> {
|
sessions.addAll(userSessions.values());
|
});
|
|
return sessions;
|
}
|
|
//查询到所有的session
|
public static Collection<ConcurrentWebSocketSession> getValueAllWorkspace() {
|
// 根据workspaceID取出每一个session
|
Set<ConcurrentWebSocketSession> sessions = ConcurrentHashMap.newKeySet();
|
|
MANAGER.forEach((wkId,workspaceSessions)->{
|
workspaceSessions.forEach((userId, userSessions) -> {
|
sessions.addAll(userSessions.values());
|
});
|
});
|
|
return sessions;
|
}
|
|
public static Collection<ConcurrentWebSocketSession> getValueWithWorkspaceAndUserType(String workspaceId, Integer userType) {
|
Set<ConcurrentWebSocketSession> sessions = ConcurrentHashMap.newKeySet();
|
Set<ConcurrentWebSocketSession> typeSessions = getSetByUserType(userType);
|
|
MANAGER.getOrDefault(workspaceId, new ConcurrentHashMap<>())
|
.forEach((userId, userSessions) -> {
|
Collection<ConcurrentWebSocketSession> sessionList = userSessions.values();
|
if (!sessionList.isEmpty()) {
|
ConcurrentWebSocketSession session = sessionList.iterator().next();
|
if (typeSessions.contains(session)) {
|
sessions.addAll(sessionList);
|
}
|
}
|
});
|
return sessions;
|
}
|
|
public static Collection<ConcurrentWebSocketSession> getValueWithAncestorWorkspaceAndUserType(
|
List<String> workspaceIds, Integer userType) {
|
Set<ConcurrentWebSocketSession> sessions = ConcurrentHashMap.newKeySet();
|
Set<ConcurrentWebSocketSession> typeSessions = getSetByUserType(userType);
|
|
workspaceIds.forEach(workspaceId->{
|
MANAGER.getOrDefault(workspaceId, new ConcurrentHashMap<>())
|
.forEach((userId, userSessions) -> {
|
Collection<ConcurrentWebSocketSession> sessionList = userSessions.values();
|
if (!sessionList.isEmpty()) {
|
ConcurrentWebSocketSession session = sessionList.iterator().next();
|
if (typeSessions.contains(session)) {
|
sessions.addAll(sessionList);
|
}
|
}
|
});
|
});
|
return sessions;
|
}
|
|
public static Set<ConcurrentWebSocketSession> getSetByUserType(Integer userType) {
|
if (UserTypeEnum.PILOT.getVal() == userType) {
|
return PILOT_SESSION;
|
}
|
|
if (UserTypeEnum.WEB.getVal() == userType) {
|
return WEB_SESSION;
|
}
|
return new HashSet<>();
|
}
|
}
|
