package com.fzzy.gateway.hx2023.websocket;
|
|
import com.alibaba.fastjson.JSONObject;
|
import com.fzzy.gateway.GatewayUtils;
|
import com.fzzy.gateway.hx2023.data.WebSocketPacket;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.lang.StringUtils;
|
import org.springframework.stereotype.Component;
|
|
import javax.websocket.*;
|
import javax.websocket.server.PathParam;
|
import javax.websocket.server.ServerEndpoint;
|
import java.util.Map;
|
import java.util.concurrent.ConcurrentHashMap;
|
|
/**
|
* 基于前端 websocket 订阅地磅数据 topic:
|
*/
|
@Slf4j
|
@Component
|
@ServerEndpoint(value = "/device/{productId}/{deviceId}/message/property/report")
|
public class WebSocketDeviceReport {
|
|
private static Map<String, Session> sessionPool = new ConcurrentHashMap<>();
|
private static Map<String, String> sessionIds = new ConcurrentHashMap<>();
|
|
// 与某个客户端的连接会话，需要通过它来给客户端发送数据
|
private Session session;
|
|
|
@OnOpen
|
public void onOpen(Session session,
|
@PathParam("productId") String productId,
|
@PathParam("deviceId") String deviceId,
|
@PathParam("clientId") String clientId
|
) throws Exception {
|
|
this.session = session;
|
|
String key = productId + "-" + deviceId;
|
|
sessionPool.put(key, session);
|
sessionIds.put(session.getId(), key);
|
|
GatewayUtils.updateOnline(deviceId);
|
|
log.info("new webSocket,clientId={}", key);
|
}
|
|
@OnClose
|
public void onClose() {
|
|
String key = sessionIds.get(session.getId());
|
|
String deviceId = key.substring(key.indexOf("-"));
|
|
GatewayUtils.updateOffOnline(deviceId);
|
|
sessionPool.remove(key);
|
sessionIds.remove(session.getId());
|
|
log.info("WebSocket连接关闭={}", key);
|
|
}
|
|
/**
|
* 收到前端发送的信息
|
*
|
* @param message
|
* @param session
|
*/
|
@OnMessage
|
public void onMessage(String message, Session session) {
|
|
log.info("来自客户端信息:\n" + message);
|
}
|
|
@OnError
|
public void onError(Session session, Throwable error) {
|
log.error("发生错误");
|
|
String key = sessionIds.get(session.getId());
|
|
String deviceId = key.substring(key.indexOf("-"));
|
|
GatewayUtils.updateOffOnline(deviceId);
|
|
sessionPool.remove(key);
|
sessionIds.remove(session.getId());
|
error.printStackTrace();
|
}
|
|
|
/**
|
* 推送信息到前端
|
*
|
* @param packet
|
*/
|
public void sendByPacket(WebSocketPacket packet) {
|
if (StringUtils.isEmpty(packet.getDeviceId())) {
|
log.error("WebSocket信息推送失败，设备编码为空。");
|
return;
|
}
|
|
String tag = packet.getDeviceId();
|
|
// 遍历推送
|
Session session;
|
String productId;
|
for (String key : sessionPool.keySet()) {
|
|
productId = key.substring(0, key.indexOf("-"));
|
|
packet.getHeaders().setProductId(productId);
|
|
log.debug("----------websocket返回信息-----{}", packet);
|
|
if (key.indexOf(tag) != -1) {
|
session = sessionPool.get(key);
|
session.getAsyncRemote().sendText(JSONObject.toJSONString(packet));
|
}
|
}
|
}
|
|
|
}
|
