feat(mqtt): 新增设备状态管理功能并优化消息处理流程

lvqi
1 parent be4fa92b
Showing 10 changed files with 189 additions and 12 deletions
mqtt-core/src/main/java/com/diligrp/mqtt/core/type/DeviceStatusEnum.java
mqtt-core/src/main/java/com/diligrp/mqtt/core/type/TopicType.java
mqtt-web/src/main/java/com/diligrp/mqtt/web/model/DeviceStatusInfoRecord.java
mqtt-web/src/main/java/com/diligrp/mqtt/web/model/PrintDevice.java
mqtt-core/src/main/java/com/diligrp/mqtt/core/model/ReceiveRelyRecord.java → mqtt-web/src/main/java/com/diligrp/mqtt/web/model/ReceiveRelyRecord.java
mqtt-web/src/main/java/com/diligrp/mqtt/web/processor/printer/PrinterOnlineReceiveProcessor.java
mqtt-web/src/main/java/com/diligrp/mqtt/web/processor/printer/PrinterPostProcessor.java
mqtt-web/src/main/java/com/diligrp/mqtt/web/processor/printer/PrinterSuccessReceiveProcessor.java
mqtt-web/src/main/java/com/diligrp/mqtt/web/service/PrintService.java
mqtt-web/src/main/java/com/diligrp/mqtt/web/service/impl/PrintServiceImpl.java
+package com.diligrp.mqtt.core.type;
+
+import lombok.Getter;
+
+/**
+ * @author lvqi
+ */
+@Getter
+public enum DeviceStatusEnum {
+    ONLINE(1, "在线"),
+    OFFLINE(2, "离线");
+
+
+    private final Integer code;
+    private final String message;
+
+
+    DeviceStatusEnum(Integer code, String message) {
+        this.code = code;
+        this.message = message;
+
+    }
+}
 package com.diligrp.mqtt.core.type;
+import lombok.Getter;
+
+@Getter
 public enum TopicType {
     PRINTER_ONLINE_SERVICE("/priGoOnline/#", 1),
+package com.diligrp.mqtt.web.model;
+
+import lombok.Data;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.time.LocalDateTime;
+
+/**
+ * 设备状态记录
+ *
+ * @author lvqi
+ */
+@Document("device_status_info_record")
+@Data
+public class DeviceStatusInfoRecord {
+    @Id
+    private String id;
+
+    private String firm;
+    private String productId;
+
+    private String sn;
+
+    private Integer status;
+
+    private LocalDateTime createTime;
+}
@@ -21,7 +21,7 @@ public class PrintDevice implements Serializable {
     /**
      * 产品id
      */
-    private Integer productId;
+    private String productId;
     /**
      * 设备编号
      */
-package com.diligrp.mqtt.core.model;
+package com.diligrp.mqtt.web.model;
 import lombok.Data;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.core.mapping.Document;
+import java.time.LocalDateTime;
+
 /**
  * 打印信息记录
@@ -31,9 +33,15 @@ public class ReceiveRelyRecord {
      */
     private Integer messStatus;
+    private LocalDateTime createTime;
+
     /**
      * 客户端回执状态
      */
     private String replyStatus;
+    private LocalDateTime replyTime;
+
+    private Integer retries;
+
 }
@@ -2,9 +2,20 @@ package com.diligrp.mqtt.web.processor.printer;
 import com.diligrp.mqtt.core.model.ReceiveModel;
 import com.diligrp.mqtt.core.service.ReceiveProcessorService;
+import com.diligrp.mqtt.core.type.DeviceStatusEnum;
+import com.diligrp.mqtt.core.type.FirmEnum;
 import com.diligrp.mqtt.core.type.ReceiveEventType;
+import com.diligrp.mqtt.core.type.TopicType;
+import com.diligrp.mqtt.core.util.JsonUtils;
+import com.diligrp.mqtt.web.model.DeviceStatusInfoRecord;
+import com.fasterxml.jackson.core.type.TypeReference;
+import jakarta.annotation.Resource;
+import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.stereotype.Component;
+import java.time.LocalDateTime;
+import java.util.Map;
+
 /**
  * @Author: zhangmeiyang
  * @CreateTime: 2026-01-15 16:10
@@ -12,6 +23,9 @@ import org.springframework.stereotype.Component;
  */
 @Component
 public class PrinterOnlineReceiveProcessor implements ReceiveProcessorService {
+    @Resource
+    private MongoTemplate mongoTemplate;
+
     @Override
     public ReceiveEventType receiveEventType() {
         return ReceiveEventType.PRINTER_ONLINE_SERVICE;
@@ -19,6 +33,24 @@ public class PrinterOnlineReceiveProcessor implements ReceiveProcessorService {
     @Override
     public void successReceive(ReceiveModel receiveModel) {
+        String topic = receiveModel.getTopic();
+        if (topic.contains(TopicType.PRINTER_ONLINE_SERVICE.getValue())) {
+            String payload = receiveModel.getPayload();
+            Map<String, String> map = JsonUtils.convertValue(payload, new TypeReference<>() {
+            });
+            String deviceSn = map.get("deviceSn");
+            String firm = FirmEnum.CLOUD.code;
+            String[] split = topic.split("/");
+            String productId = split[2];
+            DeviceStatusInfoRecord deviceStatusInfoRecord = new DeviceStatusInfoRecord();
+            deviceStatusInfoRecord.setId(String.valueOf(System.currentTimeMillis()));
+            deviceStatusInfoRecord.setFirm(firm);
+            deviceStatusInfoRecord.setProductId(productId);
+            deviceStatusInfoRecord.setSn(deviceSn);
+            deviceStatusInfoRecord.setStatus(DeviceStatusEnum.ONLINE.getCode());
+            deviceStatusInfoRecord.setCreateTime(LocalDateTime.now());
+            mongoTemplate.save(deviceStatusInfoRecord);
+        }
     }
 }
 package com.diligrp.mqtt.web.processor.printer;
+import com.diligrp.mqtt.core.model.CloudRequest;
+import com.diligrp.mqtt.web.model.ReceiveRelyRecord;
 import com.diligrp.mqtt.core.model.SendModel;
 import com.diligrp.mqtt.core.service.PostProcessorService;
+import com.diligrp.mqtt.core.type.MessStatusEnum;
 import com.diligrp.mqtt.core.type.SendEventType;
+import com.diligrp.mqtt.core.util.JsonUtils;
+import jakarta.annotation.Resource;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.stereotype.Component;
+import java.util.Map;
+
 /**
  * @Author: zhangmeiyang
  * @CreateTime: 2026-01-15 16:08
@@ -12,6 +23,9 @@ import org.springframework.stereotype.Component;
  */
 @Component
 public class PrinterPostProcessor implements PostProcessorService {
+    @Resource
+    private MongoTemplate mongoTemplate;
+
     @Override
     public SendEventType sendEventType() {
         return SendEventType.PRINTER;
@@ -24,6 +38,11 @@ public class PrinterPostProcessor implements PostProcessorService {
     @Override
     public void failProcess(SendModel sendModel) {
-
+        Map<String, Object> params = sendModel.getParams();
+        CloudRequest cloudRequest = JsonUtils.convertValue(params, CloudRequest.class);
+        Update update = new Update();
+        update.set("messStatus", MessStatusEnum.FAIL.getCode());
+        Query query = new Query(Criteria.where("_id").is(cloudRequest.getMessIndex() + cloudRequest.getSn()));
+        mongoTemplate.updateFirst(query, update, ReceiveRelyRecord.class);
     }
 }
@@ -3,8 +3,19 @@ package com.diligrp.mqtt.web.processor.printer;
 import com.diligrp.mqtt.core.model.ReceiveModel;
 import com.diligrp.mqtt.core.service.ReceiveProcessorService;
 import com.diligrp.mqtt.core.type.ReceiveEventType;
+import com.diligrp.mqtt.core.util.JsonUtils;
+import com.diligrp.mqtt.web.model.ReceiveRelyRecord;
+import com.fasterxml.jackson.core.type.TypeReference;
+import jakarta.annotation.Resource;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.stereotype.Component;
+import java.time.LocalDateTime;
+import java.util.Map;
+
 /**
  * @Author: zhangmeiyang
  * @CreateTime: 2026-01-15 16:09
@@ -12,6 +23,9 @@ import org.springframework.stereotype.Component;
  */
 @Component
 public class PrinterSuccessReceiveProcessor implements ReceiveProcessorService {
+    @Resource
+    private MongoTemplate mongoTemplate;
+
     @Override
     public ReceiveEventType receiveEventType() {
         return ReceiveEventType.PRINTER_RECEIPT_SERVICE;
@@ -19,6 +33,16 @@ public class PrinterSuccessReceiveProcessor implements ReceiveProcessorService {
     @Override
     public void successReceive(ReceiveModel receiveModel) {
-
+        String payload = receiveModel.getPayload();
+        Map<String, String> map = JsonUtils.convertValue(payload, new TypeReference<>() {
+        });
+        String messIndex = map.get("messIndex");
+        String deviceSn = map.get("deviceSn");
+        String messStatus = map.get("messStatus");
+        Update update = new Update();
+        update.set("replyStatus", messStatus);
+        update.set("replyTime", LocalDateTime.now());
+        Query query = new Query(Criteria.where("_id").is(messIndex + deviceSn));
+        mongoTemplate.updateFirst(query, update, ReceiveRelyRecord.class);
     }
 }
 package com.diligrp.mqtt.web.service;
-import com.diligrp.mqtt.web.message.Message;
 import com.diligrp.mqtt.core.model.CloudRequest;
+import com.diligrp.mqtt.web.message.Message;
+import com.diligrp.mqtt.web.model.ReceiveRelyRecord;
+
+import java.util.List;
 /**
  * @author lvqi
@@ -9,4 +12,6 @@ import com.diligrp.mqtt.core.model.CloudRequest;
 public interface PrintService {
     Message<?> cloudPrint(CloudRequest cloudRequest);
+
+    void retry(List<ReceiveRelyRecord> receiveRelyRecords);
 }
 package com.diligrp.mqtt.web.service.impl;
+import com.diligrp.mqtt.core.dto.PrintDeviceDto;
 import com.diligrp.mqtt.core.model.CloudMessContent;
 import com.diligrp.mqtt.core.model.CloudRequest;
-import com.diligrp.mqtt.core.model.ReceiveRelyRecord;
 import com.diligrp.mqtt.core.model.SendModel;
 import com.diligrp.mqtt.core.service.MqttMessageService;
 import com.diligrp.mqtt.core.type.CharsetType;
@@ -11,16 +11,16 @@ import com.diligrp.mqtt.core.type.SendEventType;
 import com.diligrp.mqtt.core.util.BuildCloudPrintCommand;
 import com.diligrp.mqtt.core.util.JsonUtils;
 import com.diligrp.mqtt.web.dto.CloudPrinter;
-import com.diligrp.mqtt.core.dto.PrintDeviceDto;
 import com.diligrp.mqtt.web.mapper.PlatTenantMapper;
 import com.diligrp.mqtt.web.mapper.PrintDeviceMapper;
 import com.diligrp.mqtt.web.message.Message;
 import com.diligrp.mqtt.web.model.PlatTenant;
+import com.diligrp.mqtt.web.model.ReceiveRelyRecord;
 import com.diligrp.mqtt.web.service.PrintService;
 import com.fasterxml.jackson.core.type.TypeReference;
+import jakarta.annotation.Resource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.util.ObjectUtils;
@@ -29,6 +29,7 @@ import java.nio.charset.StandardCharsets;
 import java.security.KeyFactory;
 import java.security.Signature;
 import java.security.spec.PKCS8EncodedKeySpec;
+import java.time.LocalDateTime;
 import java.util.*;
 /**
@@ -37,13 +38,13 @@ import java.util.*;
 @Service
 public class PrintServiceImpl implements PrintService {
     private static final Logger log = LoggerFactory.getLogger(PrintServiceImpl.class);
-    @Autowired
+    @Resource
     private PlatTenantMapper platTenantMapper;
-    @Autowired
+    @Resource
     private PrintDeviceMapper printDeviceMapper;
-    @Autowired
+    @Resource
     private MongoTemplate mongoTemplate;
-    @Autowired
+    @Resource
     private MqttMessageService mqttMessageService;
     @Override
@@ -106,10 +107,44 @@ public class PrintServiceImpl implements PrintService {
             receiveRelyRecord.setMessIndex(cloudRequest.getMessIndex());
             receiveRelyRecord.setMessage(JsonUtils.toJsonString(cloudRequest));
             receiveRelyRecord.setMessStatus(messStatus);
+            receiveRelyRecord.setCreateTime(LocalDateTime.now());
             mongoTemplate.save(receiveRelyRecord);
         }
     }
+    @Override
+    public void retry(List<ReceiveRelyRecord> receiveRelyRecords) {
+        for (ReceiveRelyRecord receiveRelyRecord : receiveRelyRecords) {
+            String message = receiveRelyRecord.getMessage();
+            CloudRequest cloudRequest = JsonUtils.fromJsonString(message, CloudRequest.class);
+            SendModel sendModel = SendModel.withDefault();
+            CloudPrinter cloudPrinter = new CloudPrinter();
+            cloudPrinter.setIndex(cloudRequest.getMessIndex());
+            cloudPrinter.setPrint_num(Integer.valueOf(cloudRequest.getNum()));
+            CloudMessContent mess = cloudRequest.getMess();
+            try {
+                byte[] bytes = BuildCloudPrintCommand.buildCommand(mess.getPc());
+                cloudPrinter.setPrint_esc(Base64.getEncoder().encodeToString(bytes));
+            } catch (Exception e) {
+                log.error("构建打印命令失败", e);
+            }
+            // 语音播报
+            cloudPrinter.setTts(mess.getSc());
+            cloudPrinter.setRtc(cloudRequest.getFirmId());
+            PrintDeviceDto printDevice = printDeviceMapper.queryBySn(cloudRequest.getSn());
+            sendModel.setTopic(printDevice.getTopic());
+            Map<String, Object> dataMap = JsonUtils.convertValue(cloudPrinter, new TypeReference<>() {
+            });
+            sendModel.setPayload(dataMap);
+            sendModel.setSendEventType(SendEventType.PRINTER.value);
+            sendModel.setCharsetType(CharsetType.GBK.code);
+            Map<String, Object> requestMap = JsonUtils.convertValue(cloudRequest, new TypeReference<>() {
+            });
+            sendModel.setParams(requestMap);
+            mqttMessageService.sendMessage(sendModel);
+        }
+    }
+
     public static String handleStr(Map<String, Object> paramsMap) {
         //转成map
         //获取参数名称转成List集合