package com.fastbee.mqtt.manager;
|
|
import com.fastbee.common.constant.FastBeeConstant;
|
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
|
import com.fastbee.common.utils.DateUtils;
|
import com.fastbee.common.utils.StringUtils;
|
import com.fastbee.base.session.Session;
|
import io.netty.handler.codec.mqtt.MqttPublishMessage;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.util.CollectionUtils;
|
|
import java.util.HashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.concurrent.ConcurrentHashMap;
|
|
/**
|
* 客户端管理
|
*
|
* @author gsb
|
* @date 2022/9/15 16:02
|
*/
|
@Slf4j
|
public class ClientManager {
|
|
/*topic本地缓存*/
|
public static Map<String, Map<String, Session>> topicMap = new ConcurrentHashMap<>();
|
/*客户端最后一次ping时间，设备不正常断开判断*/
|
private static Map<String, Long> pingMap = new ConcurrentHashMap<>();
|
/*客户端与topic关联本地缓存*/
|
public static Map<String, Map<String, Boolean>> clientTopicMap = new ConcurrentHashMap<>();
|
|
/**
|
* 将client的上下文相关信息添加到映射关系表中
|
* @param topic 主题
|
* @param session
|
*/
|
public static void push(String topic, Session session) {
|
try {
|
/*处理topic对应的topic*/
|
Map<String, Session> clientMap = topicMap.get(topic);
|
if (StringUtils.isEmpty(clientMap)) {
|
clientMap = new ConcurrentHashMap<>();
|
}
|
clientMap.put(session.getClientId(), session);
|
topicMap.put(topic, clientMap);
|
|
/*处理client对应的所有topic*/
|
Map<String, Boolean> topicsMap = null;
|
if (clientTopicMap.containsKey(session.getClientId())) {
|
topicsMap = clientTopicMap.get(session.getClientId());
|
if (!topicsMap.containsKey(topic)) {
|
topicsMap.put(topic, true);
|
}
|
} else {
|
topicsMap = new HashMap<>();
|
topicsMap.put(topic, true);
|
clientTopicMap.put(session.getClientId(), topicsMap);
|
}
|
} catch (Exception e) {
|
log.error("=>clientId映射topic出现异常:{},e=", e.getMessage(), e);
|
}
|
}
|
|
/**
|
* 清理对应client下的所有数据
|
*
|
* @param clientId 客户端id
|
*/
|
public static void remove(String clientId) {
|
try {
|
/*移除client对应的topic*/
|
Map<String, Boolean> topics = clientTopicMap.get(clientId);
|
if (null != topics) {
|
/*从topic中移除client*/
|
for (String key : topics.keySet()) {
|
Map<String, Session> clientMap = topicMap.get(key);
|
if (CollectionUtils.isEmpty(clientMap)) {
|
continue;
|
}
|
clientMap.remove(clientId);
|
}
|
clientTopicMap.remove(clientId);
|
}
|
pingMap.remove(clientId);
|
} catch (Exception e) {
|
log.warn("=>移除client[{}]异常", e.getMessage());
|
}
|
}
|
|
/**
|
* 客户端取消订阅
|
* 删除指定topic下的指定client
|
*
|
* @param topic 主题
|
* @param session 客户端
|
*/
|
public static void unsubscribe(String topic, Session session) {
|
try {
|
Map<String, Session> clientMap = topicMap.get(topic);
|
if (StringUtils.isEmpty(clientMap)) {
|
return;
|
}
|
Session s = clientMap.get(session.getClientId());
|
if (null == s) {
|
return;
|
}
|
clientMap.remove(session.getClientId());
|
} catch (Exception e) {
|
log.error("=>客户端取消订阅异常:{}", e.getMessage());
|
}
|
}
|
|
/**
|
* 将消息发送到指定topic下的所有client上去
|
*
|
* @param msg 推送消息
|
*/
|
public static void pubTopic(MqttPublishMessage msg) {
|
String topic = msg.variableHeader().topicName();
|
List<String> topicList = TopicsUtils.searchTopic(topic);
|
for (String itemTopic : topicList) {
|
Map<String, Session> clientMap = topicMap.get(itemTopic);
|
if (StringUtils.isEmpty(clientMap)) {
|
continue;
|
}
|
for (Session session : clientMap.values()) {
|
String clientId = session.getClientId();
|
if (!validClient(clientId)) {
|
///*ws的客户端不正常断开连接后，直接移除所有信息*/
|
//if (session.getClientId().startsWith(FastBeeConstant.SERVER.WS_PREFIX)) {
|
// log.debug("=>移除ws客户端,clientId={}", session);
|
// remove(clientId);
|
//}
|
log.warn("=>{}不在线", clientId);
|
continue;
|
}
|
ResponseManager.publishClients(msg, session);
|
}
|
}
|
}
|
|
/**
|
* 更新客户端在线时间，给客户端发送消息时用这个看客户端最近是否在线
|
* 用来判断设备不正常掉线没有应答服务器的情况
|
*
|
* @param clientId 客户端id
|
*/
|
public static void updatePing(String clientId) {
|
pingMap.put(clientId, DateUtils.getTimestamp());
|
}
|
|
/**
|
* 平台判定设备状态 Ping客户端是否在线
|
*
|
* @param clientId 客户端id
|
* @return 结果
|
*/
|
public static Boolean validClient(String clientId) {
|
long currTime = DateUtils.getTimestamp();
|
/*获取客户端连接时，时间*/
|
Long timestamp = pingMap.get(clientId);
|
if (null == timestamp) {
|
return false;
|
}
|
//当设备缓存的心跳时间大于 平台判断时间 1.5f 表示设备不正常断开了服务器
|
if (currTime - timestamp > FastBeeConstant.SERVER.DEVICE_PING_EXPIRED) {
|
//pingMap.remove(clientId);
|
//SessionManger.removeClient(clientId);
|
return false;
|
}
|
return true;
|
}
|
}
|
