指令

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

@@ -48,9 +48,14 @@ public class MqttCommand {
     private String sendTime;
 
     /**
-     * 指令执行时间
+     * 客户端指令上报时间
      */
-    private String execTime;
+    private String backClientTime;
+
+    /**
+     * 服务端指令上报时间
+     */
+    private String backServerTime;
 
     /**
      * 指令有效期

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

@@ -57,12 +57,12 @@ public class MqttCommandJob {
      */
     @XxlJob("commandDataCloud")
     public ReturnT<String> commandDataCloud() {
-        //过期时间小于当前时间的阈值，单位s
+        //指令执行超时分钟数
         SysDictRedis sysDictRedis = SysDictUtils.get(DictEnum.DATA_CLEAR_SIZE.getKey(), "mqtt_command_cloud");
         Integer jobParam = JobUtils.getPrams().gtInt();
         Integer threshold = jobParam != null ? jobParam
                 : sysDictRedis != null ? Integer.valueOf(sysDictRedis.getValue())
-                : 30;
+                : 10;
         LambdaEsQueryWrapper<MqttCommand> lambdaEsQueryWrapper = new LambdaEsQueryWrapper<MqttCommand>()
                 .in(MqttCommand::getStatus, Arrays.asList(MqttCommandStatusEnum.COMMAND_SEND.getKey()))
                 .le(MqttCommand::getSendTime, DataTime.getStringAround(0, 0, 0, 0, threshold, 0));

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

@@ -137,7 +137,7 @@ public class DeviceQualityServiceImpl extends ServiceImpl<DeviceQualityMapper, D
 
     @PostMapping("rest")
     @ApiOperation("重置多项质检")
-    public R rest(@RequestBody DeviceQualityDto.Rest rest) {
+    public R<List<Tuple.Tuple3<Long, Long, Boolean>>> rest(@RequestBody DeviceQualityDto.Rest rest) {
         SysDictRedis sysDictRedis1 = SysDictUtils.get(DictEnum.DEVICE_QUALITY_STATUS.getKey(), DictSonEnum.DEVICE_QUALITY_STATUS_QUALIFIED.getKey());
         SysDictRedis sysDictRedis2 = SysDictUtils.get(DictEnum.DEVICE_FACTORY_STATUS.getKey(), DictSonEnum.DEVICE_FACTORY_STATUS_CAN_SHIPPED.getKey());
         DeviceQuality deviceQuality = getById(rest.getQualityId());
@@ -171,13 +171,12 @@ public class DeviceQualityServiceImpl extends ServiceImpl<DeviceQualityMapper, D
                     .setTemplet(templet);
             commandMqtts.add(commandMqtt);
         });
-        mqttService.senCommand(commandMqtts);
-        return R.ok();
+        return mqttService.senCommand(commandMqtts);
     }
 
     @PostMapping("save")
     @ApiOperation("添加")
-    public R save(@RequestBody List<DeviceQualityDto.Save> deviceQualitySaves) {
+    public R<List<Tuple.Tuple3<Long, Long, Boolean>>> save(@RequestBody List<DeviceQualityDto.Save> deviceQualitySaves) {
         List<DeviceQualityDto.Vo> updates = new ArrayList<>();
         List<DeviceQuality> saves = new ArrayList<>();
         JList<Long> deviceIds = new JArrayList<>(deviceQualitySaves).getProperty(DeviceQualityDto.Save::getDeviceId);
@@ -241,7 +240,7 @@ public class DeviceQualityServiceImpl extends ServiceImpl<DeviceQualityMapper, D
             }
         });
         if (Emptys.check(commandMqtts)) {
-            mqttService.senCommand(commandMqtts);
+            return mqttService.senCommand(commandMqtts);
         }
         return R.ok();
     }

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

@@ -6,7 +6,7 @@ import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
 import com.xy.annotate.Runners;
 import com.xy.annotate.Timer;
-import com.xy.dto.BackMqttDto;
+import com.xy.dto.AckMqttDto;
 import com.xy.dto.CommandMqtt;
 import com.xy.dto.MqttDto;
 import com.xy.dto.PaterDto;
@@ -18,31 +18,33 @@ import com.xy.utils.consts.CommConsts;
 import com.xy.utils.enums.MqttCommandStatusEnum;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
-import lombok.AllArgsConstructor;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
+import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
+import static com.xy.utils.Beans.copy;
 import static com.xy.utils.EsBeans.toPageBean;
 
 @Slf4j
 @Service
 @Api(tags = "mqtt接口")
-@AllArgsConstructor
+@RequiredArgsConstructor
 public class MqttServiceImpl implements MqttService {
 
-    private MqttProducer mqttProducer;
+    private final MqttProducer mqttProducer;
 
-    private MqttCommandMapper mqttCommandMapper;
+    private final MqttCommandMapper mqttCommandMapper;
 
-    private RedisService redisService;
+    private final RedisService<String> redisService;
+
+    private String keyStr = "mqtt:cmd:result:%d";
 
     private long getSn() {
         return redisService.atomicIncrease(CommConsts.CMD_ATOM_SN);
@@ -52,9 +54,9 @@ public class MqttServiceImpl implements MqttService {
         return redisService.atomicIncrease(CommConsts.CMD_ATOM_WK_SN);
     }
 
-    private Map<Long, Boolean> sendMqtt(List<MqttDto> mqttDtos) {
+    private List<Tuple.Tuple3<Long, Long, Boolean>> sendMqtt(List<MqttDto> mqttDtos) {
         String time = DataTime.getSring();
-        Map<Long, Boolean> map = new HashMap<>(mqttDtos.size());
+        List<Tuple.Tuple3<Long, Long, Boolean>> list = new ArrayList<>(mqttDtos.size());
         mqttDtos.forEach(mqttDto -> {
             //给予默认值
             PaterDto pater = mqttDto.getPater()
@@ -77,17 +79,17 @@ public class MqttServiceImpl implements MqttService {
             Long wkSn = pater.getWkSn();
             try {
                 //指令记录
-                saveMqttCommand(mqttDto, value, pater.getAck() ? MqttCommandStatusEnum.COMMAND_SEND : MqttCommandStatusEnum.COMMAND_SUCCESS, null);
+                saveMqttCommand(mqttDto, value, pater.getAck() ? MqttCommandStatusEnum.COMMAND_SEND : MqttCommandStatusEnum.COMMAND_EXE_SUCCESS, null);
                 //发送消息
                 mqttProducer.sendToMqtt(value.toString(), topic, mqttDto.getLevel());
-                map.put(wkSn, true);
+                list.add(new Tuple.Tuple3<>(wkSn, pater.getSn(), true));
             } catch (Exception e) {
                 //指令记录
                 saveMqttCommand(mqttDto, value, MqttCommandStatusEnum.COMMAND_SEND_FAIL, e.getMessage());
-                map.put(wkSn, false);
+                list.add(new Tuple.Tuple3<>(wkSn, pater.getSn(), false));
             }
         });
-        return map;
+        return list;
     }
 
     @Timer
@@ -110,7 +112,7 @@ public class MqttServiceImpl implements MqttService {
     @Timer
     @Override
     @ApiOperation("发送指令")
-    public R<Map<Long, Boolean>> senCommand(List<CommandMqtt> commandMqtts) {
+    public R<List<Tuple.Tuple3<Long, Long, Boolean>>> senCommand(List<CommandMqtt> commandMqtts) {
         List<MqttDto> mqttDtos = new ArrayList<>();
         commandMqtts.forEach(commandMqtt -> {
             JSONObject templetObj = commandMqtt.getTemplet();
@@ -138,29 +140,40 @@ public class MqttServiceImpl implements MqttService {
                     .setPater(paterDto);
             mqttDtos.add(mqttDto);
         });
-        Map<Long, Boolean> map = sendMqtt(mqttDtos);
-        return R.ok(map);
+        return R.ok(sendMqtt(mqttDtos));
     }
 
-    @Override
-    @ApiOperation("指令结果通知")
-    public R commandBack(BackMqttDto backMqttDto) {
-        MqttCommand mqttCommand = mqttCommandMapper.selectById(backMqttDto.getSn());
+    @ApiOperation("指令Ack")
+    @PostMapping("commandAck")
+    public R commandAck(@RequestBody @Validated AckMqttDto ackMqttDto) {
+        MqttCommand mqttCommand = mqttCommandMapper.selectById(ackMqttDto.getSn());
         if (!Emptys.check(mqttCommand)) {
             return R.fail("数据不存在");
         }
-        //重复上报，前面的时间不能覆盖后面的时间
-        if (!StringUtils.isEmpty(mqttCommand.getExecTime())) {
-            if (DataTime.stringContrast(DataTime.toString(backMqttDto.getExecTime()), mqttCommand.getExecTime()) < 0) {
-                return R.ok();
-            }
-        }
-        mqttCommand.setStatus(Integer.parseInt(backMqttDto.getBool() ? MqttCommandStatusEnum.COMMAND_SUCCESS.getKey() : MqttCommandStatusEnum.COMMAND_EXC_FAIL.getKey()))
-                .setExecTime(DataTime.toString(backMqttDto.getExecTime()));
+        //写入es
+        mqttCommand.setStatus(Integer.valueOf(ackMqttDto.getStatusEnum().getKey()))
+                .setBackClientTime(DataTime.toString(ackMqttDto.getBackClientTime()))
+                .setBackServerTime(DataTime.getSring());
         mqttCommandMapper.updateById(mqttCommand);
+        //写入redis
+        String result = ackMqttDto.getResult().toString();
+        if (Emptys.check(ackMqttDto.getResult())) {
+            String key = String.format(keyStr, ackMqttDto.getSn());
+            redisService.set(key, result);
+        }
         return R.ok();
     }
 
+    @PostMapping("snByCmdAndResult")
+    @ApiOperation("sn查询指令和结果数据")
+    public R<MqttDto.Vo3> snByCmdAndResult(@RequestBody MqttDto.SnByCmdAndResult snByCmdAndResult) {
+        MqttCommand mqttCommand = mqttCommandMapper.selectById(snByCmdAndResult.getSn());
+        String key = String.format(keyStr, snByCmdAndResult.getSn());
+        String result = redisService.get(key);
+        MqttDto.Vo3 vo3 = copy(MqttDto.Vo3.class, mqttCommand).setResult(JSONUtil.parseObj(result));
+        return R.ok(vo3);
+    }
+
     @PostMapping("page")
     @ApiOperation("分页查询")
     public R<PageBean<MqttDto.Vo2>> page(@RequestBody MqttDto.Page page) {
@@ -196,7 +209,6 @@ public class MqttServiceImpl implements MqttService {
                 .setStatus(Integer.parseInt(status.getKey()))
                 .setValue(value.toString())
                 .setSendTime(pater.getTime())
-                .setExecTime(status.getKey().equals(MqttCommandStatusEnum.COMMAND_SUCCESS.getKey()) ? DataTime.getSring() : null)
                 .setNum(1)
                 .setError(error);
         mqttCommandMapper.insert(mqttCommand);

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

@@ -0,0 +1,32 @@
+package com.xy.dto;
+
+import cn.hutool.json.JSONObject;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.xy.utils.enums.MqttCommandStatusEnum;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+import javax.validation.constraints.NotNull;
+import java.time.LocalDateTime;
+
+@Data
+@Accessors(chain = true)
+public class AckMqttDto {
+
+    @NotNull(message = "sn不能为空")
+    @ApiModelProperty(value = "sn编号", required = true)
+    private Long sn;
+
+    @NotNull(message = "status不能为空")
+    @ApiModelProperty(value = "状态", required = true)
+    private MqttCommandStatusEnum statusEnum;
+
+    @NotNull(message = "backClientTime不能为空")
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    @ApiModelProperty(value = "上报时间 yyyy-MM-dd HH:mm:ss", required = true)
+    private LocalDateTime backClientTime;
+
+    @ApiModelProperty(value = "执行结果")
+    private JSONObject result;
+}

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

@@ -1,29 +0,0 @@
-package com.xy.dto;
-
-import com.fasterxml.jackson.annotation.JsonFormat;
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.Data;
-import lombok.experimental.Accessors;
-
-import javax.validation.constraints.NotNull;
-import java.time.LocalDateTime;
-
-@Data
-@ApiModel("指令执行结果回执交互类")
-@Accessors(chain = true)
-public class BackMqttDto {
-
-    @NotNull(message = "sn不能为空")
-    @ApiModelProperty(value = "sn编号", required = true)
-    private Long sn;
-
-    @NotNull(message = "bool不能为空")
-    @ApiModelProperty(value = "结果 true=成功 false=失败", required = true)
-    private Boolean bool;
-
-    @NotNull(message = "execTime不能为空")
-    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
-    @ApiModelProperty(value = "执行时间 yyyy-MM-dd HH:mm:ss", required = true)
-    private LocalDateTime execTime;
-}

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

@@ -1,5 +1,6 @@
 package com.xy.dto;
 
+import cn.hutool.json.JSONObject;
 import com.fasterxml.jackson.annotation.JsonFormat;
 import com.xy.utils.PageBean;
 import io.swagger.annotations.ApiModelProperty;
@@ -7,6 +8,7 @@ import lombok.Data;
 import lombok.experimental.Accessors;
 
 import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotNull;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 
@@ -32,6 +34,15 @@ public class MqttDto {
     @ApiModelProperty(value = "指令对象")
     private PaterDto pater;
 
+    @Data
+    @Accessors(chain = true)
+    public static class SnByCmdAndResult {
+
+        @NotNull(message = "sn不能为空")
+        @ApiModelProperty("sn")
+        private Long sn;
+    }
+
     @Data
     @Accessors(chain = true)
     public static class Page extends Vo2 {
@@ -113,8 +124,11 @@ public class MqttDto {
         @ApiModelProperty("指令发送时间")
         private String sendTime;
 
-        @ApiModelProperty("指令执行时间")
-        private String execTime;
+        @ApiModelProperty("客户端指令上报时间")
+        private String backClientTime;
+
+        @ApiModelProperty("服务端指令上报时间")
+        private String backServerTime;
 
         @ApiModelProperty("指令过期时间")
         private String timeout;
@@ -137,4 +151,13 @@ public class MqttDto {
         @ApiModelProperty("错误日志")
         private String error;
     }
+
+    @Data
+    @Accessors(chain = true)
+    public static class Vo3 extends Vo2 {
+
+        @ApiModelProperty("指令执行结果")
+        private JSONObject result;
+
+    }
 }

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

@@ -1,15 +1,13 @@
 package com.xy.service;
 
 import com.xy.annotate.RestMappingController;
-import com.xy.dto.BackMqttDto;
 import com.xy.dto.CommandMqtt;
 import com.xy.utils.R;
-import org.springframework.validation.annotation.Validated;
+import com.xy.utils.Tuple;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 
 import java.util.List;
-import java.util.Map;
 
 /**
  * mqtt接口
@@ -24,14 +22,6 @@ public interface MqttService {
      * @return
      */
     @PostMapping("senCommand")
-    R<Map<Long, Boolean>> senCommand(@RequestBody List<CommandMqtt> commandMqtts);
+    R<List<Tuple.Tuple3<Long, Long, Boolean>>> senCommand(@RequestBody List<CommandMqtt> commandMqtts);
 
-    /**
-     * 指令结果通知
-     *
-     * @param backMqttDto
-     * @return
-     */
-    @PostMapping("commandBack")
-    R commandBack(@RequestBody @Validated BackMqttDto backMqttDto);
 }

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

@@ -7,10 +7,10 @@ public enum MqttCommandStatusEnum {
     //指令状态
     COMMAND_SEND("1", "已发送"),
     COMMAND_SEND_FAIL("2", "发送失败"),
-    COMMAND_EXECUTE("3", "已执行"),
+    COMMAND_EXE_SUCCESS("3", "已执行"),
     COMMAND_EXC_FAIL("4", "执行失败"),
     COMMAND_EXC_TIMEOUT("5", "执行超时"),
-    COMMAND_SUCCESS("6", "已完成"),
+    COMMAND_RECEIVE("6", "已接收"),
     ;
 
     private String key;