430 lines
16 KiB
Java
430 lines
16 KiB
Java
package com.maibu.mqtt;
|
||
|
||
import com.fasterxml.jackson.databind.JsonNode;
|
||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||
import com.maibu.constant.Constant;
|
||
import com.maibu.core.business.device.DeviceStatusDetail;
|
||
import com.maibu.core.business.device.NettyDevice;
|
||
import com.maibu.core.business.dto.WebStatusMessageDTO;
|
||
import com.maibu.core.business.inter.WebsocketMesDispather;
|
||
import com.maibu.core.enums.MesType;
|
||
import com.maibu.core.host.*;
|
||
import com.maibu.memory.DeviceSessionManager;
|
||
import com.maibu.memory.GlobalMemory;
|
||
import com.maibu.utils.StringUtils;
|
||
import com.maibu.utils.json.JsonUtils;
|
||
import com.maibu.utils.spring.SpringUtils;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.util.CollectionUtils;
|
||
|
||
import java.util.ArrayList;
|
||
import java.util.List;
|
||
|
||
|
||
/**
|
||
* mqtt相关的 上位机处理器
|
||
*/
|
||
@Slf4j
|
||
public class HostMessageHandler implements CustomMqttMessageHandler {
|
||
|
||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||
|
||
private final DeviceSessionManager deviceSessionManager = SpringUtils.getBean(DeviceSessionManager.class);
|
||
|
||
private final WebsocketMesDispather websocketMesDispather = SpringUtils.getBean(WebsocketMesDispather.class);
|
||
|
||
|
||
private final String mqttClientId = "maibu-mqtt-client";
|
||
|
||
@Override
|
||
public String getHandlerType() {
|
||
return Constant.mqttHandleKey;
|
||
}
|
||
|
||
@Override
|
||
public boolean supports(String handlerType) {
|
||
return Constant.mqttHandleKey.equals(handlerType);
|
||
}
|
||
|
||
/**
|
||
* 处理 MQTT 消息
|
||
*
|
||
* <p>通配符订阅场景下，deviceKey 为配置中的固定标识（如 "host-monitor"），
|
||
* 真正的设备 ID 需从 topic 中解析，通过 {@link #extractDeviceId(String)} 获取。</p>
|
||
*/
|
||
@Override
|
||
public void handleMessage(String deviceKey, String topic, String payload, CustomMqttDeviceConfig config) {
|
||
try {
|
||
String deviceId = extractDeviceId(topic);
|
||
if (deviceId == null) {
|
||
log.warn("无法从topic解析deviceId: {}", topic);
|
||
return;
|
||
}
|
||
|
||
String action = extractAction(topic);
|
||
if (action == null) {
|
||
log.debug("未识别的topic: {}", topic);
|
||
return;
|
||
}
|
||
|
||
switch (action) {
|
||
case "location_post":
|
||
handleLocationPost(deviceId, topic, payload);
|
||
break;
|
||
case "realtime_post":
|
||
handleRealtimePost(deviceId, topic, payload);
|
||
break;
|
||
case "heartbeat_post":
|
||
handleHeartbeatPost(deviceId, topic, payload);
|
||
break;
|
||
case "error_post":
|
||
handleErrorPost(deviceId, topic, payload);
|
||
break;
|
||
// case "route_post":
|
||
// case "point_post":
|
||
// handleTaskPost(deviceId, topic, payload, action);
|
||
// break;
|
||
case "task_status_post":
|
||
handleTaskPost(deviceId, topic, payload, action);
|
||
break;
|
||
case "config_reply":
|
||
handleConfigReply(deviceId, topic, payload);
|
||
break;
|
||
case "control_reply":
|
||
handleControlReply(deviceId, topic, payload);
|
||
break;
|
||
case "route_reply":
|
||
handleRouteReply(deviceId, topic, payload);
|
||
break;
|
||
default:
|
||
log.debug("割草机消息: deviceId={}, topic={}, action={}", deviceId, topic, action);
|
||
break;
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("消息处理异常: topic={}, error={}", topic, e.getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 定位信息 经纬度
|
||
*/
|
||
private void handleLocationPost(String deviceId, String topic, String payload) {
|
||
try {
|
||
log.debug("【位置数据】deviceId={}, payload={}", deviceId, payload);
|
||
HostLocationRelTimeDTO locationRelTimeDTO = JsonUtils.parseObject(payload, HostLocationRelTimeDTO.class);
|
||
if (locationRelTimeDTO != null) {
|
||
WebStatusMessageDTO dto = createWebDeviceLocationMessage(locationRelTimeDTO);
|
||
List<NettyDevice> controlMasters = deviceSessionManager
|
||
.getAllSlaveControl(deviceId);
|
||
if (!CollectionUtils.isEmpty(controlMasters)) {
|
||
for (NettyDevice x : controlMasters) {
|
||
try {
|
||
websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(dto));
|
||
} catch (Exception e) {
|
||
log.error("位置数据websocket推送失败: connectorId={}, error={}", x.getConnectorId(), e.getMessage());
|
||
}
|
||
}
|
||
}
|
||
//todo 推送实时位置和状态消息到外部
|
||
//todo 记录到内存 开线程池队列 逐一转发
|
||
// GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_LOCATION_TOPIC, deviceId),
|
||
// JsonUtils.toJsonString(dto));
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("位置消息解析失败: deviceId={}, payload={}, error={}", deviceId, payload, e.getMessage());
|
||
}
|
||
}
|
||
|
||
|
||
public WebStatusMessageDTO createWebDeviceLocationMessage(HostLocationRelTimeDTO locationRelTimeDTO) {
|
||
WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO();
|
||
List<DeviceStatusDetail> transferData = new ArrayList<>();
|
||
DeviceStatusDetail lat = new DeviceStatusDetail();
|
||
lat.setName("latitude");
|
||
lat.setValue(String.valueOf(locationRelTimeDTO.getData().getLatitude()));
|
||
lat.setUnit("");
|
||
|
||
transferData.add(lat);
|
||
|
||
DeviceStatusDetail lng = new DeviceStatusDetail();
|
||
lng.setName("longitude");
|
||
lng.setValue(String.valueOf(locationRelTimeDTO.getData().getLongitude()));
|
||
lng.setUnit("");
|
||
transferData.add(lng);
|
||
|
||
webStatusMessageDTO.setData(transferData);
|
||
webStatusMessageDTO.setType(MesType.locationInfo);
|
||
return webStatusMessageDTO;
|
||
}
|
||
|
||
public WebStatusMessageDTO createWebDeviceRealtimeMessage(HostVehicleRelTimeDTO vehicleRelTimeDTO) {
|
||
//todo 状态消息
|
||
|
||
WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO();
|
||
List<DeviceStatusDetail> transferData = new ArrayList<>();
|
||
DeviceStatusDetail voltage = new DeviceStatusDetail();
|
||
voltage.setName("voltage");
|
||
voltage.setValue(String.valueOf(vehicleRelTimeDTO.getData().getBattery_voltage()));
|
||
voltage.setUnit("");
|
||
|
||
transferData.add(voltage);
|
||
|
||
DeviceStatusDetail speed = new DeviceStatusDetail();
|
||
speed.setName("speed");
|
||
speed.setValue(String.valueOf(vehicleRelTimeDTO.getData().getSpeed()));
|
||
speed.setUnit("");
|
||
transferData.add(speed);
|
||
|
||
DeviceStatusDetail motion = new DeviceStatusDetail();
|
||
motion.setName("motion");
|
||
motion.setValue(String.valueOf(vehicleRelTimeDTO.getData().getMotion()));
|
||
motion.setUnit("");
|
||
transferData.add(motion);
|
||
|
||
DeviceStatusDetail percentage = new DeviceStatusDetail();
|
||
percentage.setName("percentage");
|
||
percentage.setValue(String.valueOf(vehicleRelTimeDTO.getData().getBattery_percentage()));
|
||
percentage.setUnit("");
|
||
transferData.add(percentage);
|
||
|
||
webStatusMessageDTO.setData(transferData);
|
||
webStatusMessageDTO.setType(MesType.deviceInfo);
|
||
return webStatusMessageDTO;
|
||
}
|
||
|
||
/**
|
||
* 上位机数据
|
||
*/
|
||
private void handleRealtimePost(String deviceId, String topic, String payload) {
|
||
log.debug("【实时数据】deviceId={}, payload={}", deviceId, payload);
|
||
try {
|
||
HostVehicleRelTimeDTO vehicleRelTimeDTO = JsonUtils.parseObject(payload, HostVehicleRelTimeDTO.class);
|
||
if (vehicleRelTimeDTO != null) {
|
||
WebStatusMessageDTO dto = createWebDeviceRealtimeMessage(vehicleRelTimeDTO);
|
||
List<NettyDevice> controlMasters = deviceSessionManager
|
||
.getAllSlaveControl(deviceId);
|
||
if (!CollectionUtils.isEmpty(controlMasters)) {
|
||
for (NettyDevice x : controlMasters) {
|
||
try {
|
||
websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(dto));
|
||
} catch (Exception e) {
|
||
log.error("实时数据websocket推送失败: connectorId={}, error={}", x.getConnectorId(), e.getMessage());
|
||
}
|
||
}
|
||
}
|
||
// GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_STATUS_TOPIC, deviceId),
|
||
// JsonUtils.toJsonString(dto));
|
||
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("实时消息解析失败: deviceId={}, payload={}, error={}", deviceId, payload, e.getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理状态变更事件
|
||
* topic: mower/{deviceId}/event/status/post
|
||
*/
|
||
private void handleStatusPost(String deviceId, String topic, String payload) {
|
||
log.info("【状态变更】deviceId={}, payload={}", deviceId, payload);
|
||
}
|
||
|
||
/**
|
||
* 处理心跳
|
||
* topic: mower/{deviceId}/event/heartbeat/post
|
||
*/
|
||
private void handleHeartbeatPost(String deviceId, String topic, String payload) {
|
||
log.debug("【心跳】payload={}", payload);
|
||
// 心跳处理
|
||
HeartBeatDTO beatDTO = JsonUtils.parseObject(payload, HeartBeatDTO.class);
|
||
if (beatDTO != null) {
|
||
beatDTO.setSysTimestamp(System.currentTimeMillis());
|
||
GlobalMemory.customMqttDeviceMonitor.saveHeartBeat(beatDTO.getSn(), beatDTO);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理错误告警
|
||
* topic: mower/{deviceId}/event/error/post
|
||
*/
|
||
private void handleErrorPost(String deviceId, String topic, String payload) {
|
||
log.warn("【错误告警】deviceId={}, payload={}", deviceId, payload);
|
||
}
|
||
|
||
/**
|
||
* 处理任务相关上报
|
||
*/
|
||
private void handleTaskPost(String deviceId, String topic, String payload, String action) {
|
||
log.info("【任务上报】deviceId={}, action={}, payload={}", deviceId, action, payload);
|
||
|
||
|
||
HostTaskStatusDTO dto = JsonUtils.parseObject(payload, HostTaskStatusDTO.class);
|
||
if (dto != null && !StringUtils.isEmpty(dto.getStatus())) {
|
||
NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
if ("completed".equals(dto.getStatus())) {
|
||
nettyDevice.finishTask();
|
||
}
|
||
if ("cancelled".equals(dto.getStatus())) {
|
||
nettyDevice.canceledTask();
|
||
}
|
||
// if("paused".equals(dto.getStatus())){
|
||
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
// nettyDevice.finishTask();
|
||
// }
|
||
// if("failed".equals(dto.getStatus())){
|
||
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
// nettyDevice.finishTask();
|
||
// }
|
||
// if("accepted".equals(dto.getStatus())){
|
||
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
// nettyDevice.finishTask();
|
||
// }
|
||
// if("running".equals(dto.getStatus())){
|
||
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
// nettyDevice.finishTask();
|
||
// }
|
||
}
|
||
|
||
|
||
}
|
||
|
||
/**
|
||
* 处理配置响应
|
||
*/
|
||
private void handleConfigReply(String deviceId, String topic, String payload) {
|
||
log.debug("【配置响应】deviceId={}, payload={}", deviceId, payload);
|
||
}
|
||
|
||
/**
|
||
* 处理控制指令响应
|
||
*/
|
||
private void handleControlReply(String deviceId, String topic, String payload) {
|
||
log.debug("【控制响应】deviceId={}, payload={}", deviceId, payload);
|
||
}
|
||
|
||
|
||
/**
|
||
* 处理控制指令响应
|
||
*/
|
||
private void handleRouteReply(String deviceId, String topic, String payload) {
|
||
log.debug("【路径任务下发响应】deviceId={}, payload={}", deviceId, payload);
|
||
|
||
//todo 更新任务接收状态
|
||
HostNavigationReplyDTO replyDTO = JsonUtils.parseObject(payload, HostNavigationReplyDTO.class);
|
||
NettyDevice device = deviceSessionManager.getDevice(deviceId);
|
||
if (replyDTO.isAccepted()) {
|
||
if (device != null && device.getTask() != null && String.valueOf(device.getTask().getId()).equals(replyDTO.getTask_id())) {
|
||
device.getTask().setTaskAccepted(true);
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
/**
|
||
* 从 Topic 中解析 deviceId
|
||
* Topic 格式：mower/{deviceId}/...
|
||
*/
|
||
private String extractDeviceId(String topic) {
|
||
if (topic == null || topic.isEmpty()) {
|
||
return null;
|
||
}
|
||
String[] parts = topic.split("/");
|
||
if (parts.length < 2 || !"mower".equals(parts[0])) {
|
||
return null;
|
||
}
|
||
return parts[1];
|
||
}
|
||
|
||
/**
|
||
* 从 Topic 中提取动作标识
|
||
* 将 mower/{deviceId}/xxx/yyy/post 转为 xxx_yyy_post 形式
|
||
*/
|
||
private String extractAction(String topic) {
|
||
String[] parts = topic.split("/");
|
||
if (parts.length < 3 || !"mower".equals(parts[0])) {
|
||
return null;
|
||
}
|
||
StringBuilder path = new StringBuilder();
|
||
for (int i = 2; i < parts.length; i++) {
|
||
if (i > 2) path.append("/");
|
||
path.append(parts[i]);
|
||
}
|
||
String remaining = path.toString();
|
||
|
||
if (remaining.equals("property/location/post")) return "location_post";
|
||
if (remaining.equals("property/realtime/post")) return "realtime_post";
|
||
if (remaining.equals("event/heartbeat/post")) return "heartbeat_post";
|
||
if (remaining.equals("event/error/post")) return "error_post";
|
||
// if (remaining.equals("task/route/post")) return "route_post";
|
||
// if (remaining.equals("task/point/post")) return "point_post";
|
||
if (remaining.equals("task/status/post")) return "task_status_post";
|
||
if (remaining.equals("action/config/reply")) return "config_reply";
|
||
if (remaining.equals("action/control/reply")) return "control_reply";
|
||
if (remaining.equals("task/target/reply")) return "route_reply";
|
||
|
||
return null;
|
||
}
|
||
|
||
private double getDouble(JsonNode node, String... names) {
|
||
for (String name : names) {
|
||
JsonNode fn = node.get(name);
|
||
if (fn != null && !fn.isNull()) {
|
||
if (fn.isNumber()) return fn.asDouble();
|
||
try {
|
||
return Double.parseDouble(fn.asText());
|
||
} catch (Exception ignored) {
|
||
}
|
||
}
|
||
}
|
||
return -1.0;
|
||
}
|
||
|
||
private Double getDoubleObj(JsonNode node, String... names) {
|
||
for (String name : names) {
|
||
JsonNode fn = node.get(name);
|
||
if (fn != null && !fn.isNull()) {
|
||
if (fn.isNumber()) return fn.asDouble();
|
||
try {
|
||
return Double.parseDouble(fn.asText());
|
||
} catch (Exception ignored) {
|
||
}
|
||
}
|
||
}
|
||
return null;
|
||
}
|
||
|
||
private int getInt(JsonNode node, String... names) {
|
||
for (String name : names) {
|
||
JsonNode fn = node.get(name);
|
||
if (fn != null && !fn.isNull()) {
|
||
if (fn.isNumber()) return fn.asInt();
|
||
try {
|
||
return Integer.parseInt(fn.asText());
|
||
} catch (Exception ignored) {
|
||
}
|
||
}
|
||
}
|
||
return -1;
|
||
}
|
||
|
||
private String getString(JsonNode node, String... names) {
|
||
for (String name : names) {
|
||
JsonNode fn = node.get(name);
|
||
if (fn != null && !fn.isNull()) {
|
||
return fn.asText();
|
||
}
|
||
}
|
||
return null;
|
||
}
|
||
|
||
@Override
|
||
public void onConnectionLost(String deviceKey, Throwable cause) {
|
||
log.warn("主机监控连接断开: deviceKey={}, reason={}", deviceKey, cause.getMessage());
|
||
}
|
||
|
||
@Override
|
||
public void onSubscribed(String deviceKey, String topic) {
|
||
log.info("主机监控订阅主题: deviceKey={}, topic={}", deviceKey, topic);
|
||
}
|
||
}
|