使用物模块数据存储策略来进行设备数据存储

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/configuration/DeviceManagerConfiguration.java

@@ -3,15 +3,18 @@ package org.jetlinks.community.device.configuration;
 import org.jetlinks.community.device.message.DeviceMessageConnector;
 import org.jetlinks.community.device.message.writer.TimeSeriesMessageWriterConnector;
 import org.jetlinks.community.device.service.data.DeviceDataService;
+import org.jetlinks.community.device.service.data.DeviceDataStorageProperties;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.device.session.DeviceSessionManager;
 import org.jetlinks.core.event.EventBus;
 import org.jetlinks.core.server.MessageHandler;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 @Configuration
+@EnableConfigurationProperties(DeviceDataStorageProperties.class)
 public class DeviceManagerConfiguration {
 
     @Bean

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

@@ -1,15 +1,17 @@
 package org.jetlinks.community.device.entity;
 
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.annotation.JSONField;
 import io.swagger.v3.oas.annotations.Hidden;
 import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
+import lombok.*;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.hswebframework.web.bean.FastBeanCopier;
 import org.jetlinks.community.device.enums.DeviceLogType;
+import org.jetlinks.community.things.data.ThingMessageLog;
+import org.springframework.util.StringUtils;
 
-import java.util.Map;
+import java.util.*;
 
 /**
  * @author bsetfeng
@@ -19,7 +21,9 @@ import java.util.Map;
 @AllArgsConstructor
 @NoArgsConstructor
 @Builder
+@Generated
 public class DeviceOperationLogEntity {
+    private static final long serialVersionUID = -6849794470754667710L;
 
     @Schema(description = "日志ID")
     private String id;
@@ -39,23 +43,36 @@ public class DeviceOperationLogEntity {
     @Schema(description = "日志内容")
     private Object content;
 
+    @Schema(description = "消息ID")
+    private String messageId;
+
     @Hidden
     private String orgId;
 
     @Schema(description = "数据时间")
     private long timestamp;
 
-    @Schema(description = "消息ID")
-    private String messageId;
+    public static DeviceOperationLogEntity of(ThingMessageLog log) {
+        DeviceOperationLogEntity messageLog = FastBeanCopier.copy(log, new DeviceOperationLogEntity());
+        messageLog.setDeviceId(log.getThingId());
+        return messageLog;
+    }
 
     public Map<String, Object> toSimpleMap() {
-        Map<String, Object> result = (Map) JSON.toJSON(this);
+        Map<String, Object> result = FastBeanCopier.copy(this, HashMap::new);
         result.put("type", type.getValue());
-        if (getContent() instanceof String) {
-            result.put("content", getContent());
+        if (content instanceof String) {
+            result.put("content", content);
         } else {
             result.put("content", JSON.toJSONString(getContent()));
         }
         return result;
     }
+
+    public DeviceOperationLogEntity generateId() {
+        if (StringUtils.isEmpty(id)) {
+            setId(DigestUtils.md5Hex(String.join("", deviceId, type.getValue(), String.valueOf(timestamp))));
+        }
+        return this;
+    }
 }

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

@@ -2,23 +2,34 @@ package org.jetlinks.community.device.entity;
 
 import io.swagger.v3.oas.annotations.Hidden;
 import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.Getter;
-import lombok.Setter;
-import org.jetlinks.community.timeseries.TimeSeriesData;
-import org.jetlinks.community.timeseries.query.AggregationData;
+import lombok.*;
+import org.hswebframework.web.bean.FastBeanCopier;
+import org.hswebframework.web.utils.DigestUtils;
 import org.jetlinks.core.message.property.ReportPropertyMessage;
 import org.jetlinks.core.metadata.Converter;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.PropertyMetadata;
-import org.jetlinks.core.metadata.types.GeoPoint;
-import org.jetlinks.core.metadata.types.NumberType;
-import org.jetlinks.core.metadata.types.ObjectType;
+import org.jetlinks.core.metadata.UnitSupported;
+import org.jetlinks.core.metadata.types.*;
+import org.jetlinks.core.metadata.unit.ValueUnit;
+import org.jetlinks.community.things.data.ThingPropertyDetail;
+import org.jetlinks.community.timeseries.TimeSeriesData;
+import org.jetlinks.community.timeseries.query.AggregationData;
+import org.springframework.util.StringUtils;
 
 import java.io.Serializable;
+import java.util.Date;
+import java.util.Optional;
+import java.util.function.Function;
 
 @Getter
 @Setter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
 public class DeviceProperty implements Serializable {
+    private static final long serialVersionUID = -1L;
+
     @Schema(description = "ID")
     private String id;
 
@@ -34,12 +45,21 @@ public class DeviceProperty implements Serializable {
     @Schema(description = "类型")
     private String type;
 
+    @Schema(description = "单位")
+    private String unit;
+
     @Hidden
     private Object numberValue;
 
     @Hidden
     private Object objectValue;
 
+    @Hidden
+    private Date timeValue;
+
+    @Hidden
+    private String stringValue;
+
     @Hidden
     private GeoPoint geoValue;
 
@@ -88,20 +108,41 @@ public class DeviceProperty implements Serializable {
             DataType type = metadata.getValueType();
             Object value = this.getValue();
             try {
-                if (type instanceof Converter) {
+                if (type instanceof NumberType) {
+                    NumberType<?> numberType = ((NumberType<?>) type);
+
+                    Number numberValue = NumberType
+                        .convertScaleNumber(value,
+                                            numberType.getScale(),
+                                            numberType.getRound(),
+                                            Function.identity());
+
+                    if (numberValue != null) {
+                        this.setValue(value = numberValue);
+                    }
+                    this.setNumberValue(numberValue);
+                } else if (type instanceof Converter) {
                     value = ((Converter<?>) type).convert(value);
                     this.setValue(value);
                 }
-                if (type instanceof NumberType) {
-                    setNumberValue(value);
-                }
                 if (type instanceof ObjectType) {
                     setObjectValue(value);
                 }
-
+                if (type instanceof GeoType && value instanceof GeoPoint) {
+                    setGeoValue(((GeoPoint) value));
+                }
+                if (type instanceof DateTimeType && value instanceof Date) {
+                    setTimeValue(((Date) value));
+                }
                 this.setFormatValue(type.format(value));
             } catch (Exception ignore) {
 
+            }
+            if (type instanceof UnitSupported) {
+                UnitSupported unitSupported = (UnitSupported) type;
+                this.setUnit(Optional.ofNullable(unitSupported.getUnit())
+                                     .map(ValueUnit::getSymbol)
+                                     .orElse(null));
             }
             this.setType(type.getType());
         }
@@ -143,4 +184,17 @@ public class DeviceProperty implements Serializable {
         return property.withProperty(metadata);
 
     }
+
+    public static DeviceProperty of(ThingPropertyDetail detail) {
+        DeviceProperty deviceProperty = FastBeanCopier.copy(detail, new DeviceProperty());
+        deviceProperty.setDeviceId(detail.getThingId());
+        return deviceProperty;
+    }
+
+    public DeviceProperty generateId() {
+        if (StringUtils.isEmpty(id)) {
+            setId(DigestUtils.md5Hex(String.join("", deviceId, property, String.valueOf(timestamp))));
+        }
+        return this;
+    }
 }

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/AbstractDeviceDataStoragePolicy.java

@@ -1,597 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import com.alibaba.fastjson.JSON;
-import com.google.common.collect.Maps;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.codec.digest.DigestUtils;
-import org.apache.commons.collections.MapUtils;
-import org.hswebframework.ezorm.core.param.TermType;
-import org.hswebframework.web.api.crud.entity.PagerResult;
-import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.hswebframework.web.id.IDGenerator;
-import org.jetlinks.community.device.entity.DeviceEvent;
-import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.device.enums.DeviceLogType;
-import org.jetlinks.community.device.events.handler.ValueTypeTranslator;
-import org.jetlinks.community.gateway.DeviceMessageUtils;
-import org.jetlinks.community.timeseries.TimeSeriesData;
-import org.jetlinks.core.device.DeviceConfigKey;
-import org.jetlinks.core.device.DeviceProductOperator;
-import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.message.DeviceLogMessage;
-import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.message.Headers;
-import org.jetlinks.core.message.event.EventMessage;
-import org.jetlinks.core.metadata.*;
-import org.jetlinks.core.metadata.types.*;
-import org.jetlinks.core.utils.DeviceMessageTracer;
-import org.jetlinks.core.utils.TimestampUtils;
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.util.function.Tuple2;
-import reactor.util.function.Tuples;
-
-import javax.annotation.Nonnull;
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.BiConsumer;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-import static org.jetlinks.community.device.service.data.StorageConstants.propertyIsIgnoreStorage;
-import static org.jetlinks.community.device.service.data.StorageConstants.propertyIsJsonStringStorage;
-import static org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric.*;
-
-/**
- * 抽象设备数据数据存储,实现一些通用的逻辑
- *
- * @author zhouhao
- * @since 1.5.0
- */
-@Slf4j
-public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStoragePolicy {
-
-    private final AtomicInteger nanoInc = new AtomicInteger();
-    protected DeviceRegistry deviceRegistry;
-    protected DeviceDataStorageProperties properties;
-
-    public AbstractDeviceDataStoragePolicy(DeviceRegistry registry,
-                                           DeviceDataStorageProperties properties) {
-        this.deviceRegistry = registry;
-        this.properties = properties;
-    }
-
-    /**
-     * 执行保存单个数据
-     *
-     * @param metric 指标ID
-     * @param data   数据
-     * @return void
-     */
-    protected abstract Mono<Void> doSaveData(String metric, TimeSeriesData data);
-
-    /**
-     * 执行保存批量数据
-     *
-     * @param metric 指标ID
-     * @param data   数据
-     * @return void
-     */
-    protected abstract Mono<Void> doSaveData(String metric, Flux<TimeSeriesData> data);
-
-    /**
-     * 设备消息转换 二元组 {deviceId, tsData}
-     *
-     * @param productId  产品ID
-     * @param message    设备属性消息
-     * @param properties 物模型属性
-     * @return 数据集合
-     * @see AbstractDeviceDataStoragePolicy#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
-     * @see AbstractDeviceDataStoragePolicy#convertPropertiesForRowPolicy(String, DeviceMessage, Map)
-     */
-    protected abstract Flux<Tuple2<String, TimeSeriesData>> convertProperties(String productId,
-                                                                              DeviceMessage message,
-                                                                              Map<String, Object> properties);
-
-    protected abstract <T> Flux<T> doQuery(String metric,
-                                           QueryParamEntity paramEntity,
-                                           Function<TimeSeriesData, T> mapper);
-
-    protected abstract <T> Mono<PagerResult<T>> doQueryPager(String metric,
-                                                             QueryParamEntity paramEntity,
-                                                             Function<TimeSeriesData, T> mapper);
-
-    /**
-     * 保存单个设备消息,为了提升性能,存储策略会对保存请求进行缓冲,达到一定条件后
-     * 再进行批量写出,具体由不同对存储策略实现。
-     * <p>
-     * 如果保存失败,在这里不会得到错误信息.
-     *
-     * @param message 设备消息
-     * @return void
-     */
-    @Nonnull
-    @Override
-    public Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message) {
-        return this
-            .convertMessageToTimeSeriesData(message)
-            .flatMap(tp2 -> doSaveData(tp2.getT1(), tp2.getT2()))
-            .then();
-    }
-
-    @Nonnull
-    @Override
-    public Mono<Void> saveDeviceMessage(@Nonnull Publisher<DeviceMessage> message) {
-        return Flux.from(message)
-                   .flatMap(this::convertMessageToTimeSeriesData)
-                   .groupBy(Tuple2::getT1, Integer.MAX_VALUE)
-                   .flatMap(group -> doSaveData(group.key(), group.map(Tuple2::getT2)))
-                   .then();
-    }
-
-    protected String createDataId(DeviceMessage message) {
-        long ts = message.getTimestamp();
-        return DigestUtils.md5Hex(String.join("_", message.getDeviceId(), String.valueOf(createUniqueNanoTime(ts))));
-    }
-
-    protected String getDeviceLogMetric(String productId) {
-        return deviceLogMetricId(productId);
-    }
-
-    protected String getDeviceEventMetric(String productId, String eventId) {
-        return deviceEventMetricId(productId, eventId);
-    }
-
-    protected Mono<Tuple2<String, TimeSeriesData>> createDeviceMessageLog(String productId,
-                                                                          DeviceMessage message,
-                                                                          BiConsumer<DeviceMessage, DeviceOperationLogEntity> logEntityConsumer) {
-        DeviceOperationLogEntity operationLog = new DeviceOperationLogEntity();
-        operationLog.setId(IDGenerator.SNOW_FLAKE_STRING.generate());
-        operationLog.setDeviceId(message.getDeviceId());
-        operationLog.setTimestamp(TimestampUtils.toMillis(message.getTimestamp()));
-        operationLog.setCreateTime(System.currentTimeMillis());
-        operationLog.setProductId(productId);
-        operationLog.setMessageId(message.getMessageId());
-        operationLog.setType(DeviceLogType.of(message));
-
-        if (null != logEntityConsumer) {
-            logEntityConsumer.accept(message, operationLog);
-        }
-        message.getHeader("log").ifPresent(operationLog::setContent);
-        return Mono.just(Tuples.of(deviceLogMetricId(productId), TimeSeriesData.of(message.getTimestamp(), operationLog
-            .toSimpleMap())));
-    }
-
-    /**
-     * 设备消息转换成时序数据 二元组 {deviceId, tsData}
-     *
-     * @param message 设备消息
-     * @return 二元组
-     */
-    protected Flux<Tuple2<String, TimeSeriesData>> convertMessageToTimeSeriesData(DeviceMessage message) {
-        boolean ignoreStorage = message.getHeaderOrDefault(Headers.ignoreStorage);
-        boolean ignoreLog = message.getHeaderOrDefault(Headers.ignoreLog);
-        if (ignoreStorage && ignoreLog) {
-            return Flux.empty();
-        }
-        DeviceMessageTracer.trace(message, "save.before");
-        String productId = (String) message.getHeader("productId").orElse("null");
-        BiConsumer<DeviceMessage, DeviceOperationLogEntity> logEntityConsumer = null;
-        List<Publisher<Tuple2<String, TimeSeriesData>>> all = new ArrayList<>(2);
-
-        //没有忽略数据存储
-        if (!ignoreStorage) {
-            //事件上报
-            if (message instanceof EventMessage) {
-                all.add(convertEventMessageToTimeSeriesData(productId, ((EventMessage) message)));
-            } else {
-                //属性相关
-                Map<String, Object> properties = DeviceMessageUtils
-                    .tryGetProperties(message)
-                    .orElseGet(Collections::emptyMap);
-                if (MapUtils.isNotEmpty(properties)) {
-                    all.add(convertProperties(productId, message, properties));
-                }
-            }
-        }
-        //日志
-        if (message instanceof DeviceLogMessage) {
-            logEntityConsumer = (msg, log) -> log.setContent(((DeviceLogMessage) msg).getLog());
-        }
-        //配置了记录日志,并且消息头里没有标记忽略日志
-        if (properties.getLog().match(message.getMessageType())
-            && !ignoreLog) {
-            if (logEntityConsumer == null) {
-                logEntityConsumer = (msg, log) -> log.setContent(msg.toJson());
-            }
-            all.add(createDeviceMessageLog(productId, message, logEntityConsumer));
-        }
-
-        return Flux.merge(all);
-    }
-
-    /**
-     * 事件消息转换成 二元组{deviceId, tsData}
-     *
-     * @param productId 产品ID
-     * @param message   事件消息
-     * @return 二元组
-     */
-    protected Mono<Tuple2<String, TimeSeriesData>> convertEventMessageToTimeSeriesData(String productId, EventMessage message) {
-
-        return deviceRegistry
-            .getProduct(productId)
-            .flatMap(product -> product
-                .getMetadata()
-                .<TimeSeriesData>handle((metadata, sink) -> {
-                    if (metadata.getEventOrNull(message.getEvent()) == null) {
-                        log.warn("产品[{}]物模型中未定义事件:{}", productId, message.getEvent());
-                        return;
-                    }
-                    Map<String, Object> data = createEventData(message, metadata);
-                    sink.next(TimeSeriesData.of(TimestampUtils.toMillis(message.getTimestamp()), data));
-                }))
-            .map(data -> Tuples.of(getDeviceEventMetric(productId, message.getEvent()), data));
-    }
-
-    protected Map<String, Object> createEventData(EventMessage message, DeviceMetadata metadata) {
-        Object value = message.getData();
-        DataType dataType = metadata
-            .getEvent(message.getEvent())
-            .map(EventMetadata::getType)
-            .orElseGet(UnknownType::new);
-        Object tempValue = ValueTypeTranslator.translator(value, dataType);
-        Map<String, Object> data;
-        if (tempValue instanceof Map) {
-            @SuppressWarnings("all")
-            Map<String, Object> mapValue = ((Map) tempValue);
-            int size = mapValue.size();
-            data = newMap(size);
-            data.putAll(mapValue);
-        } else {
-            data = newMap(16);
-            data.put("value", tempValue);
-        }
-        data.put("id", createDataId(message));
-        data.put("deviceId", message.getDeviceId());
-        data.put("createTime", System.currentTimeMillis());
-
-        return data;
-    }
-
-    @Override
-    public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceMessageLog(@Nonnull String deviceId, @Nonnull QueryParamEntity entity) {
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMap(operator -> operator.getSelfConfig(DeviceConfigKey.productId))
-            .flatMap(productId -> this
-                .doQueryPager(deviceLogMetricId(productId),
-                              entity.and("deviceId", TermType.eq, deviceId),
-                              data -> data.as(DeviceOperationLogEntity.class)
-                ))
-            .defaultIfEmpty(PagerResult.empty());
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceEvent> queryEvent(@Nonnull String deviceId,
-                                        @Nonnull String event,
-                                        @Nonnull QueryParamEntity query,
-                                        boolean format) {
-
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMap(device -> Mono.zip(device.getProduct(), device.getMetadata()))
-            .flatMapMany(tp2 -> query
-                .toQuery()
-                .where("deviceId", deviceId)
-                .execute(param -> this
-                    .doQuery(deviceEventMetricId(tp2.getT1().getId(), event),
-                             param,
-                             data -> {
-                                 DeviceEvent deviceEvent = new DeviceEvent(data.values());
-                                 if (format) {
-                                     deviceEvent.putFormat(tp2.getT2().getEventOrNull(event));
-                                 }
-                                 deviceEvent.putIfAbsent("timestamp", data.getTimestamp());
-                                 return deviceEvent;
-                             })));
-    }
-
-    @Nonnull
-    @Override
-    public Mono<PagerResult<DeviceEvent>> queryEventPage(@Nonnull String deviceId,
-                                                         @Nonnull String event,
-                                                         @Nonnull QueryParamEntity query,
-                                                         boolean format) {
-
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMap(device -> Mono.zip(device.getProduct(), device.getMetadata()))
-            .flatMap(tp2 -> query.toQuery()
-                                 .where("deviceId", deviceId)
-                                 .execute(param -> this
-                                     .doQueryPager(deviceEventMetricId(tp2.getT1().getId(), event),
-                                                   param,
-                                                   data -> {
-                                                       DeviceEvent deviceEvent = new DeviceEvent(data.values());
-                                                       if (format) {
-                                                           deviceEvent.putFormat(tp2.getT2().getEventOrNull(event));
-                                                       }
-                                                       deviceEvent.putIfAbsent("timestamp", data.getTimestamp());
-                                                       return deviceEvent;
-                                                   }))
-            );
-    }
-
-    protected Flux<DeviceProperty> rowToProperty(TimeSeriesData row, Collection<PropertyMetadata> properties) {
-        return Flux
-            .fromIterable(properties)
-            .filter(prop -> row.get(prop.getId()).isPresent())
-            .map(property -> DeviceProperty.of(
-                row,
-                row.get(property.getId()).orElse(0),
-                property
-            ).property(property.getId()));
-    }
-
-    protected Object convertPropertyValue(Object value, PropertyMetadata metadata) {
-        if (value == null || metadata == null) {
-            return value;
-        }
-        //使用json字符串来存储
-        if (propertyIsJsonStringStorage(metadata)) {
-            return value instanceof String ? String.valueOf(value) : JSON.toJSONString(value);
-        }
-        if (metadata.getValueType() instanceof Converter) {
-            return ((Converter<?>) metadata.getValueType()).convert(value);
-        }
-        return value;
-    }
-
-    protected Flux<Tuple2<String, TimeSeriesData>> convertPropertiesForColumnPolicy(String productId,
-                                                                                    DeviceMessage message,
-                                                                                    Map<String, Object> properties) {
-        if (MapUtils.isEmpty(properties)) {
-            return Flux.empty();
-        }
-        return this
-            .deviceRegistry
-            .getDevice(message.getDeviceId())
-            .flatMapMany(device -> device
-                .getMetadata()
-                .flatMap(metadata -> {
-                    int size = properties.size();
-                    String id;
-                    //强制使用时间戳作为数据ID
-                    if (message.getHeader(Headers.useTimestampAsId).orElse(false)) {
-                        id = String.join("_", message.getDeviceId(), String.valueOf(message.getTimestamp()));
-                    } else {
-                        id = createDataId(message);
-                    }
-                    Mono<Map<String, Object>> dataSupplier;
-
-                    int metaSize = metadata.getProperties().size();
-                    //标记了是部分属性
-                    if (message.getHeader(Headers.partialProperties).orElse(false)) {
-                        dataSupplier = this
-                            .queryEachOneProperties(message.getDeviceId(), QueryParamEntity.of())
-                            .collectMap(DeviceProperty::getProperty, DeviceProperty::getValue, () -> newMap(metaSize + 5));
-                    } else {
-                        dataSupplier = Mono.just(newMap(size));
-                    }
-                    return dataSupplier
-                        .flatMap(newData -> {
-                            //转换属性数据
-                            for (Map.Entry<String, Object> entry : properties.entrySet()) {
-                                PropertyMetadata propertyMetadata = metadata.getPropertyOrNull(entry.getKey());
-                                //没有配置物模型或者忽略了存储
-                                if (propertyMetadata == null || propertyIsIgnoreStorage(propertyMetadata)) {
-                                    continue;
-                                }
-                                Object value = convertPropertyValue(entry.getValue(), propertyMetadata);
-                                if (null != value) {
-                                    newData.put(entry.getKey(), value);
-                                }
-                            }
-                            //没有属性值,可能全部都配置了不存储
-                            if (newData.isEmpty()) {
-                                return Mono.empty();
-                            }
-                            newData.put("deviceId", message.getDeviceId());
-                            newData.put("productId", productId);
-                            newData.put("timestamp", TimestampUtils.toMillis(message.getTimestamp()));
-                            newData.put("createTime", System.currentTimeMillis());
-                            newData.put("id", DigestUtils.md5Hex(id));
-                            return Mono.just(
-                                Tuples.of(getPropertyTimeSeriesMetric(productId), TimeSeriesData.of(message.getTimestamp(), newData))
-                            );
-                        });
-                }));
-    }
-
-    private Map<String, Object> newMap(int size) {
-        return Maps.newHashMapWithExpectedSize(size);
-    }
-
-    /**
-     * 设备消息转换 二元组{deviceId, tsData}
-     *
-     * @param productId  产品ID
-     * @param message    设备属性消息
-     * @param properties 物模型属性
-     * @return 数据集合
-     * @see this#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
-     * @see this#convertPropertiesForRowPolicy(String, DeviceMessage, Map)
-     */
-    protected Flux<Tuple2<String, TimeSeriesData>> convertPropertiesForRowPolicy(String productId,
-                                                                                 DeviceMessage message,
-                                                                                 Map<String, Object> properties) {
-        if (MapUtils.isEmpty(properties)) {
-            return Flux.empty();
-        }
-        Map<String, Long> propertySourceTimes = DeviceMessageUtils
-            .tryGetPropertySourceTimes(message)
-            .orElseGet(Collections::emptyMap);
-        return this
-            .deviceRegistry
-            .getDevice(message.getDeviceId())
-            .flatMapMany(device -> device
-                .getMetadata()
-                .flatMapMany(metadata -> Flux
-                    .fromIterable(properties.entrySet())
-                    .index()
-                    .flatMap(entry -> {
-                        String id;
-                        String property = entry.getT2().getKey();
-                        long ts = propertySourceTimes.getOrDefault(property, message.getTimestamp());
-                        //忽略存在没有的属性和忽略存储的属性
-                        PropertyMetadata propertyMetadata = metadata.getPropertyOrNull(property);
-                        if (propertyMetadata == null || propertyIsIgnoreStorage(propertyMetadata)) {
-                            return Mono.empty();
-                        }
-                        //强制使用时间戳作为数据ID
-                        if (message.getHeader(Headers.useTimestampAsId).orElse(false)) {
-                            id = String.join("_", message.getDeviceId(), property, String.valueOf(message.getTimestamp()));
-                        } else {
-                            id = String.join("_", message.getDeviceId(), property, String.valueOf(createUniqueNanoTime(ts)));
-                        }
-                        return Mono
-                            .just(TimeSeriesData.of(ts, this
-                                .createRowPropertyData(id,
-                                                       TimestampUtils.toMillis(ts),
-                                                       device.getDeviceId(),
-                                                       propertyMetadata,
-                                                       entry.getT2().getValue()))
-                            );
-                    })
-                    .map(data -> Tuples.of(devicePropertyMetricId(productId), data)))
-            );
-    }
-
-    protected Map<String, Object> createRowPropertyData(String id,
-                                                        long timestamp,
-                                                        String deviceId,
-                                                        PropertyMetadata property,
-                                                        Object value) {
-        Map<String, Object> propertyData = newMap(24);
-        propertyData.put("id", DigestUtils.md5Hex(id));
-        propertyData.put("deviceId", deviceId);
-        propertyData.put("timestamp", timestamp);
-        propertyData.put("property", property.getId());
-        propertyData.put("createTime", System.currentTimeMillis());
-
-        fillRowPropertyValue(propertyData, property, value);
-        return propertyData;
-    }
-
-    protected void fillRowPropertyValue(Map<String, Object> target, PropertyMetadata property, Object value) {
-        if (value == null) {
-            return;
-        }
-        if (property == null) {
-            if (value instanceof Number) {
-                target.put("numberValue", value);
-            } else if (value instanceof Date) {
-                target.put("timeValue", value);
-            }
-            target.put("value", String.valueOf(value));
-            return;
-        }
-        DataType type = property.getValueType();
-        target.put("type", type.getId());
-        String convertedValue;
-        if (type instanceof NumberType) {
-            NumberType<?> numberType = (NumberType<?>) type;
-            Number number = numberType.convertNumber(value);
-            if (number == null) {
-                throw new UnsupportedOperationException("无法将" + value + "转为" + type.getId());
-            }
-            convertedValue = String.valueOf(number);
-            target.put("numberValue", number);
-        } else if (type instanceof DateTimeType) {
-            DateTimeType dateTimeType = (DateTimeType) type;
-            convertedValue = String.valueOf(value);
-            target.put("timeValue", dateTimeType.convert(convertedValue));
-        } else if (propertyIsJsonStringStorage(property)) {
-            //使用json字符来存储
-            convertedValue = value instanceof String
-                ? String.valueOf(value)
-                : JSON.toJSONString(value);
-
-        } else if (type instanceof ObjectType) {
-            ObjectType objectType = (ObjectType) type;
-            Object val = objectType.convert(value);
-            convertedValue = JSON.toJSONString(val);
-            target.put("objectValue", val);
-        } else if (type instanceof ArrayType) {
-            ArrayType objectType = (ArrayType) type;
-            Object val = objectType.convert(value);
-            convertedValue = JSON.toJSONString(val);
-            target.put("arrayValue", val);
-        } else if (type instanceof GeoType) {
-            GeoType geoType = (GeoType) type;
-            GeoPoint val = geoType.convert(value);
-            convertedValue = String.valueOf(val);
-            target.put("geoValue", val);
-        } else {
-            convertedValue = String.valueOf(value);
-        }
-        target.put("value", convertedValue);
-    }
-
-    protected String getPropertyTimeSeriesMetric(String productId) {
-        return devicePropertyMetricId(productId);
-    }
-
-    protected Mono<Tuple2<DeviceProductOperator, DeviceMetadata>> getProductAndMetadataByDevice(String deviceId) {
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMap(device -> Mono.zip(device.getProduct(), device.getMetadata()));
-    }
-
-    protected Mono<Tuple2<DeviceProductOperator, DeviceMetadata>> getProductAndMetadataByProduct(String productId) {
-        return deviceRegistry
-            .getProduct(productId)
-            .flatMap(product -> Mono.zip(Mono.just(product), product.getMetadata()));
-    }
-
-    protected List<PropertyMetadata> getPropertyMetadata(DeviceMetadata metadata, String... properties) {
-        if (properties == null || properties.length == 0) {
-            return metadata.getProperties();
-        }
-        if (properties.length == 1) {
-            return metadata.getProperty(properties[0])
-                           .map(Arrays::asList)
-                           .orElseGet(Collections::emptyList);
-        }
-        Set<String> ids = new HashSet<>(Arrays.asList(properties));
-        return metadata
-            .getProperties()
-            .stream()
-            .filter(prop -> ids.isEmpty() || ids.contains(prop.getId()))
-            .collect(Collectors.toList());
-    }
-
-    /**
-     * 将毫秒转为纳秒，努力让数据不重复
-     *
-     * @param millis 毫秒值
-     * @return 尽可能不会重复的long值
-     */
-    protected long createUniqueNanoTime(long millis) {
-        long nano = TimeUnit.MILLISECONDS.toNanos(millis);
-
-        int inc = nanoInc.incrementAndGet();
-
-        if (inc >= 99990) {
-            nanoInc.set(inc = 1);
-        }
-
-        return nano + inc;
-    }
-
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/DefaultDeviceDataService.java

@@ -1,221 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import org.hswebframework.web.api.crud.entity.PagerResult;
-import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.jetlinks.community.device.entity.DeviceEvent;
-import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.timeseries.query.AggregationData;
-import org.jetlinks.core.Value;
-import org.jetlinks.core.device.DeviceOperator;
-import org.jetlinks.core.device.DeviceProductOperator;
-import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.metadata.DeviceMetadata;
-import org.reactivestreams.Publisher;
-import org.springframework.beans.factory.ObjectProvider;
-import org.springframework.stereotype.Component;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import javax.annotation.Nonnull;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Function;
-
-/**
- * 默认设备数据服务
- * <p>
- * 管理设备存储策略、提供数据查询和入库操作
- *
- * @author zhouhao
- */
-@Component
-public class DefaultDeviceDataService implements DeviceDataService {
-
-    private final DeviceRegistry deviceRegistry;
-
-    private final Map<String, DeviceDataStoragePolicy> policies = new ConcurrentHashMap<>();
-
-    private final Mono<DeviceDataStoragePolicy> defaultPolicyMono;
-
-    private final DeviceDataStorageProperties properties;
-
-    public DefaultDeviceDataService(DeviceRegistry registry,
-                                    DeviceDataStorageProperties storeProperties,
-                                    ObjectProvider<DeviceDataStoragePolicy> policies) {
-        this.deviceRegistry = registry;
-        this.properties = storeProperties;
-        for (DeviceDataStoragePolicy policy : policies) {
-            this.policies.put(policy.getId(), policy);
-        }
-        defaultPolicyMono = Mono
-            .fromSupplier(() -> this.policies.get(properties.getDefaultPolicy()))
-            .switchIfEmpty(Mono.error(() -> new UnsupportedOperationException("存储策略[" + storeProperties.getDefaultPolicy() + "]不存在")));
-    }
-
-    @Override
-    public Mono<Void> registerMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
-        return this
-            .getStoreStrategy(productId)
-            .flatMap(policy -> policy.registerMetadata(productId, metadata))
-            .then();
-    }
-
-    /**
-     * 通过产品ID 获取存储策略
-     *
-     * @param productId 产品ID
-     * @return 存储策略
-     */
-    Mono<DeviceDataStoragePolicy> getStoreStrategy(String productId) {
-        // 从注册中心获取产品操作接口
-        // 从配置中获取产品的存储策略
-        // 巧妙的双层switchIfEmpty 外层判断空配置 内层判断空策略
-        return deviceRegistry
-            .getProduct(productId)
-            .flatMap(product -> product
-                .getConfig("storePolicy")
-                .map(Value::asString)
-                .map(conf -> Mono
-                    .justOrEmpty(policies.get(conf))
-                    .switchIfEmpty(Mono.error(() -> new UnsupportedOperationException("存储策略[" + deviceRegistry + "]不存在")))
-                ).switchIfEmpty(Mono.just(defaultPolicyMono))
-                .flatMap(Function.identity()));
-    }
-
-    /**
-     * 通过设备ID 获取存储策略
-     *
-     * @param deviceId 设备ID
-     * @return 存储策略
-     */
-    Mono<DeviceDataStoragePolicy> getDeviceStrategy(String deviceId) {
-        // 从注册中心获取设备操作接口
-        // 转换成产品操作接口
-        // 继而通过转换的产品ID获取存储策略
-        return deviceRegistry.getDevice(deviceId)
-            .flatMap(DeviceOperator::getProduct)
-            .map(DeviceProductOperator::getId)
-            .flatMap(this::getStoreStrategy);
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId,
-                                                       @Nonnull QueryParamEntity query,
-                                                       @Nonnull String... properties) {
-        return this
-            .getDeviceStrategy(deviceId)
-            .flatMapMany(strategy -> strategy.queryEachOneProperties(deviceId, query, properties));
-    }
-
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId,
-                                                    @Nonnull QueryParamEntity query,
-                                                    @Nonnull String... properties) {
-        return this
-            .getDeviceStrategy(deviceId)
-            .flatMapMany(strategy -> strategy.queryEachProperties(deviceId, query, properties));
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryProperty(@Nonnull String deviceId,
-                                              @Nonnull QueryParamEntity query,
-                                              @Nonnull String... property) {
-
-        return this
-            .getDeviceStrategy(deviceId)
-            .flatMapMany(strategy -> strategy.queryProperty(deviceId, query, property));
-    }
-
-    @Override
-    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
-                                                                @Nonnull AggregationRequest request,
-                                                                @Nonnull DevicePropertyAggregation... properties) {
-        return this
-            .getStoreStrategy(productId)
-            .flatMapMany(strategy -> strategy.aggregationPropertiesByProduct(productId, request.copy(), properties));
-    }
-
-    @Override
-    public Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId,
-                                                               @Nonnull AggregationRequest request,
-                                                               @Nonnull DevicePropertyAggregation... properties) {
-        return this
-            .getDeviceStrategy(deviceId)
-            .flatMapMany(strategy -> strategy.aggregationPropertiesByDevice(deviceId, request.copy(), properties));
-    }
-
-    @Nonnull
-    @Override
-    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
-                                                               @Nonnull String property,
-                                                               @Nonnull QueryParamEntity query) {
-        return this
-            .getDeviceStrategy(deviceId)
-            .flatMap(strategy -> strategy.queryPropertyPage(deviceId, property, query))
-            .defaultIfEmpty(PagerResult.empty());
-    }
-
-    @Override
-    public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceMessageLog(@Nonnull String deviceId, @Nonnull QueryParamEntity query) {
-        return this
-            .getDeviceStrategy(deviceId)
-            .flatMap(strategy -> strategy.queryDeviceMessageLog(deviceId, query))
-            .defaultIfEmpty(PagerResult.empty());
-    }
-
-    /**
-     * 保存单个设备消息,为了提升性能,存储策略会对保存请求进行缓冲,达到一定条件后
-     * 再进行批量写出,具体由不同对存储策略实现。
-     * <p>
-     * 如果保存失败,在这里不会得到错误信息.
-     *
-     * @param message 设备消息
-     * @return void
-     */
-    @Nonnull
-    @Override
-    public Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message) {
-        return this
-            .getDeviceStrategy(message.getDeviceId())
-            .flatMap(strategy -> strategy.saveDeviceMessage(message));
-    }
-
-    @Nonnull
-    @Override
-    public Mono<Void> saveDeviceMessage(@Nonnull Publisher<DeviceMessage> message) {
-        return Flux
-            .from(message)
-            .groupBy(DeviceMessage::getDeviceId)
-            .flatMap(group -> this
-                .getDeviceStrategy(group.key())
-                .flatMap(policy -> policy.saveDeviceMessage(group)))
-            .then();
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceEvent> queryEvent(@Nonnull String deviceId,
-                                        @Nonnull String event,
-                                        @Nonnull QueryParamEntity query, boolean format) {
-
-        return this
-            .getDeviceStrategy(deviceId)
-            .flatMapMany(strategy -> strategy.queryEvent(deviceId, event, query, format));
-    }
-
-    @Nonnull
-    @Override
-    public Mono<PagerResult<DeviceEvent>> queryEventPage(@Nonnull String deviceId, @Nonnull String event, @Nonnull QueryParamEntity query, boolean format) {
-        return this
-            .getDeviceStrategy(deviceId)
-            .flatMap(strategy -> strategy.queryEventPage(deviceId, event, query, format))
-            .defaultIfEmpty(PagerResult.empty());
-    }
-
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/DeviceDataService.java

@@ -43,6 +43,8 @@ public interface DeviceDataService {
     Mono<Void> registerMetadata(@Nonnull String productId,
                                 @Nonnull DeviceMetadata metadata);
 
+    Mono<Void> reloadMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata);
+
     /**
      * 批量保存消息
      *

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/DeviceDataStorageConfiguration.java

@@ -1,12 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Configuration;
-
-@Configuration
-@EnableConfigurationProperties(DeviceDataStorageProperties.class)
-public class DeviceDataStorageConfiguration {
-
-
-
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/DeviceDataStoragePolicy.java

@@ -1,193 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import org.hswebframework.web.api.crud.entity.PagerResult;
-import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.jetlinks.community.device.entity.DeviceEvent;
-import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.timeseries.query.AggregationData;
-import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.metadata.ConfigMetadata;
-import org.jetlinks.core.metadata.DeviceMetadata;
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import javax.annotation.Nonnull;
-
-/**
- * 设备数据存储策略
- *
- * @author zhouhao
- * @since 1.5
- */
-public interface DeviceDataStoragePolicy {
-
-    /**
-     * @return 策略唯一标识
-     */
-    String getId();
-
-    /**
-     * @return 策略名称
-     */
-    String getName();
-
-    /**
-     * @return 说明
-     */
-    String getDescription();
-
-    /**
-     * 保存单个设备消息,为了提升性能,存储策略会对保存请求进行缓冲,达到一定条件后
-     * 再进行批量写出,具体由不同对存储策略实现。
-     * <p>
-     * 如果保存失败,在这里不会得到错误信息.
-     *
-     * @param message 设备消息
-     * @return void
-     */
-    @Nonnull
-    Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message);
-
-    /**
-     * 批量保存设备消息,通常此操作会立即保存数据.如果失败也会立即得到错误信息.
-     *
-     * @param message 设备消息
-     * @return void
-     */
-    @Nonnull
-    Mono<Void> saveDeviceMessage(@Nonnull Publisher<DeviceMessage> message);
-
-    /**
-     * 获取配置信息
-     *
-     * @return 配置信息
-     */
-    @Nonnull
-    Mono<ConfigMetadata> getConfigMetadata();
-
-    /**
-     * 注册设备物模型
-     *
-     * @param productId 产品ID
-     * @param metadata  模型
-     * @return void
-     */
-    @Nonnull
-    Mono<Void> registerMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata);
-
-    /**
-     * 获取设备最新属性,Map key为属性标识，值为属性值
-     *
-     * @param deviceId   设备ID
-     * @param properties 指定设备属性标识,如果不传,则返回全部属性.
-     * @return 设备属性
-     */
-    @Nonnull
-    Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId,
-                                                @Nonnull QueryParamEntity query,
-                                                @Nonnull String... properties);
-
-    /**
-     * 查询设备事件
-     *
-     * @param deviceId 设备ID
-     * @param event    事件标识
-     * @param query    查询条件
-     * @return 设备事件数据
-     */
-    @Nonnull
-    Flux<DeviceEvent> queryEvent(@Nonnull String deviceId,
-                                 @Nonnull String event,
-                                 @Nonnull QueryParamEntity query,
-                                 boolean format);
-
-    /**
-     * 分页查询设备事件
-     *
-     * @param deviceId 设备ID
-     * @param event    事件标识
-     * @param query    查询条件
-     * @return 设备事件数据
-     */
-    @Nonnull
-    Mono<PagerResult<DeviceEvent>> queryEventPage(@Nonnull String deviceId,
-                                                  @Nonnull String event,
-                                                  @Nonnull QueryParamEntity query,
-                                                  boolean format);
-
-
-    /**
-     * 查询所有设备属性
-     *
-     * @param deviceId 设备ID
-     * @param query    查询条件
-     * @return 设备属性
-     */
-    @Nonnull
-    Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId,
-                                             @Nonnull QueryParamEntity query,
-                                             @Nonnull String... property);
-
-    /**
-     * 查询指定的设备属性列表
-     *
-     * @param deviceId 设备ID
-     * @param query    查询条件
-     * @param property 属性列表
-     * @return 设备属性
-     */
-    @Nonnull
-    Flux<DeviceProperty> queryProperty(@Nonnull String deviceId,
-                                       @Nonnull QueryParamEntity query,
-                                       @Nonnull String... property);
-
-    /**
-     * 根据产品ID聚合查询属性
-     *
-     * @param productId  产品ID
-     * @param request    聚合请求
-     * @param properties 指定聚合属性
-     * @return 聚合查询结果
-     */
-    Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
-                                                         @Nonnull DeviceDataService.AggregationRequest request,
-                                                         @Nonnull DeviceDataService.DevicePropertyAggregation... properties);
-
-    /**
-     * 根据设备ID聚合查询属性
-     *
-     * @param deviceId   设备ID
-     * @param request    聚合请求
-     * @param properties 指定聚合属性
-     * @return 聚合查询结果
-     */
-    Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId,
-                                                        @Nonnull DeviceDataService.AggregationRequest request,
-                                                        @Nonnull DeviceDataService.DevicePropertyAggregation... properties);
-
-    /**
-     * 分页查询属性
-     *
-     * @param deviceId 设备ID
-     * @param query    查询条件
-     * @return 分页查询结果
-     */
-    @Nonnull
-    Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
-                                                        @Nonnull String property,
-                                                        @Nonnull QueryParamEntity query);
-
-    /**
-     * 分页查询设备日志
-     *
-     * @param deviceId 设备ID
-     * @param query    查询条件
-     * @return 查询结果
-     */
-    Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceMessageLog(@Nonnull String deviceId,
-                                                                      @Nonnull QueryParamEntity query);
-
-
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/DeviceDataStorageProperties.java

@@ -2,24 +2,20 @@ package org.jetlinks.community.device.service.data;
 
 import lombok.Getter;
 import lombok.Setter;
-import org.jetlinks.community.utils.MessageTypeMatcher;
+import org.jetlinks.community.things.data.operations.DataSettings;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
 @ConfigurationProperties(prefix = "jetlinks.device.storage")
 @Getter
 @Setter
-public class DeviceDataStorageProperties {
+public class DeviceDataStorageProperties extends DataSettings {
 
     //默认数据存储策略,每个属性为一行数据
     private String defaultPolicy = "default-row";
 
-    private Log log = new Log();
-
-    @Getter
-    @Setter
-    public static class Log extends MessageTypeMatcher {
 
+    public Log getLog() {
+        return getLogFilter();
     }
 
-
 }

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/DeviceThingsDataCustomizer.java

@@ -0,0 +1,56 @@
+package org.jetlinks.community.device.service.data;
+
+import lombok.AllArgsConstructor;
+import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
+import org.jetlinks.community.things.data.ThingsDataContext;
+import org.jetlinks.community.things.data.ThingsDataCustomizer;
+import org.jetlinks.community.things.data.operations.MetricBuilder;
+import org.springframework.stereotype.Component;
+
+@Component
+@AllArgsConstructor
+public class DeviceThingsDataCustomizer implements ThingsDataCustomizer {
+
+    private final DeviceDataStorageProperties properties;
+
+    @Override
+    public void custom(ThingsDataContext context) {
+
+        //兼容之前版本的表名策略
+        context.customMetricBuilder(
+            ThingsBridgingDeviceDataService.thingType,
+            new MetricBuilder() {
+                @Override
+                public String getThingIdProperty() {
+                    return "deviceId";
+                }
+
+                @Override
+                public String createLogMetric(String thingType, String thingTemplateId, String thingId) {
+                    return DeviceTimeSeriesMetric.deviceLogMetricId(thingTemplateId);
+                }
+
+                @Override
+                public String createPropertyMetric(String thingType, String thingTemplateId, String thingId) {
+                    return DeviceTimeSeriesMetric.devicePropertyMetricId(thingTemplateId);
+                }
+
+                @Override
+                public String createEventAllInOneMetric(String thingType, String thingTemplateId, String thingId) {
+                    return MetricBuilder.super.createEventAllInOneMetric(thingType, thingTemplateId, thingId);
+                }
+
+                @Override
+                public String createEventMetric(String thingType, String thingTemplateId, String thingId, String eventId) {
+                    return DeviceTimeSeriesMetric.deviceEventMetricId(thingTemplateId, eventId);
+                }
+            }
+        );
+
+        context.setDefaultPolicy(properties.getDefaultPolicy());
+
+        context.customSettings(ThingsBridgingDeviceDataService.thingType,
+                               properties);
+
+    }
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/NoneDeviceDataStoragePolicy.java

@@ -1,124 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import org.hswebframework.web.api.crud.entity.PagerResult;
-import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.jetlinks.community.device.entity.DeviceEvent;
-import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.timeseries.query.AggregationData;
-import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.metadata.ConfigMetadata;
-import org.jetlinks.core.metadata.DeviceMetadata;
-import org.reactivestreams.Publisher;
-import org.springframework.stereotype.Component;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import javax.annotation.Nonnull;
-
-@Component
-public class NoneDeviceDataStoragePolicy implements DeviceDataStoragePolicy {
-    @Override
-    public String getId() {
-        return "none";
-    }
-
-    @Override
-    public String getName() {
-        return "不存储";
-    }
-
-    @Override
-    public String getDescription() {
-        return "不存储设备相关数据";
-    }
-
-    @Nonnull
-    @Override
-    public Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message) {
-        return Mono.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Mono<Void> saveDeviceMessage(@Nonnull Publisher<DeviceMessage> message) {
-        return Mono.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Mono<ConfigMetadata> getConfigMetadata() {
-        return Mono.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Mono<Void> registerMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
-        return Mono.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId,
-                                                       @Nonnull QueryParamEntity query,
-                                                       @Nonnull String... properties) {
-        return Flux.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceEvent> queryEvent(@Nonnull String deviceId,
-                                        @Nonnull String event,
-                                        @Nonnull QueryParamEntity query, boolean format) {
-        return Flux.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Mono<PagerResult<DeviceEvent>> queryEventPage(@Nonnull String deviceId,
-                                                         @Nonnull String event,
-                                                         @Nonnull QueryParamEntity query, boolean format) {
-        return Mono.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId,
-                                                    @Nonnull QueryParamEntity query,
-                                                    @Nonnull String... property) {
-        return Flux.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryProperty(@Nonnull String deviceId,
-                                              @Nonnull QueryParamEntity query,
-                                              @Nonnull String... property) {
-        return Flux.empty();
-    }
-
-    @Override
-    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
-                                                                @Nonnull DeviceDataService.AggregationRequest request, @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
-        return Flux.empty();
-    }
-
-    @Override
-    public Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId,
-                                                               @Nonnull DeviceDataService.AggregationRequest request, @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
-        return Flux.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
-                                                               @Nonnull String property, @Nonnull QueryParamEntity query) {
-        return Mono.empty();
-    }
-
-    @Override
-    public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceMessageLog(@Nonnull String deviceId,
-                                                                             @Nonnull QueryParamEntity query) {
-        return Mono.empty();
-    }
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/StorageConstants.java

@@ -1,38 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import org.jetlinks.core.metadata.PropertyMetadata;
-
-public interface StorageConstants {
-    String storePolicyConfigKey = "storePolicy";
-
-    String propertyStorageType = "storageType";
-    String propertyStorageTypeJson = "json-string";
-    String propertyStorageTypeIgnore = "ignore";
-
-    /**
-     * 判断属性是否使用json字符串来存储
-     *
-     * @param metadata 属性物模型
-     * @return 是否使用json字符串存储
-     */
-    static boolean propertyIsJsonStringStorage(PropertyMetadata metadata) {
-        return metadata
-            .getExpand(propertyStorageType)
-            .map(propertyStorageTypeJson::equals)
-            .orElse(false);
-    }
-
-    /**
-     * 判断属性是否忽略存储
-     *
-     * @param metadata 属性物模型
-     * @return 属性是否忽略存储
-     */
-    static boolean propertyIsIgnoreStorage(PropertyMetadata metadata) {
-        return metadata
-            .getExpand(propertyStorageType)
-            .map(propertyStorageTypeIgnore::equals)
-            .orElse(false);
-    }
-
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/ThingsBridgingDeviceDataService.java

@@ -0,0 +1,168 @@
+package org.jetlinks.community.device.service.data;
+
+import lombok.AllArgsConstructor;
+import org.hswebframework.web.api.crud.entity.PagerResult;
+import org.hswebframework.web.api.crud.entity.QueryParamEntity;
+import org.hswebframework.web.bean.FastBeanCopier;
+import org.jetlinks.core.device.DeviceThingType;
+import org.jetlinks.core.message.DeviceMessage;
+import org.jetlinks.core.metadata.DeviceMetadata;
+import org.jetlinks.community.device.entity.DeviceEvent;
+import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
+import org.jetlinks.community.device.entity.DeviceProperty;
+import org.jetlinks.community.things.ThingsDataRepository;
+import org.jetlinks.community.things.data.PropertyAggregation;
+import org.jetlinks.community.timeseries.query.AggregationData;
+import org.reactivestreams.Publisher;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import javax.annotation.Nonnull;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+@Component
+@AllArgsConstructor
+public class ThingsBridgingDeviceDataService implements DeviceDataService {
+    private final ThingsDataRepository repository;
+
+    static final String thingType = DeviceThingType.device.getId();
+
+    @Override
+    public Mono<Void> registerMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
+
+        return repository
+            .opsForTemplate(thingType, productId)
+            .flatMap(opt -> opt.forDDL().registerMetadata(metadata));
+    }
+
+    @Override
+    public Mono<Void> reloadMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
+        return repository
+            .opsForTemplate(thingType, productId)
+            .flatMap(opt -> opt.forDDL().reloadMetadata(metadata));
+    }
+
+    @Nonnull
+    @Override
+    public Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message) {
+        return repository.opsForSave().save(message);
+    }
+
+    @Nonnull
+    @Override
+    public Mono<Void> saveDeviceMessage(@Nonnull Publisher<DeviceMessage> message) {
+        return repository.opsForSave().save(message);
+    }
+
+    @Nonnull
+    @Override
+    public Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId, @Nonnull QueryParamEntity query, @Nonnull String... properties) {
+        return repository
+            .opsForThing(thingType, deviceId)
+            .flatMapMany(opt -> opt.forQuery().queryEachProperty(query, properties))
+            .map(DeviceProperty::of);
+    }
+
+    @Nonnull
+    @Override
+    public Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId, @Nonnull QueryParamEntity query, @Nonnull String... properties) {
+        return queryEachOneProperties(deviceId, query.clone().doPaging(0, 1), properties);
+    }
+
+    @Nonnull
+    @Override
+    public Flux<DeviceProperty> queryProperty(@Nonnull String deviceId, @Nonnull QueryParamEntity query, @Nonnull String... property) {
+        return repository
+            .opsForThing(thingType, deviceId)
+            .flatMapMany(opt -> opt.forQuery().queryProperty(query, property))
+            .map(DeviceProperty::of);
+    }
+
+    @Override
+    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
+                                                                @Nonnull AggregationRequest request,
+                                                                @Nonnull DevicePropertyAggregation... properties) {
+        return repository
+            .opsForTemplate(thingType, productId)
+            .flatMapMany(opt -> opt.forQuery()
+                                   .aggregationProperties(
+                                       FastBeanCopier.copy(request, new org.jetlinks.community.things.data.AggregationRequest()),
+                                       Stream
+                                           .of(properties)
+                                           .map(prop -> FastBeanCopier.copy(prop, new PropertyAggregation()))
+                                           .toArray(PropertyAggregation[]::new)));
+    }
+
+    @Override
+    public Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId, @Nonnull AggregationRequest request, @Nonnull DevicePropertyAggregation... properties) {
+        return repository
+            .opsForThing(thingType, deviceId)
+            .flatMapMany(opt -> opt
+                .forQuery()
+                .aggregationProperties(
+                    FastBeanCopier.copy(request, new org.jetlinks.community.things.data.AggregationRequest()),
+                    Stream
+                        .of(properties)
+                        .map(prop -> FastBeanCopier.copy(prop, new PropertyAggregation()))
+                        .toArray(PropertyAggregation[]::new)));
+    }
+
+    @Nonnull
+    @Override
+    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId, @Nonnull String property, @Nonnull QueryParamEntity query) {
+        return queryPropertyPage(deviceId, query, property);
+    }
+
+    @Nonnull
+    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId, @Nonnull QueryParamEntity query, @Nonnull String... property) {
+        return repository
+            .opsForThing(thingType, deviceId)
+            .flatMap(opt -> opt.forQuery().queryPropertyPage(query, property))
+            .map(page -> convertPage(page, DeviceProperty::of));
+    }
+
+    private <R, T> PagerResult<R> convertPage(PagerResult<T> source, Function<T, R> mapper) {
+        @SuppressWarnings("all")
+        PagerResult<R> newResult = FastBeanCopier.copy(source, source.getClass());
+
+        newResult.setData(
+            source.getData()
+                  .stream()
+                  .map(mapper)
+                  .collect(Collectors.toList())
+        );
+        return newResult;
+    }
+
+
+    @Override
+    public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceMessageLog(@Nonnull String deviceId, @Nonnull QueryParamEntity query) {
+        return repository
+            .opsForThing(thingType, deviceId)
+            .flatMap(opt -> opt.forQuery().queryMessageLogPage(query))
+            .map(page -> convertPage(page,DeviceOperationLogEntity::of));
+    }
+
+
+    @Nonnull
+    @Override
+    public Flux<DeviceEvent> queryEvent(@Nonnull String deviceId, @Nonnull String event, @Nonnull QueryParamEntity query, boolean format) {
+        return repository
+            .opsForThing(thingType, deviceId)
+            .flatMapMany(opt -> opt.forQuery().queryEvent(event, query, format))
+            .map(DeviceEvent::new);
+    }
+
+    @Nonnull
+    @Override
+    public Mono<PagerResult<DeviceEvent>> queryEventPage(@Nonnull String deviceId, @Nonnull String event, @Nonnull QueryParamEntity query, boolean format) {
+        return repository
+            .opsForThing(thingType, deviceId)
+            .flatMap(opt -> opt.forQuery().queryEventPage(event, query, format))
+            .map(page -> convertPage(page,DeviceEvent::new));
+    }
+
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/TimeSeriesColumnDeviceDataStoragePolicy.java

@@ -1,293 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import org.hswebframework.web.api.crud.entity.PagerResult;
-import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetadata;
-import org.jetlinks.community.timeseries.TimeSeriesData;
-import org.jetlinks.community.timeseries.TimeSeriesManager;
-import org.jetlinks.community.timeseries.query.*;
-import org.jetlinks.core.device.DeviceOperator;
-import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.metadata.ConfigMetadata;
-import org.jetlinks.core.metadata.Converter;
-import org.jetlinks.core.metadata.DeviceMetadata;
-import org.jetlinks.core.metadata.PropertyMetadata;
-import org.joda.time.DateTime;
-import org.joda.time.format.DateTimeFormat;
-import org.springframework.stereotype.Component;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.util.function.Tuple2;
-
-import javax.annotation.Nonnull;
-import java.util.*;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import static org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric.devicePropertyMetric;
-
-/**
- * 时序数据列存储策略
- *
- * @author zhouhao
- */
-@Component
-public class TimeSeriesColumnDeviceDataStoragePolicy extends TimeSeriesDeviceDataStoragePolicy implements DeviceDataStoragePolicy {
-
-    public TimeSeriesColumnDeviceDataStoragePolicy(DeviceRegistry deviceRegistry,
-                                                   TimeSeriesManager timeSeriesManager,
-                                                   DeviceDataStorageProperties properties) {
-        super(deviceRegistry, timeSeriesManager, properties);
-    }
-
-    @Override
-    public String getId() {
-        return "default-column";
-    }
-
-    @Override
-    public String getName() {
-        return "默认-列式存储";
-    }
-
-    @Override
-    public String getDescription() {
-        return "每个设备的全部属性为一行数据.需要设备每次上报全部属性.";
-    }
-
-    @Nonnull
-    @Override
-    public Mono<ConfigMetadata> getConfigMetadata() {
-        return Mono.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Mono<Void> registerMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
-        return Flux
-            .concat(Flux
-                    .fromIterable(metadata.getEvents())
-                    .flatMap(event -> timeSeriesManager.registerMetadata(DeviceTimeSeriesMetadata.event(productId, event))),
-                timeSeriesManager.registerMetadata(DeviceTimeSeriesMetadata.properties(productId, metadata.getProperties())),
-                timeSeriesManager.registerMetadata(DeviceTimeSeriesMetadata.log(productId)))
-            .then();
-    }
-
-
-    private Flux<DeviceProperty> queryEachDeviceProperty(String productId,
-                                                         String deviceId,
-                                                         Map<String, PropertyMetadata> property,
-                                                         QueryParamEntity param) {
-        //查询多个属性,分组聚合获取第一条数据
-        return param
-            .toQuery()
-            .includes(property.keySet().toArray(new String[0]))
-            .where("deviceId", deviceId)
-            .execute(q -> timeSeriesManager.getService(getPropertyTimeSeriesMetric(productId)).query(q))
-            .flatMap(data -> rowToProperty(data, property.values()));
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId,
-                                                       @Nonnull QueryParamEntity query,
-                                                       @Nonnull String... properties) {
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMapMany(device -> Mono
-                .zip(device.getProduct(), device.getMetadata())
-                .flatMapMany(tp2 -> {
-
-                    Map<String, PropertyMetadata> propertiesMap = (properties.length == 0
-                        ? tp2.getT2().getProperties().stream()
-                        : Stream.of(properties).map(tp2.getT2()::getPropertyOrNull).filter(Objects::nonNull))
-                        .collect(Collectors.toMap(PropertyMetadata::getId, Function.identity(), (a, b) -> a));
-
-                    return queryEachDeviceProperty(tp2.getT1().getId(), deviceId, propertiesMap, query
-                        .clone()
-                        .doPaging(0, 1));
-                }));
-    }
-
-    @Nonnull
-    @Override
-    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
-                                                               @Nonnull String property,
-                                                               @Nonnull QueryParamEntity param) {
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMap(device -> Mono.zip(device.getProduct(), device.getMetadata()))
-            .flatMap(tp2 -> {
-                    PropertyMetadata prop = tp2.getT2().getPropertyOrNull(property);
-
-                    return param
-                        .toQuery()
-                        .includes(property)
-                        .where("deviceId", deviceId)
-                        .execute(query -> timeSeriesManager
-                            .getService(devicePropertyMetric(tp2.getT1().getId()))
-                            .queryPager(query,
-                                data -> DeviceProperty
-                                    .of(data, data.get(property).orElse(0), prop)
-                                    .property(property)
-                            ));
-                }
-            );
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryProperty(@Nonnull String deviceId,
-                                              @Nonnull QueryParamEntity query,
-                                              @Nonnull String... property) {
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMapMany(device -> Mono
-                .zip(device.getProduct(), device.getMetadata())
-                .flatMapMany(tp2 -> {
-                    Set<String> includes = new HashSet<>(Arrays.asList(property));
-                    Map<String, PropertyMetadata> propertiesMap = tp2
-                        .getT2()
-                        .getProperties()
-                        .stream()
-                        .filter(prop -> includes.size() > 0 && includes.contains(prop.getId()))
-                        .collect(Collectors.toMap(PropertyMetadata::getId, Function
-                            .identity(), (a, b) -> a));
-
-                    return query
-                        .toQuery()
-                        .where("deviceId", deviceId)
-                        .includes(property)
-                        .execute(timeSeriesManager.getService(getPropertyTimeSeriesMetric(tp2.getT1().getId()))::query)
-                        .flatMap(data -> Flux
-                            .fromIterable(propertiesMap.entrySet())
-                            .map(entry -> DeviceProperty.of(
-                                data,
-                                data.get(entry.getKey()).orElse(null),
-                                entry.getValue()
-                            ).property(entry.getKey()))
-                        );
-                }));
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId,
-                                                    @Nonnull QueryParamEntity query,
-                                                    @Nonnull String... property) {
-
-        return this
-            .getProductAndMetadataByDevice(deviceId)
-            .flatMapMany(tp2 -> {
-
-                Map<String, PropertyMetadata> propertiesMap = getPropertyMetadata(tp2.getT2(), property)
-                    .stream()
-                    .collect(Collectors.toMap(PropertyMetadata::getId, Function.identity(), (a, b) -> a));
-
-                return queryEachDeviceProperty(tp2.getT1().getId(), deviceId, propertiesMap, query);
-            });
-    }
-
-
-    @Override
-    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
-                                                                @Nonnull DeviceDataService.AggregationRequest request,
-                                                                @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
-        org.joda.time.format.DateTimeFormatter formatter = DateTimeFormat.forPattern(request.getFormat());
-
-        return AggregationQueryParam
-            .of()
-            .as(param -> {
-                for (DeviceDataService.DevicePropertyAggregation property : properties) {
-                    param.agg(property.getProperty(), property.getAlias(), property.getAgg());
-                }
-                return param;
-            })
-            .as(param -> {
-                if (request.interval == null) {
-                    return param;
-                }
-                return param.groupBy((Group) new TimeGroup(request.interval, "time", request.format));
-            })
-            .limit(request.limit * properties.length)
-            .from(request.from)
-            .to(request.to)
-            .filter(request.filter)
-            .execute(timeSeriesManager.getService(getPropertyTimeSeriesMetric(productId))::aggregation)
-            .groupBy(agg -> agg.getString("time", ""), Integer.MAX_VALUE)
-            .flatMap(group -> group
-                .map(data -> {
-                    Map<String, Object> newMap = new HashMap<>();
-                    newMap.put("time", data.get("time").orElse(null));
-                    for (DeviceDataService.DevicePropertyAggregation property : properties) {
-                        Object val;
-                        if(property.getAgg() == Aggregation.FIRST || property.getAgg()==Aggregation.TOP){
-                            val = data
-                                .get(property.getProperty())
-                                .orElse(null);
-                        }else {
-                            val = data
-                                .get(property.getAlias())
-                                .orElse(null);
-                        }
-                        if (null != val) {
-                            newMap.put(property.getAlias(), val);
-                        }
-                    }
-                    return newMap;
-                })
-                .reduce((a, b) -> {
-                    a.putAll(b);
-                    return a;
-                })
-                .map(AggregationData::of))
-            .sort(Comparator.<AggregationData, Date>comparing(agg -> DateTime
-                .parse(agg.getString("time", ""), formatter)
-                .toDate()).reversed())
-            .take(request.getLimit())
-            ;
-    }
-
-    @Override
-    public Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId,
-                                                               @Nonnull DeviceDataService.AggregationRequest request,
-                                                               @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
-
-        request.filter.and("deviceId", "eq", deviceId);
-
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMap(DeviceOperator::getProduct)
-            .flatMapMany(product -> aggregationPropertiesByProduct(product.getId(), request, properties))
-            .doOnNext(agg -> agg.values().remove("_time"));
-    }
-
-    /**
-     * 设备消息转换 二元组{deviceId, tsData}
-     *
-     * @param productId  产品ID
-     * @param message    设备属性消息
-     * @param properties 物模型属性
-     * @return 数据集合
-     * @see this#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
-     * @see this#convertPropertiesForRowPolicy(String, DeviceMessage, Map)
-     */
-    @Override
-    protected Flux<Tuple2<String, TimeSeriesData>> convertProperties(String productId, DeviceMessage message, Map<String, Object> properties) {
-        return convertPropertiesForColumnPolicy(productId, message, properties);
-    }
-
-    @Override
-    protected Object convertPropertyValue(Object value, PropertyMetadata metadata) {
-        if (value == null || metadata == null) {
-            return value;
-        }
-        if (metadata instanceof Converter) {
-            return ((Converter<?>) metadata).convert(value);
-        }
-        return value;
-    }
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/TimeSeriesDeviceDataStoragePolicy.java

@@ -1,66 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import org.hswebframework.web.api.crud.entity.PagerResult;
-import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.jetlinks.community.timeseries.TimeSeriesData;
-import org.jetlinks.community.timeseries.TimeSeriesManager;
-import org.jetlinks.core.device.DeviceRegistry;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import java.util.function.Function;
-
-/**
- * 抽象时序数据存储策略
- * <p>
- * 提供时序数据通用的查询存储逻辑
- * </p>
- *
- * @author zhouhao
- */
-public abstract class TimeSeriesDeviceDataStoragePolicy extends AbstractDeviceDataStoragePolicy {
-
-
-    protected TimeSeriesManager timeSeriesManager;
-
-    public TimeSeriesDeviceDataStoragePolicy(DeviceRegistry registry,
-                                             TimeSeriesManager timeSeriesManager,
-                                             DeviceDataStorageProperties properties) {
-        super(registry, properties);
-        this.timeSeriesManager = timeSeriesManager;
-    }
-
-    @Override
-    protected Mono<Void> doSaveData(String metric, TimeSeriesData data) {
-        return timeSeriesManager
-            .getService(metric)
-            .commit(data);
-    }
-
-    @Override
-    protected Mono<Void> doSaveData(String metric, Flux<TimeSeriesData> data) {
-        return timeSeriesManager
-            .getService(metric)
-            .save(data);
-    }
-
-    @Override
-    protected <T> Flux<T> doQuery(String metric,
-                                  QueryParamEntity paramEntity,
-                                  Function<TimeSeriesData, T> mapper) {
-        return timeSeriesManager
-            .getService(metric)
-            .query(paramEntity)
-            .map(mapper);
-    }
-
-
-    @Override
-    protected <T> Mono<PagerResult<T>> doQueryPager(String metric,
-                                                    QueryParamEntity paramEntity,
-                                                    Function<TimeSeriesData, T> mapper) {
-        return timeSeriesManager
-            .getService(metric)
-            .queryPager(paramEntity, mapper);
-    }
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/TimeSeriesRowDeviceDataStoreStoragePolicy.java

@@ -1,356 +0,0 @@
-package org.jetlinks.community.device.service.data;
-
-import org.hswebframework.web.api.crud.entity.PagerResult;
-import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetadata;
-import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
-import org.jetlinks.community.timeseries.TimeSeriesData;
-import org.jetlinks.community.timeseries.TimeSeriesManager;
-import org.jetlinks.community.timeseries.query.*;
-import org.jetlinks.core.device.DeviceOperator;
-import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.metadata.ConfigMetadata;
-import org.jetlinks.core.metadata.DeviceMetadata;
-import org.jetlinks.core.metadata.PropertyMetadata;
-import org.jetlinks.reactor.ql.utils.CastUtils;
-import org.springframework.stereotype.Component;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.util.function.Tuple2;
-
-import javax.annotation.Nonnull;
-import java.util.*;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import static org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric.devicePropertyMetric;
-
-/**
- * 设备时序数据行存储策略
- *
- * @author zhouhao
- */
-@Component
-public class TimeSeriesRowDeviceDataStoreStoragePolicy extends TimeSeriesDeviceDataStoragePolicy implements DeviceDataStoragePolicy {
-
-    public TimeSeriesRowDeviceDataStoreStoragePolicy(DeviceRegistry deviceRegistry,
-                                                     TimeSeriesManager timeSeriesManager,
-                                                     DeviceDataStorageProperties properties) {
-        super(deviceRegistry, timeSeriesManager, properties);
-    }
-
-    @Override
-    public String getId() {
-        return "default-row";
-    }
-
-    @Override
-    public String getName() {
-        return "默认-行式存储";
-    }
-
-    @Override
-    public String getDescription() {
-        return "每个设备的每一个属性为一行数据.适合设备每次上报部分属性.";
-    }
-
-    @Nonnull
-    @Override
-    public Mono<ConfigMetadata> getConfigMetadata() {
-        return Mono.empty();
-    }
-
-    @Nonnull
-    @Override
-    public Mono<Void> registerMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
-        return Flux
-            .concat(Flux
-                    .fromIterable(metadata.getEvents())
-                    .flatMap(event -> timeSeriesManager.registerMetadata(DeviceTimeSeriesMetadata.event(productId, event))),
-                timeSeriesManager.registerMetadata(DeviceTimeSeriesMetadata.properties(productId)),
-                timeSeriesManager.registerMetadata(DeviceTimeSeriesMetadata.log(productId)))
-            .then();
-    }
-
-    private Flux<DeviceProperty> queryEachDeviceProperty(String productId,
-                                                         String deviceId,
-                                                         Map<String, PropertyMetadata> property,
-                                                         QueryParamEntity param) {
-        if (property.isEmpty()) {
-            return Flux.empty();
-        }
-        //只查询一个属性
-        if (property.size() == 1) {
-            return param
-                .toQuery()
-                .where("deviceId", deviceId)
-                .and("property", property.keySet().iterator().next())
-                .execute(timeSeriesManager.getService(devicePropertyMetric(productId))::query)
-                .map(data ->
-                    DeviceProperty
-                        .of(data, data.getString("property").map(property::get).orElse(null))
-                        .deviceId(deviceId));
-        }
-
-        //查询多个属性,分组聚合获取第一条数据
-        return timeSeriesManager
-            .getService(devicePropertyMetric(productId))
-            .aggregation(AggregationQueryParam
-                .of()
-                .agg("property", Aggregation.FIRST)
-                .groupBy(new LimitGroup("property", "property", property.size() * 2)) //按property分组
-                .limit(property.size())
-                .filter(param)
-                .filter(query -> query.where("deviceId", deviceId))
-            ).map(data -> DeviceProperty
-                .of(data, data.getString("property").map(property::get).orElse(null))
-                .deviceId(deviceId));
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId,
-                                                       @Nonnull QueryParamEntity query,
-                                                       @Nonnull String... properties) {
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMapMany(device -> Mono
-                .zip(device.getProduct(), device.getMetadata())
-                .flatMapMany(tp2 -> {
-
-                    Map<String, PropertyMetadata> propertiesMap = (properties.length == 0
-                        ? tp2.getT2().getProperties().stream()
-                        : Stream.of(properties).map(tp2.getT2()::getPropertyOrNull).filter(Objects::nonNull))
-                        .collect(Collectors.toMap(PropertyMetadata::getId, Function.identity(), (a, b) -> a));
-
-                    return queryEachDeviceProperty(tp2.getT1().getId(), deviceId, propertiesMap, query.clone().doPaging(0, 1));
-                }));
-    }
-
-    @Nonnull
-    @Override
-    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
-                                                               @Nonnull String property,
-                                                               @Nonnull QueryParamEntity param) {
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMap(device -> Mono.zip(device.getProduct(), device.getMetadata()))
-            .flatMap(tp2 -> param.toQuery()
-                .where("property", property)
-                .and("deviceId", deviceId)
-                .execute(query -> timeSeriesManager
-                    .getService(devicePropertyMetric(tp2.getT1().getId()))
-                    .queryPager(query, data -> DeviceProperty.of(data, tp2.getT2().getPropertyOrNull(property)))));
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryProperty(@Nonnull String deviceId,
-                                              @Nonnull QueryParamEntity query,
-                                              @Nonnull String... property) {
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMapMany(device -> Mono
-                .zip(device.getProduct(), device.getMetadata())
-                .flatMapMany(tp2 -> {
-
-                    Map<String, PropertyMetadata> propertiesMap = tp2.getT2()
-                        .getProperties()
-                        .stream()
-                        .collect(Collectors.toMap(PropertyMetadata::getId, Function.identity(), (a, b) -> a));
-
-                    return query.toQuery()
-                        .where("deviceId", deviceId)
-                        .when(property.length > 0, q -> q.in("property", Arrays.asList(property)))
-                        .execute(timeSeriesManager
-                            .getService(DeviceTimeSeriesMetric.devicePropertyMetricId(tp2.getT1().getId()))::query)
-                        .map(data -> DeviceProperty.of(data, propertiesMap.get(data.getString("property", null))));
-                }));
-    }
-
-    @Nonnull
-    @Override
-    public Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId,
-                                                    @Nonnull QueryParamEntity query,
-                                                    @Nonnull String... property) {
-
-        return getProductAndMetadataByDevice(deviceId)
-            .flatMapMany(tp2 -> {
-
-                Map<String, PropertyMetadata> propertiesMap = getPropertyMetadata(tp2.getT2(), property)
-                    .stream()
-                    .collect(Collectors.toMap(PropertyMetadata::getId, Function.identity(), (a, b) -> a));
-                if (propertiesMap.isEmpty()) {
-                    return Flux.empty();
-                }
-                return timeSeriesManager
-                    .getService(devicePropertyMetric(tp2.getT1().getId()))
-                    .aggregation(AggregationQueryParam
-                        .of()
-                        .agg(new LimitAggregationColumn("property", "property", Aggregation.TOP, query.getPageSize()))
-                        .groupBy(new LimitGroup("property", "property", propertiesMap.size() * 2)) //按property分组
-                        .filter(query)
-                        .filter(q -> q.where("deviceId", deviceId).in("property", propertiesMap.keySet()))
-                    ).map(data -> DeviceProperty
-                        .of(data, data.getString("property").map(propertiesMap::get).orElse(null))
-                        .deviceId(deviceId));
-            });
-    }
-
-    protected String getTimeSeriesMetric(String productId) {
-        return DeviceTimeSeriesMetric.devicePropertyMetricId(productId);
-    }
-
-    @Override
-    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
-                                                                @Nonnull DeviceDataService.AggregationRequest request,
-                                                                @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
-        //只聚合一个属性时
-        if (properties.length == 1) {
-            return AggregationQueryParam.of()
-                .agg("numberValue", properties[0].getAlias(), properties[0].getAgg())
-                .groupBy(request.interval, request.format)
-                .limit(request.limit)
-                .from(request.from)
-                .to(request.to)
-                .filter(request.filter)
-                .filter(query -> query.where("property", properties[0].getProperty()))
-                .execute(timeSeriesManager.getService(getTimeSeriesMetric(productId))::aggregation)
-                .doOnNext(agg -> agg.values().remove("_time"));
-        }
-
-        Map<String, String> propertyAlias = Arrays.stream(properties)
-            .collect(Collectors.toMap(DeviceDataService.DevicePropertyAggregation::getAlias,
-                                      DeviceDataService.DevicePropertyAggregation::getProperty));
-
-        Map<String, DeviceDataService.DevicePropertyAggregation> aliasProperty = Arrays
-            .stream(properties)
-            .collect(Collectors.toMap(DeviceDataService.DevicePropertyAggregation::getAlias,
-                                      Function.identity()));
-
-        return AggregationQueryParam
-            .of()
-            .as(param -> {
-                Arrays.stream(properties)
-                      .forEach(agg -> param.agg("numberValue", "value_" + agg.getAlias(), agg.getAgg()));
-                return param;
-            })
-            .as(param -> {
-                if (request.interval == null) {
-                    return param;
-                }
-                return param.groupBy((Group) new TimeGroup(request.interval, "time", request.format));
-            })
-            .groupBy(new LimitGroup("property", "property", properties.length))
-            .limit(request.limit * properties.length)
-            .from(request.from)
-            .to(request.to)
-            .filter(request.filter)
-            .filter(query -> query
-                .where()
-                .in("property", new HashSet<>(propertyAlias.values())))
-            //执行查询
-            .execute(timeSeriesManager.getService(getTimeSeriesMetric(productId))::aggregation)
-            //按时间分组,然后将返回的结果合并起来
-            .groupBy(agg -> agg.getString("time", ""), Integer.MAX_VALUE)
-            .as(flux -> {
-                //按时间分组
-                if (request.getInterval() != null) {
-                    return flux
-                        .flatMap(group -> {
-                                     String time = group.key();
-                                     return group
-                                         //按属性分组
-                                         .groupBy(agg -> agg.getString("property", ""), Integer.MAX_VALUE)
-                                         .flatMap(propsGroup -> {
-                                             String property = String.valueOf(propsGroup.key());
-                                             return propsGroup
-                                                 .reduce(AggregationData::merge)
-                                                 .map(agg -> {
-                                                     Map<String, Object> data = new HashMap<>();
-                                                     data.put("_time", agg.get("_time").orElse(time));
-                                                     data.put("time", time);
-                                                     aliasProperty.forEach((alias, prp) -> {
-                                                         if (prp.getAgg() == Aggregation.FIRST || prp.getAgg() == Aggregation.TOP) {
-                                                             data.putIfAbsent(alias, agg
-                                                                 .get("numberValue")
-                                                                 .orElse(agg.get("value").orElse(null)));
-                                                         } else if (property.equals(prp.getProperty())) {
-                                                             data.putIfAbsent(alias, agg
-                                                                 .get("value_" + alias)
-                                                                 .orElse(0));
-                                                         }
-                                                     });
-                                                     return data;
-                                                 });
-                                         })
-                                         .<Map<String, Object>>reduceWith(HashMap::new, (a, b) -> {
-                                             a.putAll(b);
-                                             return a;
-                                         });
-                                 }
-                        );
-                } else {
-                    return flux
-                        .flatMap(group -> group
-                            .reduce(AggregationData::merge)
-                            .map(agg -> {
-                                Map<String, Object> values = new HashMap<>();
-                                //values.put("time", group.key());
-                                for (Map.Entry<String, String> props : propertyAlias.entrySet()) {
-                                    values.put(props.getKey(), agg
-                                        .get("value_" + props.getKey())
-                                        .orElse(0));
-                                }
-                                return values;
-                            }));
-                }
-            })
-            .map(map -> {
-                map.remove("");
-                propertyAlias
-                    .keySet()
-                    .forEach(key -> map.putIfAbsent(key, 0));
-                return AggregationData.of(map);
-            })
-            .sort(Comparator.<AggregationData, Date>comparing(agg -> CastUtils.castDate(agg
-                                                                                            .values()
-                                                                                            .get("_time")))
-                            .reversed())
-            .doOnNext(agg -> agg.values().remove("_time"))
-            .take(request.getLimit())
-            ;
-    }
-
-    @Override
-    public Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId,
-                                                               @Nonnull DeviceDataService.AggregationRequest request,
-                                                               @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
-
-        request.filter.and("deviceId", "eq", deviceId);
-
-        return deviceRegistry
-            .getDevice(deviceId)
-            .flatMap(DeviceOperator::getProduct)
-            .flatMapMany(product -> aggregationPropertiesByProduct(product.getId(), request, properties))
-            .doOnNext(agg -> agg.values().remove("_time"));
-    }
-
-    /**
-     * 设备消息转换 二元组{deviceId, tsData}
-     *
-     * @param productId  产品ID
-     * @param message    设备属性消息
-     * @param properties 物模型属性
-     * @return 数据集合
-     * @see this#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
-     * @see this#convertPropertiesForRowPolicy(String, DeviceMessage, Map)
-     */
-    @Override
-    protected Flux<Tuple2<String, TimeSeriesData>> convertProperties(String productId, DeviceMessage message, Map<String, Object> properties) {
-        return convertPropertiesForRowPolicy(productId, message, properties);
-    }
-}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/DeviceProductController.java

@@ -15,8 +15,8 @@ import org.jetlinks.community.device.entity.DeviceProductEntity;
 import org.jetlinks.community.device.service.DeviceConfigMetadataManager;
 import org.jetlinks.community.device.service.LocalDeviceProductService;
 import org.jetlinks.community.device.service.data.DeviceDataService;
-import org.jetlinks.community.device.service.data.DeviceDataStoragePolicy;
 import org.jetlinks.community.device.web.request.AggRequest;
+import org.jetlinks.community.things.data.ThingsDataRepositoryStrategy;
 import org.jetlinks.community.timeseries.query.AggregationData;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DeviceConfigScope;
@@ -39,8 +39,7 @@ public class DeviceProductController implements ReactiveServiceCrudController<De
 
     private final LocalDeviceProductService productService;
 
-    private final List<DeviceDataStoragePolicy> policies;
-
+    private final List<ThingsDataRepositoryStrategy> policies;
     private final DeviceDataService deviceDataService;
 
     private final DeviceConfigMetadataManager configMetadataManager;
@@ -50,7 +49,7 @@ public class DeviceProductController implements ReactiveServiceCrudController<De
     private final DeviceMetadataCodec defaultCodec = new JetLinksDeviceMetadataCodec();
 
     public DeviceProductController(LocalDeviceProductService productService,
-                                   List<DeviceDataStoragePolicy> policies,
+                                   List<ThingsDataRepositoryStrategy> policies,
                                    DeviceDataService deviceDataService,
                                    DeviceConfigMetadataManager configMetadataManager,
                                    ObjectProvider<DeviceMetadataCodec> metadataCodecs) {
@@ -141,7 +140,7 @@ public class DeviceProductController implements ReactiveServiceCrudController<De
     @Operation(summary = "获取支持的数据存储策略")
     public Flux<DeviceDataStorePolicyInfo> storePolicy() {
         return Flux.fromIterable(policies)
-            .flatMap(DeviceDataStorePolicyInfo::of);
+            .map(DeviceDataStorePolicyInfo::of);
     }
 
     @PostMapping("/{productId:.+}/agg/_query")
@@ -173,10 +172,8 @@ public class DeviceProductController implements ReactiveServiceCrudController<De
 
         private ConfigMetadata configMetadata;
 
-        public static Mono<DeviceDataStorePolicyInfo> of(DeviceDataStoragePolicy policy) {
-            return policy.getConfigMetadata()
-                .map(metadata -> new DeviceDataStorePolicyInfo(policy.getId(), policy.getName(), policy.getDescription(), metadata))
-                .defaultIfEmpty(new DeviceDataStorePolicyInfo(policy.getId(), policy.getName(), policy.getDescription(), null));
+        public static DeviceDataStorePolicyInfo of(ThingsDataRepositoryStrategy strategy) {
+            return new DeviceDataStorePolicyInfo(strategy.getId(), strategy.getName(), null, null);
         }
     }