[add]

1.添加 agent 删除接口
2.添加 redis 队列数据数量字段
3.服务新增编辑删除添加下发 mqtt server 的逻辑
[fix] 关注 mall 列表返回事件记录 bug

src/main/java/vion/constant/MqttMessageType.java

@@ -4,10 +4,20 @@ import lombok.Getter;
 
 import java.util.Arrays;
 
+/**
+ * mqtt 消息枚举
+ */
 @Getter
 public enum MqttMessageType {
 
     /**
+     * 服务列表增删改下发
+     */
+    SERVICE_ADD("add"),
+    SERVICE_UPDATE("update"),
+    SERVICE_REMOVE("remove"),
+    SERVICE_ASSIGN("assign"),
+    /**
      * 注册
      */
     REGISTER("REGISTER"),

src/main/java/vion/controller/monitor/AgentController.java

@@ -44,6 +44,12 @@ public class AgentController {
         return agentService.update(dto);
     }
 
+    @DeleteMapping("/{id}")
+    @SaCheckPermission(value = "agent:remove", orRole = "admin")
+    public String remove(@PathVariable Long id) {
+        return agentService.removeById(id) ? "删除成功" : "删除失败";
+    }
+
     @PostMapping("/upgrade/{uid}")
     @SaCheckPermission(value = "agent:upgrade", orRole = "admin")
     public String upgradeCommand(@PathVariable String uid, Long upgradeId) {

src/main/java/vion/event/mqtt/MqttClientMessageListener.java

@@ -39,65 +39,36 @@ public class MqttClientMessageListener {
             JsonNode jsonObj = JsonUtil.parseTree(payload);
             String type = jsonObj.path("eventType").asText();
             switch (MqttMessageType.getEnumByType(type)) {
-                case MqttMessageType.REGISTER:
+                case REGISTER -> {
                     String agentUid = jsonObj.path("agentUid").asText();
                     updateTaskByAgent(agentUid);
-                    break;
-                case MqttMessageType.PASSENGER_FLOW_INTERRUPT:
-                    handlePassengerFlowInterrupt(jsonObj.toString());
-                    break;
-                case MqttMessageType.DEVICE_OFFLINE:
-                    handleDeviceOffline(jsonObj.toString());
-                    break;
-                case MqttMessageType.REID_ANALYZE:
-                    handleReid(jsonObj.toString());
-                    break;
-                case HEADCOUNT_RATIO:
-                    handleHeadcountRatio(jsonObj.toString());
-                    break;
-                case STAFF_RECOGNIZE:
-                    handleStaffRecognize(jsonObj.toString());
-                    break;
-                // region store 事件
-                case STORE_CUSTOMER_UNDULATE:
-                    handleStoreCustomerUndulate(jsonObj.toString());
-                    break;
-                case STORE_GATE_DATA_UNDULATE:
-                    handleStoreGateDateUndulate(jsonObj.toString());
-                    break;
-                case STORE_INOUT_MATCH_RATIO:
-                    handleStoreInoutMatchRatio(jsonObj.toString());
-                    break;
-                case STORE_SINGLE_CLUSTER:
-                    handleStoreSingleCluster(jsonObj.toString());
-                    break;
-                case STORE_ENTER_RATIO:
-                    handleStoreEnterRatio(jsonObj.toString());
-                    break;
+                }
+                // region mall + store 共有指标
+                case PASSENGER_FLOW_INTERRUPT -> handlePassengerFlowInterrupt(jsonObj.toString());
+                case DEVICE_OFFLINE -> handleDeviceOffline(jsonObj.toString());
+                case REID_ANALYZE -> handleReid(jsonObj.toString());
+                case DEVICE_REGISTRATION -> handleDeviceRegistration(jsonObj.toString());
+                case HEADCOUNT_RATIO -> handleHeadcountRatio(jsonObj.toString());
+                case STAFF_RECOGNIZE -> handleStaffRecognize(jsonObj.toString());
                 // endregion
-                // region mall 事件
-                case MALL_INOUT_DIFF:
-                    handleMallInoutDiff(jsonObj.toString());
-                    break;
-                case MALL_SHOP_INOUT_DIFF:
-                    handleMallShopInoutDiff(jsonObj.toString());
-                    break;
-                case MALL_DATA_UNDULATE:
-                    handleMallDataUndulate(jsonObj.toString());
-                    break;
-                case MALL_GATE_DATA_UNDULATE:
-                    handleMallGateDataUndulate(jsonObj.toString());
-                    break;
-                case MALL_SHOP_DATA_UNDULATE:
-                    handleMallShopDataUndulate(jsonObj.toString());
-                    break;
-                case MALL_DILATATION_UNDULATE:
-                    handleMallDilatationUndulate(jsonObj.toString());
-                    break;
+
+                // region store 指标
+                case STORE_CUSTOMER_UNDULATE -> handleStoreCustomerUndulate(jsonObj.toString());
+                case STORE_GATE_DATA_UNDULATE -> handleStoreGateDateUndulate(jsonObj.toString());
+                case STORE_INOUT_MATCH_RATIO -> handleStoreInoutMatchRatio(jsonObj.toString());
+                case STORE_SINGLE_CLUSTER -> handleStoreSingleCluster(jsonObj.toString());
+                case STORE_ENTER_RATIO -> handleStoreEnterRatio(jsonObj.toString());
                 // endregion
-                default:
-                    log.info("未定义的消息类型:{}, payload:{}", type, jsonObj);
-                    break;
+
+                // region mall 指标
+                case MALL_INOUT_DIFF -> handleMallInoutDiff(jsonObj.toString());
+                case MALL_SHOP_INOUT_DIFF -> handleMallShopInoutDiff(jsonObj.toString());
+                case MALL_DATA_UNDULATE -> handleMallDataUndulate(jsonObj.toString());
+                case MALL_GATE_DATA_UNDULATE -> handleMallGateDataUndulate(jsonObj.toString());
+                case MALL_SHOP_DATA_UNDULATE -> handleMallShopDataUndulate(jsonObj.toString());
+                case MALL_DILATATION_UNDULATE -> handleMallDilatationUndulate(jsonObj.toString());
+                // endregion
+                default -> log.info("未定义的消息类型:{}, payload:{}", type, jsonObj);
             }
         } catch (Exception e) {
             log.error("解析数据异常", e);
@@ -114,7 +85,7 @@ public class MqttClientMessageListener {
                 .eq(RAgentEvent::getAgentUid, agentUid)
                 .eq(RAgentEvent::getControlSwitch, 1).list();
         if (CollUtil.isEmpty(list)) {
-            log.error("agent:{} 未配置事件监测", agentUid);
+            log.info("agent:{} 未配置事件监测", agentUid);
             return;
         }
         var topic = TopicUtil.getEventTopic(agentUid);
@@ -150,6 +121,16 @@ public class MqttClientMessageListener {
     }
 
     /**
+     * 设备注册次数异常
+     *
+     * @param payloadStr 事件记录
+     */
+    private void handleDeviceRegistration(String payloadStr) {
+        EventRecord eventRecord = JsonUtil.parseObject(payloadStr, EventRecord.class);
+        recordService.save(eventRecord);
+    }
+
+    /**
      * 人数/人次异常
      *
      * @param payloadStr 事件记录

src/main/java/vion/model/monitor/AgentRecord.java

@@ -48,6 +48,12 @@ public class AgentRecord {
     private Object networkInfo;
 
     /**
+     * redis 队列数据数量
+     */
+    @TableField(value = "queue_data", typeHandler = JsonbTypeHandler.class)
+    private Object queueData;
+
+    /**
      * 启动时间
      */
     @TableField(value = "boot_time")

src/main/java/vion/service/impl/monitor/AgentServiceImpl.java

@@ -21,6 +21,7 @@ import org.dromara.hutool.core.util.RandomUtil;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import vion.constant.MqttMessageType;
 import vion.dto.monitor.AgentDTO;
 import vion.dto.monitor.MqttAuthDTO;
 import vion.dto.monitor.OrgDTO;
@@ -180,17 +181,52 @@ public class AgentServiceImpl extends MPJBaseServiceImpl<AgentMapper, Agent> imp
                 return rAgentService;
             }).toList();
             rAgentServiceService.saveBatch(rAgentServiceList);
+            var topic = TopicUtil.getServiceTopic(uid);
+            if (StrUtil.isBlank(topic)) {
+                var msg = StrUtil.format("agent:{} 获取topic失败", uid);
+                log.error(msg);
+                return msg;
             }
-        return "更新成功";
+            var objNode = JsonUtil.createObj()
+                    .put("type", MqttMessageType.SERVICE_ADD.getType())
+                    .putPOJO("services", JsonUtil.toJsonString(rAgentServiceList));
+            client.publish(topic, JsonUtil.toJsonByte(objNode), MqttQoS.QOS2);
+        }
+        return "保存成功";
     }
 
     @Override
     public String updateAgent2ServiceById(Long id, RAgentService rAgentService) {
-        return rAgentServiceService.updateById(rAgentService) ? "更新成功" : "更新失败";
+        if (rAgentServiceService.updateById(rAgentService)) {
+            var agentService = rAgentServiceService.getById(id);
+            var topic = TopicUtil.getServiceTopic(agentService.getAgentUid());
+            if (StrUtil.isBlank(topic)) {
+                var msg = StrUtil.format("agent:{} 获取topic失败", agentService.getAgentUid());
+                log.error(msg);
+                return msg;
+            }
+            var objNode = JsonUtil.createObj()
+                    .put("type", MqttMessageType.SERVICE_UPDATE.getType())
+                    .putPOJO("services", JsonUtil.toJsonString(agentService));
+            client.publish(topic, JsonUtil.toJsonByte(objNode), MqttQoS.QOS2);
+            return "更新成功";
+        }
+        return "更新失败";
     }
 
     @Override
     public String removeAgent2ServiceById(Long id) {
+        var agentService = rAgentServiceService.getById(id);
+        var topic = TopicUtil.getServiceTopic(agentService.getAgentUid());
+        if (StrUtil.isBlank(topic)) {
+            var msg = StrUtil.format("agent:{} 获取topic失败", agentService.getAgentUid());
+            log.error(msg);
+            return msg;
+        }
+        var objNode = JsonUtil.createObj()
+                .put("type", MqttMessageType.SERVICE_REMOVE.getType())
+                .putPOJO("services", JsonUtil.toJsonString(agentService));
+        client.publish(topic, JsonUtil.toJsonByte(objNode), MqttQoS.QOS2);
         return rAgentServiceService.removeById(id) ? "删除成功 : " : "删除失败";
     }
 
@@ -202,7 +238,10 @@ public class AgentServiceImpl extends MPJBaseServiceImpl<AgentMapper, Agent> imp
             log.error(msg);
             return msg;
         }
-        return client.publish(topic, JsonUtil.toJsonByte(serviceInfoList), MqttQoS.QOS2) ? "服务列表下发成功" :
+        var objNode = JsonUtil.createObj()
+                .put("type", MqttMessageType.SERVICE_ASSIGN.getType())
+                .putPOJO("services", JsonUtil.toJsonString(serviceInfoList));
+        return client.publish(topic, JsonUtil.toJsonByte(objNode), MqttQoS.QOS2) ? "服务列表下发成功" :
                 "服务列表下发失败";
     }
 

src/main/java/vion/service/impl/monitor/EventServiceImpl.java

@@ -52,9 +52,9 @@ public class EventServiceImpl extends MPJBaseServiceImpl<EventMapper, Event> imp
                             .orderByDesc("mall_uid", "create_time");
                     var eventRecordList1 = eventRecordService.selectJoinList(EventRecord.class, eventRecWrapper1);*/
                     var eventRecWrapper = Wrappers.<EventRecord>query()
-                            .select("DISTINCT on (mall_uid) *")
+                            .select("DISTINCT on (mall_uid,event_type) *")
                             .eq("mall_uid", dto.getMallUid())
-                            .orderByDesc("mall_uid", "create_time");
+                            .orderByDesc("mall_uid", "event_type", "create_time");
                     var eventRecordList = eventRecordService.list(eventRecWrapper);
                     var eventUid2SelfMap = eventRecordList.stream().collect(Collectors.toMap(EventRecord::getEventUid, Function.identity()));
 

src/main/java/vion/service/impl/monitor/MallServiceImpl.java

@@ -87,9 +87,9 @@ public class MallServiceImpl extends MPJBaseServiceImpl<MallMapper, Mall> implem
                     var mallUidList = r.stream().map(MallVO::getUid).toList();
                     // fixme {@link EventServiceImpl #51}
                     var eventRecWrapper = Wrappers.<EventRecord>query()
-                            .select("DISTINCT on (mall_uid) *")
+                            .select("DISTINCT on (mall_uid,event_type) *")
                             .in("mall_uid", mallUidList)
-                            .orderByDesc("mall_uid", "create_time");
+                            .orderByDesc("mall_uid", "event_type", "create_time");
                     var eventRecordList = eventRecordService.list(eventRecWrapper);
                     var mallUid2RecMap = eventRecordList.stream().collect(Collectors.groupingBy(EventRecord::getMallUid));
 
@@ -107,15 +107,14 @@ public class MallServiceImpl extends MPJBaseServiceImpl<MallMapper, Mall> implem
     @Override
     public Page<MallVO> listErrorAttention(MallDTO dto) {
         var eventRecWrapper = Wrappers.<EventRecord>query()
-                .select("DISTINCT on (mall_uid) *")
+                .select("DISTINCT on (mall_uid,event_type) *")
                 .eq("agent_type", dto.getAgentType())
-                .ne("status", 1)
-                .orderByDesc("mall_uid", "create_time");
+                .orderByDesc("mall_uid", "event_type", "create_time");
         var eventRecordList = eventRecordService.list(eventRecWrapper);
-        if (CollUtil.isEmpty(eventRecordList)) {
-            return null;
+        var mallUidList = eventRecordList.stream().filter(r -> r.getStatus() != 1).map(EventRecord::getMallUid).toList();
+        if (CollUtil.isEmpty(mallUidList)) {
+            return new Page<>();
         }
-        var mallUidList = eventRecordList.stream().map(EventRecord::getMallUid).toList();
         var agentEventList = agentEventService.lambdaQuery().in(RAgentEvent::getMallUid, mallUidList).list();
         var mallUid2AgentEventMap = agentEventList.stream().collect(Collectors.groupingBy(RAgentEvent::getMallUid));