refactor: 优化标签存储

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/entity/DeviceTagEntity.java

@@ -127,24 +127,27 @@ public class DeviceTagEntity extends GenericEntity<String> {
         return tag;
     }
 
+
     public DeviceProperty toProperty() {
         DeviceProperty property = new DeviceProperty();
         property.setProperty(getKey());
         property.setDeviceId(deviceId);
         property.setType(type);
         property.setPropertyName(name);
+        property.setValue(parseValue());
+        return property;
+    }
+
+    public Object parseValue() {
         DataType type = Optional
             .ofNullable(DataTypes.lookup(getType()))
             .map(Supplier::get)
             .orElseGet(UnknownType::new);
         if (type instanceof Converter) {
-            property.setValue(((Converter<?>) type).convert(getValue()));
+            return ((Converter<?>) type).convert(getValue());
         } else {
-            property.setValue(getValue());
+            return getValue();
         }
-        return property;
-
-
     }
 
     //以物模型标签基础数据为准，重构数据库保存的可能已过时的标签数据

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/tag/DeviceTagSynchronizer.java

@@ -4,6 +4,8 @@ import lombok.*;
 import org.apache.commons.collections4.MapUtils;
 import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
 import org.hswebframework.web.crud.events.EntityCreatedEvent;
+import org.hswebframework.web.crud.events.EntityDeletedEvent;
+import org.hswebframework.web.crud.events.EntityModifyEvent;
 import org.hswebframework.web.crud.events.EntitySavedEvent;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
@@ -79,7 +81,15 @@ public class DeviceTagSynchronizer implements CommandLineRunner {
                     tagEntity.setTimestamp(message.getTimestamp());
                     tagEntity.setDeviceId(deviceId);
                     tagEntity.setId(DeviceTagEntity.createTagId(deviceId, tagEntity.getKey()));
-                    return writeBuffer(tagEntity);
+
+                    return dataWriter
+                        .updateTag(DeviceThingType.device.getId(),
+                                   tagEntity.getDeviceId(),
+                                   tagEntity.getKey(),
+                                   System.currentTimeMillis(),
+                                   e.getValue())
+                        .then(writeBuffer(tagEntity));
+
                 }))
             .then();
     }
@@ -126,6 +136,25 @@ public class DeviceTagSynchronizer implements CommandLineRunner {
         event.async(updateTag(event.getEntity()));
     }
 
+    @EventListener
+    public void handleDeviceTagEvent(EntityModifyEvent<DeviceTagEntity> event) {
+        event.async(updateTag(event.getAfter()));
+    }
+
+    @EventListener
+    public void handleDeviceTagEvent(EntityDeletedEvent<DeviceTagEntity> event) {
+        event.async(
+            Flux
+                .fromIterable(event.getEntity())
+                .flatMap(entity -> dataWriter
+                    .removeTag(DeviceThingType.device.getId(),
+                               entity.getDeviceId(),
+                               entity.getKey())
+                    .then()
+                ));
+    }
+
+
     /**
      * 更新标签,界面上手动修改标签?
      *
@@ -145,7 +174,7 @@ public class DeviceTagSynchronizer implements CommandLineRunner {
                                entity.getDeviceId(),
                                entity.getKey(),
                                System.currentTimeMillis(),
-                               entity.getValue()))
+                               entity.parseValue()))
                 .then();
         });
     }