fix: 设备上报数据过滤

server-web/src/main/java/com/makeit/mqtt/PushCallback.java

@@ -199,22 +199,35 @@ public class PushCallback implements MqttCallbackExtended {
         Long timestamp = device.getTimestamp();
         Long timestamp = device.getTimestamp();
         long currentSecond = timestamp / 1000;
         long currentSecond = timestamp / 1000;
         //  先通过产品名称判断
         //  先通过产品名称判断
+        JSONObject properties = device.getProperties();
+        if (properties == null) {
+            return;
+        }
         if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("呼吸")) {
         if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("呼吸")) {
+            if (!properties.containsKey("hr") || !properties.containsKey("br")) {
+                return;
+            }
             cacheBrDeviceData(device, currentSecond);
             cacheBrDeviceData(device, currentSecond);
 
 
-            redisTemplate.opsForValue().set(DEVICE_BR_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()),
+            redisTemplate.opsForValue().set(DEVICE_BR_DATA + device.getDeviceId(),JSON.toJSONString(properties),
                     5000, TimeUnit.MILLISECONDS);
                     5000, TimeUnit.MILLISECONDS);
         }
         }
         if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("空间")) {
         if (REPORT_PROPERTY.equals(device.getMessageType()) && headers.getProductName().contains("空间")) {
+            if (!properties.containsKey("personState")) {
+                return;
+            }
             cacheSpaceFallDeviceData(DEVICE_SPACE_TEMP_DATA, device, currentSecond);
             cacheSpaceFallDeviceData(DEVICE_SPACE_TEMP_DATA, device, currentSecond);
 
 
-            redisTemplate.opsForValue().set(DEVICE_SPACE_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()),
+            redisTemplate.opsForValue().set(DEVICE_SPACE_DATA + device.getDeviceId(),JSON.toJSONString(properties),
                     5000, TimeUnit.MILLISECONDS);
                     5000, TimeUnit.MILLISECONDS);
         }
         }
         if (REPORT_PROPERTY.equals(device.getMessageType())  && headers.getProductName().contains("跌倒")) {
         if (REPORT_PROPERTY.equals(device.getMessageType())  && headers.getProductName().contains("跌倒")) {
+            if (!properties.containsKey("personState")) {
+                return;
+            }
             cacheSpaceFallDeviceData(DEVICE_FALL_TEMP_DATA, device, currentSecond);
             cacheSpaceFallDeviceData(DEVICE_FALL_TEMP_DATA, device, currentSecond);
 
 
-            redisTemplate.opsForValue().set(DEVICE_FALL_DATA + device.getDeviceId(),JSON.toJSONString(device.getProperties()),
+            redisTemplate.opsForValue().set(DEVICE_FALL_DATA + device.getDeviceId(),JSON.toJSONString(properties),
                     5000, TimeUnit.MILLISECONDS);
                     5000, TimeUnit.MILLISECONDS);
         }
         }
         String deviceId = device.getDeviceId();
         String deviceId = device.getDeviceId();
@@ -226,7 +239,6 @@ public class PushCallback implements MqttCallbackExtended {
         for (HeaderInfo.Bind binding : bindings) {
         for (HeaderInfo.Bind binding : bindings) {
             try {
             try {
                 String iot_tenantId = binding.getId();
                 String iot_tenantId = binding.getId();
-                JSONObject properties = device.getProperties();
 
 
                 // 设备上报到第三方，一分钟一次
                 // 设备上报到第三方，一分钟一次
                 reportBrData(device, iot_tenantId);
                 reportBrData(device, iot_tenantId);