mqtt指令

device-api-service/src/main/java/com/xy/entity/MqttCommand.java

@@ -23,9 +23,14 @@ public class MqttCommand {
     private Boolean debug;
 
     /**
-     * 指令类型
+     * 指令大类
      */
-    private String type;
+    private String type1;
+
+    /**
+     * 指令小类
+     */
+    private String type2;
 
     /**
      * 指令状态 1=已发送 2=发送失败 3=已执行 4=执行失败，5=执行超时 6=已完成

device-api-service/src/main/java/com/xy/job/MqttCommandJob.java

@@ -9,7 +9,7 @@ import com.xy.entity.MqttCommand;
 import com.xy.mapper.MqttCommandMapper;
 import com.xy.utils.DataTime;
 import com.xy.utils.JobUtils;
-import com.xy.utils.enums.MqttEnum;
+import com.xy.utils.enums.MqttCommandStatusEnum;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -55,7 +55,7 @@ public class MqttCommandJob {
         Integer threshold = JobUtils.getPrams().gtInt();
         threshold = threshold == null ? 0 : threshold;
         LambdaEsQueryWrapper<MqttCommand> lambdaEsQueryWrapper = new LambdaEsQueryWrapper<MqttCommand>()
-                .in(MqttCommand::getStatus, Arrays.asList(MqttEnum.COMMAND_SEND.getKey(), MqttEnum.COMMAND_SEND_FAIL.getKey()))
+                .in(MqttCommand::getStatus, Arrays.asList(MqttCommandStatusEnum.COMMAND_SEND.getKey(), MqttCommandStatusEnum.COMMAND_SEND_FAIL.getKey()))
                 .le(MqttCommand::getTimeout, DataTime.getStringAround(0, 0, 0, 0, 0, (~(threshold - 1))));
         List<MqttCommand> mqttCommands = mqttCommandMapper.selectList(lambdaEsQueryWrapper);
         if (mqttCommands.size() == 0) {
@@ -63,12 +63,12 @@ public class MqttCommandJob {
         }
         JList<MqttCommand> list = new JArrayList<>(mqttCommands);
         //发送但没有回执的数据
-        JList<MqttCommand> sends = list.filter().eq(MqttCommand::getStatus, MqttEnum.COMMAND_SEND.getKey()).list();
-        sends.forEach(mqttCommand -> mqttCommand.setStatus(Integer.parseInt(MqttEnum.COMMAND_EXC_TIMEOUT.getKey())));
+        JList<MqttCommand> sends = list.filter().eq(MqttCommand::getStatus, MqttCommandStatusEnum.COMMAND_SEND.getKey()).list();
+        sends.forEach(mqttCommand -> mqttCommand.setStatus(Integer.parseInt(MqttCommandStatusEnum.COMMAND_EXC_TIMEOUT.getKey())));
         mqttCommandMapper.updateBatchByIds(sends);
         log.info("已处理发送但没有回执的指令数据：{}条", sends.size());
         //发送失败的数据
-        JList<MqttCommand> sendFails = list.filter().eq(MqttCommand::getStatus, MqttEnum.COMMAND_SEND_FAIL.getKey()).list();
+        JList<MqttCommand> sendFails = list.filter().eq(MqttCommand::getStatus, MqttCommandStatusEnum.COMMAND_SEND_FAIL.getKey()).list();
         //todo 计数报表
         return ReturnT.SUCCESS;
     }

device-api-service/src/main/java/com/xy/service/impl/MqttServiceImpl.java

@@ -1,6 +1,7 @@
 package com.xy.service.impl;
 
 import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
 import com.github.yitter.idgen.YitIdHelper;
 import com.xy.dto.BackMqttDto;
 import com.xy.dto.MqttDto;
@@ -10,16 +11,18 @@ import com.xy.mapper.MqttCommandMapper;
 import com.xy.producer.MqttProducer;
 import com.xy.service.MqttService;
 import com.xy.utils.*;
-import com.xy.utils.enums.MqttEnum;
+import com.xy.utils.enums.MqttCommandStatusEnum;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 
 import java.util.ArrayList;
 import java.util.List;
 
+@Slf4j
 @Service
 @Api(tags = "mqtt接口")
 @AllArgsConstructor
@@ -35,33 +38,35 @@ public class MqttServiceImpl implements MqttService {
         List<MqttDto.Vo> list = new ArrayList<>(mqttDtos.size());
         mqttDtos.forEach(mqttDto -> {
             //给予默认值
-            JSONObject command = mqttDto.getCommand();
-            command.set(LambdaUtils.getProperty(PaterDto::getSn), YitIdHelper.nextId())
-                    .set(LambdaUtils.getProperty(PaterDto::getTime), DataTime.getSring());
+            PaterDto pater = mqttDto.getPater()
+                    .setSn(String.valueOf(YitIdHelper.nextId()))
+                    .setTime(DataTime.getSring());
             //延迟发布处理
             String delayTime = mqttDto.getDelayTime();
             String topic = mqttDto.getTopic();
             if (!StringUtils.isEmpty(delayTime)) {
                 //消息消费时间 - 发布时间的差值 = 距今被消费的秒数
-                long m = DataTime.diff(DataTime.toLocal(command.getStr(LambdaUtils.getProperty(PaterDto::getTime))), DataTime.toLocal(delayTime), "s");
+                long m = DataTime.diff(DataTime.toLocal(pater.getTime()), DataTime.toLocal(delayTime), "s");
                 topic = String.format("$delayed/%d/%s", m, mqttDto.getTopic());
                 mqttDto.setTimeout(mqttDto.getTimeout() + (int) m);
             }
             //处理消息过期时间
-            command.set(LambdaUtils.getProperty(PaterDto::getTimeout), DataTime.getStringAround(0, 0, 0, 0, 0, mqttDto.getTimeout()));
-            //写入新的指令json
-            mqttDto.setCommand(command);
-            String workSn = command.getStr(LambdaUtils.getProperty(PaterDto::getWorkSn));
+            pater.setTimeout(DataTime.getStringAround(0, 0, 0, 0, 0, mqttDto.getTimeout()));
+            //翻译枚举
+            JSONObject value = JSONUtil.parseObj(pater)
+                    .set(LambdaUtils.getProperty(PaterDto::getType1), pater.getType1().getKey())
+                    .set(LambdaUtils.getProperty(PaterDto::getType2), pater.getType2().getKey());
+            String wkSn = pater.getWkSn();
             try {
                 //发送消息
-                mqttProducer.sendToMqtt(command.toString(), topic, mqttDto.getLevel());
+                mqttProducer.sendToMqtt(value.toString(), topic, mqttDto.getLevel());
                 //指令记录
-                saveMqttCommand(mqttDto, command.getBool(LambdaUtils.getProperty(PaterDto::getAck)) ? MqttEnum.COMMAND_SEND.getKey() : MqttEnum.COMMAND_SUCCESS.getKey(), null);
-                list.add(new MqttDto.Vo().setWorkSn(workSn).setResult(true));
+                saveMqttCommand(mqttDto, value, pater.getAck() ? MqttCommandStatusEnum.COMMAND_SEND.getKey() : MqttCommandStatusEnum.COMMAND_SUCCESS.getKey(), null);
+                list.add(new MqttDto.Vo().setWorkSn(wkSn).setResult(true));
             } catch (Exception e) {
                 //指令记录
-                saveMqttCommand(mqttDto, MqttEnum.COMMAND_SEND_FAIL.getKey(), e.getMessage());
-                list.add(new MqttDto.Vo().setWorkSn(workSn).setResult(false));
+                saveMqttCommand(mqttDto, value, MqttCommandStatusEnum.COMMAND_SEND_FAIL.getKey(), e.getMessage());
+                list.add(new MqttDto.Vo().setWorkSn(wkSn).setResult(false));
             }
         });
         return list;
@@ -80,7 +85,7 @@ public class MqttServiceImpl implements MqttService {
                 return R.ok();
             }
         }
-        mqttCommand.setStatus(Integer.parseInt(backMqttDto.getBool() ? MqttEnum.COMMAND_SUCCESS.getKey() : MqttEnum.COMMAND_EXC_FAIL.getKey()))
+        mqttCommand.setStatus(Integer.parseInt(backMqttDto.getBool() ? MqttCommandStatusEnum.COMMAND_SUCCESS.getKey() : MqttCommandStatusEnum.COMMAND_EXC_FAIL.getKey()))
                 .setExecTime(DataTime.toString(backMqttDto.getExecTime()));
         mqttCommandMapper.updateById(mqttCommand);
         return R.ok();
@@ -90,17 +95,20 @@ public class MqttServiceImpl implements MqttService {
      * 新增指令记录
      *
      * @param mqttDto 参数对象
+     * @param value   参数json对象
      * @param status  状态
      * @param error   错误信息
      */
-    private void saveMqttCommand(MqttDto mqttDto, String status, String error) {
-        PaterDto command = mqttDto.getCommand().toBean(PaterDto.class);
-        MqttCommand mqttCommand = Beans.copy(Beans.copy(MqttCommand.class, mqttDto), command)
-                .setId(command.getSn())
+    private void saveMqttCommand(MqttDto mqttDto, JSONObject value, String status, String error) {
+        PaterDto pater = mqttDto.getPater();
+        MqttCommand mqttCommand = Beans.copy(Beans.copy(MqttCommand.class, mqttDto), pater)
+                .setId(pater.getSn())
+                .setType1(pater.getType1().getKey())
+                .setType2(pater.getType2().getKey())
                 .setStatus(Integer.parseInt(status))
-                .setValue(mqttDto.getCommand().toString())
-                .setSendTime(command.getTime())
-                .setExecTime(status.equals(MqttEnum.COMMAND_SUCCESS.getKey()) ? DataTime.getSring() : null)
+                .setValue(value.toString())
+                .setSendTime(pater.getTime())
+                .setExecTime(status.equals(MqttCommandStatusEnum.COMMAND_SUCCESS.getKey()) ? DataTime.getSring() : null)
                 .setNum(1)
                 .setError(error);
         mqttCommandMapper.insert(mqttCommand);

device-api/src/main/java/com/xy/dto/MqttDto.java

@@ -1,16 +1,19 @@
 package com.xy.dto;
 
-import cn.hutool.json.JSONObject;
+import com.xy.dto.pater.PaterDto;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 import lombok.experimental.Accessors;
 
+import javax.validation.constraints.NotBlank;
+
 @Data
 @ApiModel("指令交互类")
 @Accessors(chain = true)
 public class MqttDto {
 
+    @NotBlank(message = "topic不能为空")
     @ApiModelProperty(value = "设备topic", required = true)
     private String topic;
 
@@ -26,8 +29,8 @@ public class MqttDto {
     @ApiModelProperty("指令有效期，单位：s 默认=60")
     private Integer timeout = 60;
 
-    @ApiModelProperty(value = "指令json", required = true)
-    private JSONObject command;
+    @ApiModelProperty(value = "指令对象", required = true)
+    private PaterDto pater;
 
     @Data
     @ApiModel("指令返参类")

device-api/src/main/java/com/xy/dto/pater/PaterDto.java

@@ -1,5 +1,7 @@
 package com.xy.dto.pater;
 
+import com.xy.utils.enums.MqttType1Enum;
+import com.xy.utils.enums.MqttType2Enum;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
@@ -13,14 +15,14 @@ public class PaterDto {
     @ApiModelProperty(value = "sn编号", hidden = true)
     private String sn;
 
-    @ApiModelProperty(value = "业务系统sn编号")
-    private String workSn;
+    @ApiModelProperty(value = "业务系统sn编号", required = true)
+    private String wkSn;
 
-    @ApiModelProperty("业务类型（大类） issue=服务器下发设备执行指令 report=服务器通知设备上报信息")
-    private Integer work;
+    @ApiModelProperty(value = "大类", required = true)
+    private MqttType1Enum type1;
 
-    @ApiModelProperty(value = "指令类型（小类） 对应枚举类MqttEnum")
-    private String type;
+    @ApiModelProperty(value = "小类", required = true)
+    private MqttType2Enum type2;
 
     @ApiModelProperty("是否应答 默认=true")
     private Boolean ack = true;

device-api/src/main/java/com/xy/service/MqttService.java

@@ -20,7 +20,7 @@ public interface MqttService {
      * 指令发布
      *
      * @param mqttDtos
-     * @return k=topic v=布尔结果
+     * @return k=wkSn v=布尔结果
      */
     @PostMapping("sendMqtt")
     List<MqttDto.Vo> sendMqtt(@RequestBody List<MqttDto> mqttDtos);

device-api/src/main/java/com/xy/utils/CommandUtils.java

@@ -1,93 +0,0 @@
-package com.xy.utils;
-
-import com.xy.dto.son.*;
-import com.xy.utils.enums.MqttEnum;
-import lombok.SneakyThrows;
-
-import java.lang.reflect.Method;
-
-/**
- * 指令构造工具类
- */
-public class CommandUtils {
-
-    @SneakyThrows
-    private static <T> T news(Class<T> tClass, String type) {
-        T t = tClass.newInstance();
-        Method setType = tClass.getMethod("setType", String.class);
-        setType.invoke(t, type);
-        return t;
-    }
-
-    /**
-     * 摄像头
-     *
-     * @return
-     */
-    public static CameraDto camera() {
-        return news(CameraDto.class, MqttEnum.COMMAND_CAMERAS.getKey());
-    }
-
-    /**
-     * 电池
-     *
-     * @return
-     */
-    public static CellDto cell() {
-        return news(CellDto.class, MqttEnum.COMMAND_CELL.getKey());
-    }
-
-    /**
-     * 自检
-     *
-     * @return
-     */
-    public static CheckDto check() {
-        return news(CheckDto.class, MqttEnum.COMMAND_CHECK.getKey());
-    }
-
-    /**
-     * 灯
-     *
-     * @return
-     */
-    public static LampDto lamp() {
-        return news(LampDto.class, MqttEnum.COMMAND_LAMP.getKey());
-    }
-
-    /**
-     * 锁机
-     *
-     * @return
-     */
-    public static LockDeviceDto LockDevice() {
-        return news(LockDeviceDto.class, MqttEnum.COMMAND_LOCK_DEVICE.getKey());
-    }
-
-    /**
-     * 门锁
-     *
-     * @return
-     */
-    public static LockDto lock() {
-        return news(LockDto.class, MqttEnum.COMMAND_LOCK.getKey());
-    }
-
-    /**
-     * 日志
-     *
-     * @return
-     */
-    public static LogDto log() {
-        return news(LogDto.class, MqttEnum.COMMAND_LOG.getKey());
-    }
-
-    /**
-     * 温度
-     *
-     * @return
-     */
-    public static TemperatureDto temperature() {
-        return news(TemperatureDto.class, MqttEnum.COMMAND_TEMPERATURE.getKey());
-    }
-}

device-api/src/main/java/com/xy/utils/enums/MqttCommandStatusEnum.java

@@ -0,0 +1,24 @@
+package com.xy.utils.enums;
+
+import lombok.Getter;
+
+@Getter
+public enum MqttCommandStatusEnum {
+    //指令状态
+    COMMAND_SEND("1", "已发送"),
+    COMMAND_SEND_FAIL("2", "发送失败"),
+    COMMAND_EXECUTE("3", "已执行"),
+    COMMAND_EXC_FAIL("4", "执行失败"),
+    COMMAND_EXC_TIMEOUT("5", "执行超时"),
+    COMMAND_SUCCESS("6", "已完成"),
+    ;
+
+    private String key;
+
+    private String msg;
+
+    MqttCommandStatusEnum(String key, String msg) {
+        this.key = key;
+        this.msg = msg;
+    }
+}

device-api/src/main/java/com/xy/utils/enums/MqttType1Enum.java

@@ -0,0 +1,19 @@
+package com.xy.utils.enums;
+
+import lombok.Getter;
+
+@Getter
+public enum MqttType1Enum {
+    ISSUE("issue", "服务器下发设备执行指令"),
+    REPORT("report", "服务器通知设备上报信息"),
+    ;
+
+    private String key;
+
+    private String msg;
+
+    MqttType1Enum(String key, String msg) {
+        this.key = key;
+        this.msg = msg;
+    }
+}

device-api/src/main/java/com/xy/utils/enums/MqttEnum.java → device-api/src/main/java/com/xy/utils/enums/MqttType2Enum.java

@@ -3,7 +3,7 @@ package com.xy.utils.enums;
 import lombok.Getter;
 
 @Getter
-public enum MqttEnum {
+public enum MqttType2Enum {
     //指令类型
     COMMAND_CHECK("check", "自检指令"),
     COMMAND_LOCK_DEVICE("lockDevice", "锁机指令"),
@@ -13,20 +13,13 @@ public enum MqttEnum {
     COMMAND_CAMERAS("camera", "摄像头指令"),
     COMMAND_CELL("cell", "电池指令"),
     COMMAND_LOG("log", "日志指令"),
-    //指令状态
-    COMMAND_SEND("1", "已发送"),
-    COMMAND_SEND_FAIL("2", "发送失败"),
-    COMMAND_EXECUTE("3", "已执行"),
-    COMMAND_EXC_FAIL("4", "执行失败"),
-    COMMAND_EXC_TIMEOUT("5", "执行超时"),
-    COMMAND_SUCCESS("6", "已完成"),
     ;
 
     private String key;
 
     private String msg;
 
-    MqttEnum(String key, String msg) {
+    MqttType2Enum(String key, String msg) {
         this.key = key;
         this.msg = msg;
     }