mqtt消息添加心跳包处理，设备新增运行状态字段以及接口

fastbee-gateway/fastbee-mq/src/main/java/com/fastbee/mq/redischannel/consumer/DeviceOtherMsgConsumer.java

@@ -76,18 +76,21 @@ public class DeviceOtherMsgConsumer {
                 NgWaterPumpUsageRecords pumpUsageRecords=new NgWaterPumpUsageRecords();
                 pumpUsageRecords.setDeviceNumber(serialNumber);
                 JSONObject jsonObject = JSONUtil.parseObj(data);
-                jsonObject.set("time", (LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))));//收到消息的当地时间
+                // jsonObject.set("time", (LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))));//收到消息的当地时间
                 if(jsonObject.getStr("type")!=null){
+                    //上报正常数据
                     if(jsonObject.getStr("type").equals("waterEleData")){
                         String data1 = jsonObject.getStr("data");
-                        //收到定时报文
+                        //收到开阀定时报文
                         if(JSONUtil.parseObj(data1).get("action").equals("timeMsg")){
                             pumpUsageRecords.setMessageContent(JSONUtil.toJsonStr(jsonObject));
                             // int i = ngWaterPumpUsageRecordsMapper.insertNgWaterPumpUsageRecords(pumpUsageRecords);
                             // if(i<1){
                             //     System.err.println("--------------------------保存使用记录失败!---------------------------");
                             // }
-                        }else if(JSONUtil.parseObj(data1).get("action").equals("safeMsg")){
+                        }
+                        //收到关阀时定时报文
+                        else if(JSONUtil.parseObj(data1).get("action").equals("safeMsg")){
                             //续费在线状态
                             updateDeviceOnlineStatus( productId,serialNumber);
                             //保存数据
@@ -97,14 +100,28 @@ public class DeviceOtherMsgConsumer {
                             //     System.err.println("--------------------------保存使用记录失败!---------------------------");
                             // }
 
-                        }//开阀门，关阀门
+                        }
+                        //开阀门，关阀门
                         else if(JSONUtil.parseObj(data1).get("action").equals("startPump")||JSONUtil.parseObj(data1).get("action").equals("stopPump")){
                             int i = ngWaterPumpUsageRecordsMapper.insertNgWaterPumpUsageRecords(pumpUsageRecords);
                             if(i<1){
                                 System.err.println("--------------------------保存使用记录失败!---------------------------");
                             }
                         }
+
                     }
+                    //收到心跳报文
+                    else if(jsonObject.getStr("type").equals("heartBeat")){
+                            //需要回应
+                            //构建回复消息-----------------------------------------------------------------------------
+                            // Map<String,Object> reportMsg=new HashMap<>();
+                            // reportMsg.put("code",100);
+                            // reportMsg.put("pakSn","ok");
+                            // reportMsg.put("data","heartBeat");
+                            jsonObject.set("code",100);
+                            issueInstructionsProducer.receiveDataReportResponse(productId.toString(),serialNumber,JSONUtil.toJsonStr(jsonObject));
+                        }
+
                 }
                 //构建回复消息-----------------------------------------------------------------------------
                 // Map<String,Object> reportMsg=new HashMap<>();

fastbee-open-api/src/main/java/com/fastbee/data/controller/DeviceOperationController.java

@@ -2,8 +2,12 @@ package com.fastbee.data.controller;
 
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
+import com.baomidou.mybatisplus.extension.conditions.update.LambdaUpdateChainWrapper;
 import com.dtflys.forest.annotation.Post;
 import com.fastbee.common.core.domain.AjaxResult;
+import com.fastbee.common.exception.ServiceException;
+import com.fastbee.iot.domain.DeviceReportInfo;
+import com.fastbee.iot.mapper.DeviceReportInfoMapper;
 import com.fastbee.iot.model.dto.DeviceOperationDTO;
 import com.fastbee.mqttclient.PubMqttCallBack;
 import com.fastbee.mqttclient.PubMqttClient;
@@ -20,6 +24,7 @@ import java.util.Map;
 
 /**
  * 下发指令操作设备
+ * @author mijiupro
  */
 @RestController
 @RequestMapping("/device/operation")
@@ -31,6 +36,9 @@ public class DeviceOperationController {
     @Autowired
     private PubMqttCallBack pubMqttCallBack;
 
+    @Autowired
+    private DeviceReportInfoMapper deviceReportInfoMapper;
+
     /**
      * 下发指令控制阀门设备
      */

fastbee-open-api/src/main/java/com/fastbee/data/controller/DeviceReportInfoController.java

@@ -8,6 +8,10 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import javax.servlet.http.HttpServletResponse;
 
+import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
+import com.baomidou.mybatisplus.extension.conditions.update.LambdaUpdateChainWrapper;
+import com.fastbee.common.exception.ServiceException;
+import com.fastbee.iot.mapper.DeviceReportInfoMapper;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import org.apache.commons.lang3.StringUtils;
@@ -42,6 +46,9 @@ public class DeviceReportInfoController extends BaseController
     @Autowired
     private IDeviceReportInfoService deviceReportInfoService;
 
+    @Autowired
+    private DeviceReportInfoMapper deviceReportInfoMapper;
+
     @Autowired
     private StringRedisTemplate stringRedisTemplate;
 
@@ -83,27 +90,25 @@ public class DeviceReportInfoController extends BaseController
     //统计是设备相关信息
     @GetMapping("/list/count")
     @ApiOperation("统计是设备相关信息")
-    public AjaxResult listCount(DeviceReportInfo deviceReportInfo)
+    public AjaxResult listCount(DeviceReportInfo deviceReportInfo) {
-        {
+        List<DeviceReportInfo> list = deviceReportInfoService.selectDeviceReportInfoList(deviceReportInfo);
-            List<DeviceReportInfo> list = deviceReportInfoService.selectDeviceReportInfoList(deviceReportInfo);
+        int onLineTotal = 0; // 使用普通的整型变量来计数
-            AtomicInteger onLineTotal= new AtomicInteger();
+        for (DeviceReportInfo d : list) {
-            list.forEach(d->{
+            if (d.getType() == 1) {
-                if(d.getType()==1){
+                if (Boolean.TRUE.equals(stringRedisTemplate.hasKey("neixiang_device_online_status:" + 147 + ":" + d.getSerialNumber()))) {
-                    if(Boolean.TRUE.equals(stringRedisTemplate.hasKey("neixiang_device_online_status:" + 147 + ":" + d.getSerialNumber())))
+                    onLineTotal++;
-                    {
-                        onLineTotal.getAndIncrement();
-                    }
-                } else if ( d.getType()==2) {
-                    if(d.getOnLine()==1){
-                        onLineTotal.getAndIncrement();
-                    }
                 }
-            });
+            } else if (d.getType() == 2) {
-            Map<String,Object> resp=new HashMap<>();
+                if (d.getOnLine() == 1) {
-            resp.put("allTotal",list.size());
+                    onLineTotal++;
-            resp.put("onLineTotal",onLineTotal.get());
+                }
-            return AjaxResult.success((resp));
+            }
         }
+        Map<String, Object> resp = new HashMap<>();
+        resp.put("allTotal", list.size());
+        resp.put("onLineTotal", onLineTotal);
+        return AjaxResult.success(resp);
+    }
 
     /**
      * 导出设备上电审核前上报的基础信息列表
@@ -171,4 +176,34 @@ public class DeviceReportInfoController extends BaseController
     public AjaxResult updateStatus(@RequestBody List<DeviceReportInfo> deviceReportInfos){
         return toAjax(deviceReportInfoService.updateDeviceReportStatus(deviceReportInfos));
     }
+
+    /**
+     * 更新设备当前状态
+     */
+    @PostMapping("/update/run/status")
+    public AjaxResult updateStatus(String deviceNumber,Integer status){
+        boolean update = new LambdaUpdateChainWrapper<>(deviceReportInfoMapper)
+                .set(DeviceReportInfo::getStatus, status)
+                .eq(DeviceReportInfo::getSerialNumber, deviceNumber)
+                .update();
+        if(!update){
+            throw new ServiceException("更新设备运行状态失败！");
+        }
+        return AjaxResult.success();
+    }
+    /**
+     * 查询设备运行状态
+     */
+    @GetMapping("/get/run/status")
+    public AjaxResult getStatus(String deviceNumber){
+        List<DeviceReportInfo> list = new LambdaQueryChainWrapper<>(deviceReportInfoMapper)
+                .select(DeviceReportInfo::getStatus)
+                .eq(DeviceReportInfo::getSerialNumber, deviceNumber)
+                .list();
+        if(!list.isEmpty()){
+            return AjaxResult.success(list.get(0).getStatus());
+        }
+        return AjaxResult.error("设备不存在！");
+    }
+
 }

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/DeviceReportInfo.java

@@ -169,6 +169,11 @@ public class DeviceReportInfo extends BaseEntity
     @Excel(name = "所属机构")
     @ApiModelProperty("所属机构")
     private Long deptId;
+
+    /** yun'xing */
+    @Excel(name = "yun'xing")
+    @ApiModelProperty("yun'xing")
+    private Long runStatus;
     //----------------------------------------------------------业务字段-----------------------------------------------------------
     /** 是否自动审核 */
     @TableField(exist = false)

fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceReportInfoMapper.xml

@@ -35,10 +35,11 @@
         <result property="onLine"    column="on_line"    />
         <result property="areaCode"    column="area_code"    />
         <result property="deptId"    column="dept_id"    />
+        <result property="runStatus"    column="run_status"    />
     </resultMap>
 
     <sql id="selectDeviceReportInfoVo">
-        select id, imei, iccid, mcu_id, bsp_type, lte_type, mcu_type, mcu_fw, lte_fw, lcd_manufacturer, voice_manufacturer, fram_model, replace_manufacturer, test_record, batch_number, serial_number, qr_code, name, lcd_manufacturer_name, voice_manufacturer_name, replace_manufacturer_name, device_id, status, powers_time, type, longitude, latitude,on_line,area_code,dept_id from iot_device_report_info
+        select id, imei, iccid, mcu_id, bsp_type, lte_type, mcu_type, mcu_fw, lte_fw, lcd_manufacturer, voice_manufacturer, fram_model, replace_manufacturer, test_record, batch_number, serial_number, qr_code, name, lcd_manufacturer_name, voice_manufacturer_name, replace_manufacturer_name, device_id, status, powers_time, type, longitude, latitude,on_line,area_code,dept_id,run_status from iot_device_report_info
     </sql>
 
     <select id="selectDeviceReportInfoList" parameterType="DeviceReportInfo" resultMap="DeviceReportInfoResult">
@@ -73,6 +74,7 @@
             <if test="onLine != null "> and on_line = #{onLine}</if>
             <if test="areaCode != null  and areaCode != ''"> and area_code = #{areaCode}</if>
             <if test="deptId != null "> and dept_id = #{deptId}</if>
+            <if test="runStatus != null "> and run_status = #{runStatus}</if>
         </where>
         <!-- 添加排序 -->
         order by powers_time desc