增加定时更新设备状态任务

springboot/fastbee-gateway/fastbee-mq/src/main/java/com/fastbee/mq/service/IMqttMessagePublish.java

@@ -6,6 +6,7 @@ import com.fastbee.common.core.mq.message.DeviceDownMessage;
 import com.fastbee.common.core.mq.message.InstructionsMessage;
 import com.fastbee.common.core.mq.ota.OtaUpgradeBo;
 import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem;
+import com.fastbee.common.enums.DeviceStatus;
 import com.fastbee.common.enums.TopicType;
 import com.fastbee.iot.domain.Device;
 import com.fastbee.mq.model.ReportDataBo;
@@ -74,5 +75,13 @@ public interface IMqttMessagePublish {
      */
     public Device deviceSynchronization(String deviceNumber);
 
+    /**
+     * 推送设备状态
+     *
+     * @param device 设备
+     * @param status       状态
+     */
+    public void pushDeviceStatus(Device device, DeviceStatus status) ;
+
 
 }

springboot/fastbee-open-api/src/main/java/com/fastbee/data/service/impl/DeviceJob.java

@@ -1,7 +1,12 @@
 package com.fastbee.data.service.impl;
 
+import com.fastbee.base.session.Session;
+import com.fastbee.common.enums.DeviceStatus;
 import com.fastbee.iot.domain.Device;
+import com.fastbee.iot.model.DeviceStatusVO;
 import com.fastbee.iot.service.IDeviceService;
+import com.fastbee.mq.service.IMqttMessagePublish;
+import com.fastbee.mqtt.manager.SessionManger;
 import com.fastbee.sip.domain.SipDevice;
 import com.fastbee.sip.domain.SipDeviceChannel;
 import com.fastbee.sip.enums.DeviceChannelStatus;
@@ -9,7 +14,9 @@ import com.fastbee.sip.mapper.SipDeviceChannelMapper;
 import com.fastbee.sip.mapper.SipDeviceMapper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
 
 import javax.annotation.Resource;
 import java.util.List;
@@ -28,6 +35,12 @@ public class DeviceJob {
     @Autowired
     private SipDeviceChannelMapper sipDeviceChannelMapper;
 
+    @Resource
+    private IMqttMessagePublish mqttMessagePublish;
+
+    @Value("${server.broker.enabled}")
+    private Boolean enabled;
+
     public void updateSipDeviceOnlineStatus(Integer timeout) {
         List<SipDevice> devs = sipDeviceMapper.selectOfflineSipDevice(timeout);
         devs.forEach(item -> {
@@ -48,4 +61,37 @@ public class DeviceJob {
             }
         });
     }
+
+    /**
+     * 定期同步设备状态
+     *  1.将异常在线设备变更为离线状态
+     *  2.将离线设备但实际在线设备变更为在线
+     */
+    public void syncDeviceStatus() {
+        if (enabled) {
+            //获取所有已激活并不是禁用的设备
+            List<DeviceStatusVO> deviceStatusVOList = deviceService.selectDeviceActive();
+            if (!CollectionUtils.isEmpty(deviceStatusVOList)) {
+                for (DeviceStatusVO statusVO : deviceStatusVOList) {
+                    Session session = SessionManger.getSession(statusVO.getSerialNumber());
+                    Device device = new Device();
+                    device.setSerialNumber(statusVO.getSerialNumber());
+                    device.setRssi(statusVO.getRssi());
+                    device.setProductId(statusVO.getProductId());
+                    device.setIsShadow(statusVO.getIsShadow());
+                    // 如果session中设备在线，数据库状态离线 ,则更新设备的状态为在线
+                    if (!Objects.isNull(session) && statusVO.getStatus() == DeviceStatus.OFFLINE.getType()) {
+                        device.setStatus(DeviceStatus.ONLINE.getType());
+                        deviceService.updateDeviceStatus(device);
+                        mqttMessagePublish.pushDeviceStatus(device, DeviceStatus.ONLINE);
+                    }
+                    if (Objects.isNull(session) && statusVO.getStatus() == DeviceStatus.ONLINE.getType()) {
+                        device.setStatus(DeviceStatus.OFFLINE.getType());
+                        deviceService.updateDeviceStatus(device);
+                        mqttMessagePublish.pushDeviceStatus(device, DeviceStatus.OFFLINE);
+                    }
+                }
+            }
+        }
+    }
 }

springboot/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/service/impl/MqttMessagePublishImpl.java

@@ -13,6 +13,7 @@ import com.fastbee.common.core.mq.ota.OtaUpgradeBo;
 import com.fastbee.common.core.protocol.modbus.ModbusCode;
 import com.fastbee.common.core.redis.RedisCache;
 import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem;
+import com.fastbee.common.enums.DeviceStatus;
 import com.fastbee.common.enums.FunctionReplyStatus;
 import com.fastbee.common.enums.ServerType;
 import com.fastbee.common.enums.TopicType;
@@ -63,6 +64,10 @@ public class MqttMessagePublishImpl implements IMqttMessagePublish {
     private IProductService productService;
     @Resource
     private PubMqttClient mqttClient;
+
+    @Resource
+    private PubMqttClient pubMqttClient;
+
     @Resource
     private TopicsUtils topicsUtils;
     @Resource
@@ -78,6 +83,7 @@ public class MqttMessagePublishImpl implements IMqttMessagePublish {
     private IThingsModelService thingsModelService;
     @Resource
     private JsonProtocolService jsonProtocolService;
+
     private SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(3);
 
     @Resource
@@ -368,6 +374,18 @@ public class MqttMessagePublishImpl implements IMqttMessagePublish {
         return result;
     }
 
+    /**
+     * 推送设备状态
+     *
+     * @param device 设备
+     * @param status       状态
+     */
+    public void pushDeviceStatus(Device device, DeviceStatus status) {
+        String message = "{\"status\":" + status.getType() + ",\"isShadow\":" + device.getIsShadow() + ",\"rssi\":" + device.getRssi() + "}";
+        String topic = topicsUtils.buildTopic(device.getProductId(), device.getSerialNumber(), TopicType.STATUS_POST);
+        pubMqttClient.publish(0, false, topic, message);
+    }
+
 
     /**
      * 搭建消息

springboot/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/DeviceMapper.java

@@ -2,15 +2,8 @@ package com.fastbee.iot.mapper;
 
 import com.fastbee.common.core.thingsModel.ThingsModelValuesInput;
 import com.fastbee.iot.domain.Device;
-import com.fastbee.iot.model.AuthenticateInputModel;
+import com.fastbee.iot.model.*;
-import com.fastbee.iot.model.DeviceAllShortOutput;
-import com.fastbee.iot.model.DeviceMqttVO;
-import com.fastbee.iot.model.DeviceRelateAlertLogVO;
-import com.fastbee.iot.model.DeviceShortOutput;
-import com.fastbee.iot.model.DeviceStatistic;
-import com.fastbee.iot.model.ProductAuthenticateModel;
 import com.fastbee.iot.model.ThingsModels.ThingsModelValuesOutput;
-import com.fastbee.iot.model.UserIdDeviceIdModel;
 import org.apache.ibatis.annotations.Param;
 import org.springframework.stereotype.Repository;
 
@@ -249,6 +242,12 @@ public interface DeviceMapper
      */
     public List<Device> selectDevicesByProductId(@Param("productId") Long productId,@Param("hasSub") Integer hasSub);
 
+    /**
+     * 获取所有已经激活并不是禁用的设备
+     * @return
+     */
+    List<DeviceStatusVO> selectDeviceActive();
+
     /**
      * 查询子设备总数
      * @param gwDevCode 网关编号

springboot/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceStatusVO.java

@@ -0,0 +1,26 @@
+package com.fastbee.iot.model;
+
+import lombok.Data;
+
+/**
+ * 设备状态
+ * @author gsb
+ * @date 2024/4/11 10:39
+ */
+@Data
+public class DeviceStatusVO {
+
+    private String serialNumber;
+
+    private Integer status;
+
+    private String transport;
+
+    private Long productId;
+
+    private Integer deviceType;
+
+    private Integer rssi;
+
+    private Integer isShadow;
+}

springboot/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IDeviceService.java

@@ -227,6 +227,12 @@ public interface IDeviceService
      */
     public List<Device> selectDevicesByProductId(Long productId,Integer hasSub);
 
+    /**
+     * 获取所有已经激活并不是禁用的设备
+     * @return
+     */
+    List<DeviceStatusVO> selectDeviceActive();
+
     /**
      * 查询子设备总数
      * @param gwDevCode 网关编号

springboot/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java

@@ -1082,6 +1082,16 @@ public class DeviceServiceImpl implements IDeviceService {
         return deviceMapper.selectDevicesByProductId(productId, hasSub);
     }
 
+    /**
+     * 获取所有已经激活并不是禁用的设备
+     *
+     * @return
+     */
+    @Override
+    public List<DeviceStatusVO> selectDeviceActive() {
+        return deviceMapper.selectDeviceActive();
+    }
+
     /**
      * 查询子设备总数
      *

springboot/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceMapper.xml

@@ -565,6 +565,13 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
 
     </select>
 
+    <select id="selectDeviceActive" resultType="com.fastbee.iot.model.DeviceStatusVO">
+        select d.status , d.serial_number as serialNumber, d.rssi,d.is_shadow, d.product_id
+        from iot_device d inner join iot_product p on d.product_id = p.product_id
+        where d.status in (3,4) and p.transport not in ('GB28181')
+          and d.del_flag = '0' and p.del_flag = '0'
+    </select>
+
     <select id="getSubDeviceCount" resultType="java.lang.Integer">
         select count(*) from iot_device d where d.gw_dev_code = #{gwDevCode,jdbcType=VARCHAR}
     </select>