//package org.jeecg.modules.dry.websocket;
|
//
|
//import cn.hutool.core.bean.BeanUtil;
|
//import com.alibaba.fastjson.JSON;
|
//import com.alibaba.fastjson.JSONObject;
|
//import lombok.extern.slf4j.Slf4j;
|
//import org.jeecg.common.constant.DrySocketConst;
|
//import org.jeecg.common.util.SpringContextUtils;
|
//import org.jeecg.modules.dry.service.IDryRealTimeDataService;
|
//import org.jeecg.modules.dry.vo.RealTimeDataVo;
|
//import org.springframework.beans.factory.annotation.Autowired;
|
//import org.springframework.stereotype.Component;
|
//
|
//import javax.websocket.*;
|
//import javax.websocket.server.PathParam;
|
//import javax.websocket.server.ServerEndpoint;
|
//import java.util.Collection;
|
//import java.util.HashMap;
|
//import java.util.Map;
|
//
|
//@Slf4j
|
//@Component
|
//@ServerEndpoint("/drySocket/{tenantId}/{machineId}")
|
//public class DrySocket {
|
//
|
// /**
|
// * 当前 session
|
// */
|
// private Session session;
|
// /**
|
// * 当前租户id
|
// */
|
// private String tenantId;
|
// /**
|
// * 设备id，用于标识同一租户，不同设备的数据
|
// */
|
// private String machineId;
|
// /**
|
// * 当前socket唯一id
|
// */
|
// private String socketId;
|
//
|
// /**
|
// * 租户连接池，包含单个租户的所有socket连接；
|
// * 因为一个租户可能打开多个设备，多个设备就会有多个连接；
|
// * key是tenantId，value是Map对象；子Map的key是machineId，value是drySocket对象
|
// */
|
// private Map<String, Map<String, DrySocket>> tenantPool = new HashMap<>();
|
// /**
|
// * 连接池，包含所有WebSocket连接；
|
// * key是socketId，value是drySocket对象
|
// */
|
// private Map<String, DrySocket> machinePool = new HashMap<>();
|
//
|
// /**
|
// * 获取某个租户所有的设备
|
// */
|
// public Map<String, DrySocket> getTenantPool(String tenantId) {
|
// return tenantPool.computeIfAbsent(tenantId, k -> new HashMap<>(5));
|
// }
|
//
|
// /**
|
// * 向当前租户发送消息
|
// *
|
// * @param message 消息内容
|
// */
|
// public void sendMessage(String message) {
|
// try {
|
// this.session.getAsyncRemote().sendText(message);
|
// } catch (Exception e) {
|
// log.error("【drySocket】消息发送失败：" + e.getMessage());
|
// }
|
// }
|
//
|
// /**
|
// * 封装消息json
|
// *
|
// * @param data 消息内容
|
// */
|
// public static String packageMessage(String type, Object data) {
|
// JSONObject message = new JSONObject();
|
// message.put(DrySocketConst.TYPE, type);
|
// message.put(DrySocketConst.DATA, data);
|
// return message.toJSONString();
|
// }
|
//
|
// /**
|
// * 向指定租户的所有设备发送消息
|
// *
|
// * @param tenantId 接收消息的租户ID
|
// * @param message 消息内容
|
// */
|
// public void sendMessageTo(String tenantId, String message) {
|
// Collection<DrySocket> values = getTenantPool(tenantId).values();
|
// if (values.size() > 0) {
|
// for (DrySocket socketItem : values) {
|
// socketItem.sendMessage(message);
|
// }
|
// } else {
|
// log.warn("【drySocket】消息发送失败：tenantId\"" + tenantId + "\"不存在或未在线！");
|
// }
|
// }
|
//
|
// /**
|
// * 向指定租户的指定设备发送消息
|
// *
|
// * @param tenantId 接收消息的租户ID
|
// * @param message 消息内容
|
// */
|
// public void sendMessageTo(String tenantId, String machineId, String message) {
|
// DrySocket socketItem = getTenantPool(tenantId).get(machineId);
|
// if (socketItem != null) {
|
// socketItem.sendMessage(message);
|
// } else {
|
// log.warn("【drySocket】消息发送失败：tenantId\"" + tenantId + "\"的machineId\"" + machineId + "\"不存在或未在线！");
|
// }
|
// }
|
//
|
// /**
|
// * 向多个租户的所有设备发送消息
|
// *
|
// * @param tenantIds 接收消息的租户ID数组
|
// * @param message 消息内容
|
// */
|
// public void sendMessageTo(String[] tenantIds, String message) {
|
// for (String tenantId : tenantIds) {
|
// this.sendMessageTo(tenantId, message);
|
// }
|
// }
|
//
|
// /**
|
// * 向所有租户的所有设备发送消息
|
// *
|
// * @param message 消息内容
|
// */
|
// public void sendMessageToAll(String message) {
|
// for (DrySocket socketItem : machinePool.values()) {
|
// socketItem.sendMessage(message);
|
// }
|
// }
|
//
|
// /**
|
// * websocket 开启连接
|
// */
|
// @OnOpen
|
// public void onOpen(Session session, @PathParam("tenantId") String tenantId, @PathParam("machineId") String machineId) {
|
// try {
|
// this.tenantId = tenantId;
|
// this.machineId = machineId;
|
// this.socketId = tenantId + machineId;
|
// this.session = session;
|
//
|
// machinePool.put(this.socketId, this);
|
// getTenantPool(tenantId).put(this.machineId, this);
|
//
|
// log.info("【drySocket】有新的连接，总数为:" + machinePool.size());
|
// log.info("ssss"+tenantPool.size());
|
// } catch (Exception ignored) {
|
// }
|
// }
|
//
|
// /**
|
// * websocket 断开连接
|
// */
|
// @OnClose
|
// public void onClose() {
|
// try {
|
// machinePool.remove(this.socketId);
|
// getTenantPool(this.tenantId).remove(this.machineId);
|
//
|
// log.info("【drySocket】连接断开，总数为:" + machinePool.size());
|
// } catch (Exception ignored) {
|
// }
|
// }
|
//
|
// /**
|
// * websocket 收到消息
|
// */
|
// @OnMessage
|
// public void onMessage(String message) {
|
// log.info("【drySocket】onMessage:" + message);
|
// IDryRealTimeDataService realTimeDataService = SpringContextUtils.getBean(IDryRealTimeDataService.class);
|
// JSONObject json;
|
// try {
|
// json = JSON.parseObject(message);
|
// } catch (Exception e) {
|
// log.warn("【drySocket】收到不合法的消息:" + message);
|
// return;
|
// }
|
// String type = json.getString(DrySocketConst.TYPE);
|
// switch (type) {
|
// // 心跳检测
|
// case DrySocketConst.TYPE_HB:
|
// this.sendMessage(DrySocket.packageMessage(type, true));
|
// break;
|
// // 实时数据处理
|
// case DrySocketConst.TYPE_RDT:
|
// Object o = json.get(DrySocketConst.DATA);
|
// RealTimeDataVo realTimeDataVo = BeanUtil.toBean(o, RealTimeDataVo.class);
|
// realTimeDataService.realTimeDataHandle(realTimeDataVo);
|
// break;
|
//
|
// default:
|
// log.warn("【drySocket】收到不识别的消息类型:" + type);
|
// break;
|
// }
|
//
|
//
|
// }
|
//
|
// /** * 配置错误信息处理 * @param session * @param t */
|
// @OnError
|
// public void onError(Session session, Throwable t) {
|
//
|
////什么都不想打印都去掉就好了
|
// log.info("【websocket消息】出现未知错误 ");
|
////打印错误信息，如果你不想打印错误信息，去掉就好了
|
////这里打印的也是 java.io.EOFException: null
|
// t.printStackTrace();
|
// }
|
//
|
//
|
//}
|
