增加设备存储策略功能

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

@@ -72,9 +72,6 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
     @Autowired
     private EventBus eventBus;
 
-    @Autowired
-    private TimeSeriesManager timeSeriesManager;
-
     @Autowired
     @SuppressWarnings("all")
     private ReactiveRepository<DeviceTagEntity, String> tagRepository;
@@ -191,6 +188,22 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                     .execute()));
     }
 
+    /**
+     * 注销设备,取消后,设备无法再连接到服务. 注册中心也无法再获取到该设备信息.
+     *
+     * @param id 设备ID
+     * @return 注销结果
+     */
+    public Mono<Integer> unregisterDevice(String id) {
+        return this.findById(Mono.just(id))
+            .flatMap(device -> registry
+                .unregisterDevice(id)
+                .then(createUpdate()
+                    .set(DeviceInstanceEntity::getState, DeviceState.notActive.getValue())
+                    .where(DeviceInstanceEntity::getId, id)
+                    .execute()));
+    }
+
     /**
      * 批量注销设备
      *
@@ -259,78 +272,6 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
             .defaultIfEmpty(DeviceState.notActive);
     }
 
-    public Mono<PagerResult<DevicePropertiesEntity>> queryDeviceProperties(String deviceId, QueryParamEntity entity) {
-        return registry.getDevice(deviceId)
-            .flatMap(operator -> operator.getSelfConfig(DeviceConfigKey.productId))
-            .flatMap(productId -> timeSeriesManager
-                .getService(devicePropertyMetric(productId))
-                .queryPager(entity.and("deviceId", TermType.eq, deviceId), data -> data.as(DevicePropertiesEntity.class)))
-            .defaultIfEmpty(PagerResult.empty());
-    }
-
-    public Mono<PagerResult<Map<String, Object>>> queryDeviceEvent(String deviceId, String eventId, QueryParamEntity entity, boolean format) {
-        return registry
-            .getDevice(deviceId)
-            .flatMap(operator -> operator.getSelfConfig(DeviceConfigKey.productId).zipWith(operator.getMetadata()))
-            .flatMap(tp -> timeSeriesManager
-                .getService(DeviceTimeSeriesMetric.deviceEventMetric(tp.getT1(), eventId))
-                .queryPager(entity.and("deviceId", TermType.eq, deviceId), data -> {
-                    if (!format) {
-                        return data.getData();
-                    }
-                    Map<String, Object> formatData = new HashMap<>(data.getData());
-                    tp.getT2()
-                        .getEvent(eventId)
-                        .ifPresent(eventMetadata -> {
-                            DataType type = eventMetadata.getType();
-                            if (type instanceof ObjectType) {
-                                @SuppressWarnings("all")
-                                Map<String, Object> val = (Map<String, Object>) type.format(formatData);
-                                val.forEach((k, v) -> formatData.put(k + "_format", v));
-                            } else {
-                                formatData.put("value_format", type.format(data.get("value")));
-                            }
-                        });
-                    return formatData;
-                })).defaultIfEmpty(PagerResult.empty());
-    }
-
-    public Mono<DevicePropertiesEntity> getDeviceLatestProperty(String deviceId, String property) {
-        return registry
-            .getDevice(deviceId)
-            .flatMap(operator -> operator.getSelfConfig(DeviceConfigKey.productId))
-            .flatMap(productId -> doGetLatestDeviceProperty(productId, deviceId, property));
-    }
-
-    public Flux<DevicePropertiesEntity> getDeviceLatestProperties(String deviceId) {
-        return registry.getDevice(deviceId)
-            .flatMap(operator -> Mono.zip(operator.getMetadata(), operator.getSelfConfig(DeviceConfigKey.productId)))
-            .flatMapMany(zip -> Flux.merge(zip.getT1().getProperties()
-                .stream()
-                .map(property -> doGetLatestDeviceProperty(zip.getT2(), deviceId, property.getId()))
-                .collect(Collectors.toList())));
-    }
-
-    private Mono<DevicePropertiesEntity> doGetLatestDeviceProperty(String productId, String deviceId, String property) {
-        return Query.of()
-            .and(DevicePropertiesEntity::getDeviceId, deviceId)
-            .and(DevicePropertiesEntity::getProperty, property)
-            .doPaging(0, 1)
-            .execute(timeSeriesManager.getService(devicePropertyMetric(productId))::query)
-            .map(data -> data.as(DevicePropertiesEntity.class))
-            .singleOrEmpty();
-    }
-
-    public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceLog(String deviceId, QueryParamEntity entity) {
-        return registry.getDevice(deviceId)
-            .flatMap(operator -> operator.getSelfConfig(DeviceConfigKey.productId))
-            .flatMap(productId -> timeSeriesManager
-                .getService(DeviceTimeSeriesMetric.deviceLogMetric(productId))
-                .queryPager(entity.and("deviceId", TermType.eq, deviceId),
-                    data -> data.as(DeviceOperationLogEntity.class)))
-            .defaultIfEmpty(PagerResult.empty());
-    }
-
     @PostConstruct
     public void init() {
 
@@ -358,12 +299,6 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
             .subscribe((i) -> log.info("同步设备状态成功:{}", i));
     }
 
-    public Mono<DeviceInfo> getDeviceInfoById(String id) {
-        return findById(Mono.justOrEmpty(id))
-            .zipWhen(instance -> deviceProductService
-                .findById(Mono.justOrEmpty(instance.getProductId())), DeviceInfo::of)
-            .switchIfEmpty(Mono.error(NotFoundException::new));
-    }
 
     public Flux<List<DeviceStateInfo>> syncStateBatch(Flux<List<String>> batch, boolean force) {
 
@@ -594,6 +529,38 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
         return Mono.empty();
     }
 
+    //保存标签
+    @Subscribe("/device/*/*/tags/update")
+    public Mono<Void> updateDeviceTag(UpdateTagMessage message) {
+        Map<String, Object> tags = message.getTags();
+        String deviceId = message.getDeviceId();
+
+        return registry
+            .getDevice(deviceId)
+            .flatMap(DeviceOperator::getMetadata)
+            .flatMapMany(metadata ->
+                Flux.fromIterable(tags.entrySet())
+                    .map(e -> {
+                        DeviceTagEntity tagEntity =
+                            metadata.getTag(e.getKey())
+                                .map(DeviceTagEntity::of)
+                                .orElseGet(() -> {
+                                    DeviceTagEntity entity = new DeviceTagEntity();
+                                    entity.setKey(e.getKey());
+                                    entity.setType("string");
+                                    entity.setName(e.getKey());
+                                    entity.setCreateTime(new Date());
+                                    entity.setDescription("设备上报");
+                                    return entity;
+                                });
+                        tagEntity.setDeviceId(deviceId);
+                        tagEntity.setId(DeviceTagEntity.createTagId(deviceId, tagEntity.getKey()));
+                        return tagEntity;
+                    }))
+            .as(tagRepository::save)
+            .then();
+    }
+
 
 
 }

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

@@ -0,0 +1,397 @@
+package org.jetlinks.community.device.service.data;
+
+import com.alibaba.fastjson.JSON;
+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.core.device.DeviceConfigKey;
+import org.jetlinks.core.device.DeviceProductOperator;
+import org.jetlinks.core.device.DeviceRegistry;
+import org.jetlinks.core.message.DeviceMessage;
+import org.jetlinks.core.message.DeviceMessageReply;
+import org.jetlinks.core.message.event.EventMessage;
+import org.jetlinks.core.message.property.ReadPropertyMessageReply;
+import org.jetlinks.core.message.property.ReportPropertyMessage;
+import org.jetlinks.core.message.property.WritePropertyMessageReply;
+import org.jetlinks.core.metadata.*;
+import org.jetlinks.core.metadata.types.UnknownType;
+import org.jetlinks.community.device.entity.DeviceEvent;
+import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
+import org.jetlinks.community.device.entity.DevicePropertiesEntity;
+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.device.timeseries.DeviceTimeSeriesMetric;
+import org.jetlinks.community.timeseries.TimeSeriesData;
+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.Consumer;
+import java.util.function.Function;
+
+/**
+ * 抽象设备数据数据存储,实现一些通用的逻辑
+ *
+ * @author zhouhao
+ * @since 1.5.0
+ */
+public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStoragePolicy {
+
+    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);
+
+    /**
+     * @param productId  产品ID
+     * @param message    原始消息
+     * @param properties 属性
+     * @return 数据集合
+     * @see this#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
+     * @see this#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);
+
+
+    @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)
+            .flatMap(group -> doSaveData(group.key(), group.map(Tuple2::getT2)))
+            .then();
+    }
+
+    protected String createDataId(DeviceMessage message) {
+        long ts = message.getTimestamp();
+        return String.join("_", message.getDeviceId(), String.valueOf(createUniqueNanoTime(ts)));
+    }
+
+    protected Mono<Tuple2<String, TimeSeriesData>> createDeviceMessageLog(String productId,
+                                                                          DeviceMessage message,
+                                                                          Consumer<DeviceOperationLogEntity> logEntityConsumer) {
+        DeviceOperationLogEntity operationLog = new DeviceOperationLogEntity();
+        operationLog.setId(IDGenerator.SNOW_FLAKE_STRING.generate());
+        operationLog.setDeviceId(message.getDeviceId());
+        operationLog.setCreateTime(message.getTimestamp());
+        operationLog.setProductId(productId);
+        operationLog.setType(DeviceLogType.of(message));
+
+        if (null != logEntityConsumer) {
+            logEntityConsumer.accept(operationLog);
+        }
+        message.getHeader("log").ifPresent(operationLog::setContent);
+        return Mono.just(Tuples.of(DeviceTimeSeriesMetric.deviceLogMetricId(productId), TimeSeriesData.of(message.getTimestamp(), operationLog.toSimpleMap())));
+    }
+
+    protected Flux<Tuple2<String, TimeSeriesData>> convertMessageToTimeSeriesData(DeviceMessage message) {
+        String productId = (String) message.getHeader("productId").orElse("null");
+        Consumer<DeviceOperationLogEntity> logEntityConsumer = null;
+        List<Publisher<Tuple2<String, TimeSeriesData>>> all = new ArrayList<>();
+
+        if (message instanceof EventMessage) {
+            logEntityConsumer = log -> log.setContent(JSON.toJSONString(((EventMessage) message).getData()));
+            all.add(convertEventMessageToTimeSeriesData(productId, ((EventMessage) message)));
+        }
+        //上报属性
+        else if (message instanceof ReportPropertyMessage) {
+            ReportPropertyMessage reply = (ReportPropertyMessage) message;
+            Map<String, Object> properties = reply.getProperties();
+            if (MapUtils.isNotEmpty(properties)) {
+                logEntityConsumer = log -> log.setContent(properties);
+                all.add(convertProperties(productId, message, properties));
+            }
+        }
+        //消息回复
+        else if (message instanceof DeviceMessageReply) {
+            //失败的回复消息
+            if (!((DeviceMessageReply) message).isSuccess()) {
+                logEntityConsumer = log -> log.setContent(message.toString());
+            } else if (message instanceof ReadPropertyMessageReply) {
+                ReadPropertyMessageReply reply = (ReadPropertyMessageReply) message;
+                Map<String, Object> properties = reply.getProperties();
+                logEntityConsumer = log -> log.setContent(properties);
+                all.add(convertProperties(productId, message, properties));
+            } else if (message instanceof WritePropertyMessageReply) {
+                WritePropertyMessageReply reply = (WritePropertyMessageReply) message;
+                Map<String, Object> properties = reply.getProperties();
+                logEntityConsumer = log -> log.setContent(properties);
+                all.add(convertProperties(productId, message, properties));
+            } else {
+                logEntityConsumer = log -> log.setContent(message.toJson().toJSONString());
+            }
+        }
+        //其他
+        else {
+            logEntityConsumer = log -> log.setContent(message.toJson().toJSONString());
+        }
+        //配置了记录日志
+        if (properties.getLog().match(message.getMessageType())) {
+            all.add(createDeviceMessageLog(productId, message, logEntityConsumer));
+        }
+
+        return Flux.merge(all);
+    }
+
+    protected Mono<Tuple2<String, TimeSeriesData>> convertEventMessageToTimeSeriesData(String productId, EventMessage message) {
+
+        return deviceRegistry
+            .getDevice(message.getDeviceId())
+            .flatMap(device -> device.getMetadata()
+                .map(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 = new HashMap<>((int) ((size / 0.75) + 7));
+                        data.putAll(mapValue);
+                    } else {
+                        data = new HashMap<>();
+                        data.put("value", tempValue);
+                    }
+                    data.put("id", createDataId(message));
+                    data.put("deviceId", device.getDeviceId());
+                    data.put("createTime", System.currentTimeMillis());
+
+                    return TimeSeriesData.of(message.getTimestamp(), data);
+                }))
+            .map(data -> Tuples.of(DeviceTimeSeriesMetric.deviceEventMetricId(productId, message.getEvent()), data));
+    }
+
+
+    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(DeviceTimeSeriesMetric.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(DeviceTimeSeriesMetric.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(DeviceTimeSeriesMetric.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;
+        }
+        if (metadata instanceof Converter) {
+            return ((Converter<?>) metadata).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()
+                .map(metadata -> {
+                    int size = properties.size();
+
+                    Map<String, Object> newData = new HashMap<>(size < 5 ? 16 : (int) ((size + 5) / 0.75D) + 1);
+                    properties.forEach((k, v) -> newData.put(k, convertPropertyValue(v, metadata.getPropertyOrNull(k))));
+                    newData.put("deviceId", message.getDeviceId());
+                    newData.put("productId", productId);
+                    newData.put("timestamp", message.getTimestamp());
+                    newData.put("createTime", System.currentTimeMillis());
+                    newData.put("id", createDataId(message));
+                    return Tuples.of(getPropertyTimeSeriesMetric(productId), TimeSeriesData.of(message.getTimestamp(), newData));
+                }));
+    }
+
+    protected Flux<Tuple2<String, TimeSeriesData>> convertPropertiesForRowPolicy(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()
+                .flatMapMany(metadata -> Flux
+                    .fromIterable(properties.entrySet())
+                    .index()
+                    .map(entry -> {
+                        long ts = message.getTimestamp() + entry.getT1();
+                        String id = String.join("_", message.getDeviceId(), String.valueOf(createUniqueNanoTime(ts)));
+                        DevicePropertiesEntity entity = DevicePropertiesEntity.builder()
+                            .id(id)
+                            .deviceId(device.getDeviceId())
+                            .timestamp(ts)
+                            .property(entry.getT2().getKey())
+                            .productId(productId)
+                            .createTime(System.currentTimeMillis())
+                            .build()
+                            .withValue(metadata.getPropertyOrNull(entry.getT2().getKey()), entry.getT2().getValue());
+
+                        return TimeSeriesData.of(entity.getTimestamp(), entity.toMap());
+                    })
+                    .map(data -> Tuples.of(DeviceTimeSeriesMetric.devicePropertyMetricId(productId), data)))
+            );
+    }
+
+    protected String getPropertyTimeSeriesMetric(String productId) {
+        return DeviceTimeSeriesMetric.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()));
+    }
+
+
+    private final AtomicInteger nanoInc = new AtomicInteger();
+
+    //将毫秒转为纳秒，努力让数据不重复
+    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

@@ -0,0 +1,188 @@
+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.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.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.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;
+
+@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();
+    }
+
+    Mono<DeviceDataStoragePolicy> getStoreStrategy(String productId) {
+
+        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()));
+    }
+
+    Mono<DeviceDataStoragePolicy> getDeviceStrategy(String deviceId) {
+        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) {
+        return this
+            .getDeviceStrategy(deviceId)
+            .flatMapMany(strategy -> strategy.queryEachProperties(deviceId, query));
+    }
+
+    @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));
+    }
+
+    @Override
+    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
+                                                                @Nonnull AggregationRequest request,
+                                                                @Nonnull DevicePropertyAggregation... properties) {
+        return this
+            .getStoreStrategy(productId)
+            .flatMapMany(strategy -> strategy.aggregationPropertiesByProduct(productId, request, 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, 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());
+    }
+
+
+    @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

@@ -0,0 +1,277 @@
+package org.jetlinks.community.device.service.data;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.*;
+import org.hswebframework.web.api.crud.entity.PagerResult;
+import org.hswebframework.web.api.crud.entity.QueryParamEntity;
+import org.jetlinks.core.message.DeviceMessage;
+import org.jetlinks.core.metadata.DeviceMetadata;
+import org.jetlinks.core.metadata.EventMetadata;
+import org.jetlinks.community.Interval;
+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.Aggregation;
+import org.jetlinks.community.timeseries.query.AggregationData;
+import org.joda.time.DateTime;
+import org.reactivestreams.Publisher;
+import org.springframework.util.StringUtils;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import javax.annotation.Nonnull;
+import java.util.Collection;
+import java.util.Date;
+import java.util.Map;
+
+/**
+ * 设备数据服务
+ *
+ * @author zhouhao
+ * @since 1.5
+ */
+public interface DeviceDataService {
+
+    /**
+     * 注册设备物模型信息
+     *
+     * @param productId 产品ID
+     * @param metadata  物模型
+     * @return void
+     */
+    Mono<Void> registerMetadata(@Nonnull String productId,
+                                @Nonnull DeviceMetadata metadata);
+
+    /**
+     * 批量保存消息
+     *
+     * @param message 设备消息
+     * @return void
+     * @see this#saveDeviceMessage(Publisher)
+     */
+    @Nonnull
+    default Mono<Void> saveDeviceMessage(@Nonnull Collection<DeviceMessage> message) {
+        return saveDeviceMessage(Flux.fromIterable(message));
+    }
+
+    /**
+     * 保存单个设备消息,为了提升性能,存储策略会对保存请求进行缓冲,达到一定条件后
+     * 再进行批量写出,具体由不同对存储策略实现。
+     * <p>
+     * 如果保存失败,在这里不会得到错误信息.
+     *
+     * @param message 设备消息
+     * @return void
+     */
+    @Nonnull
+    Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message);
+
+    /**
+     * 批量保存设备消息,通常此操作会立即保存数据.如果失败也会立即得到错误信息.
+     *
+     * @param message 设备消息
+     * @return void
+     */
+    @Nonnull
+    Mono<Void> saveDeviceMessage(@Nonnull Publisher<DeviceMessage> message);
+
+    /**
+     * 获取设备每个属性,只取一个结果.
+     *
+     * @param deviceId   设备ID
+     * @param properties 指定设备属性标识,如果不传,则返回全部属性.
+     * @return 设备属性
+     */
+    @Nonnull
+    Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId,
+                                                @Nonnull QueryParamEntity query,
+                                                @Nonnull String... properties);
+
+
+    /**
+     * 查询设备每个属性,可指定通过{@link QueryParamEntity#setPageSize(int)} () }每个属性的数量.
+     *
+     * @param deviceId 设备ID
+     * @param query    查询条件
+     * @return 设备属性
+     */
+    @Nonnull
+    Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId,
+                                             @Nonnull QueryParamEntity query);
+
+    /**
+     * 查询指定的设备属性列表
+     *
+     * @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 AggregationRequest request,
+                                                         @Nonnull DevicePropertyAggregation... properties);
+
+    /**
+     * 根据设备ID聚合查询属性
+     *
+     * @param deviceId   设备ID
+     * @param request    聚合请求
+     * @param properties 指定聚合属性,不指定是聚合所有属性
+     * @return 聚合查询结果
+     */
+    Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId,
+                                                        @Nonnull AggregationRequest request,
+                                                        @Nonnull 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);
+
+
+    /**
+     * 查询设备事件,如果设置里format为true，将根据物模型对数据进行{@link org.jetlinks.core.metadata.DataType#format(Object)}.
+     * 并将format后对值添加_format后缀设置到结果中.例如:
+     * <pre>
+     *     {
+     *         "value":26.5,
+     *         "value_format":"26.5℃"
+     *     }
+     * </pre>
+     * <p>
+     * 如果类型是结构体({@link org.jetlinks.core.metadata.types.ObjectType})类型,
+     * 则会把对整个数据进行格式化后合并到{@link DeviceEvent#putAll(Map)}
+     *
+     * @param deviceId 设备ID
+     * @param event    事件标识
+     * @param query    查询条件
+     * @param format   是否对数据进行格式化
+     * @return 设备事件数据
+     * @see DeviceEvent#putFormat(EventMetadata)
+     */
+    @Nonnull
+    Flux<DeviceEvent> queryEvent(@Nonnull String deviceId,
+                                 @Nonnull String event,
+                                 @Nonnull QueryParamEntity query,
+                                 boolean format);
+
+    /**
+     * 分页查询设备事件数据
+     *
+     * @param deviceId 设备ID
+     * @param event    事件ID
+     * @param query    查询条件
+     * @param format   是否对数据进行格式化
+     * @return 分页查询结果
+     */
+    @Nonnull
+    Mono<PagerResult<DeviceEvent>> queryEventPage(@Nonnull String deviceId,
+                                                  @Nonnull String event,
+                                                  @Nonnull QueryParamEntity query,
+                                                  boolean format);
+
+    /**
+     * 分页查询设备事件
+     *
+     * @param deviceId 设备ID
+     * @param event    事件标识
+     * @param query    查询条件
+     * @return 设备事件数据
+     */
+    @Nonnull
+    default Mono<PagerResult<DeviceEvent>> queryEventPage(@Nonnull String deviceId,
+                                                          @Nonnull String event,
+                                                          @Nonnull QueryParamEntity query) {
+        return queryEventPage(deviceId, event, query, false);
+    }
+
+
+    @Getter
+    @Setter
+    @AllArgsConstructor
+    @NoArgsConstructor
+    class DevicePropertyAggregation {
+        @Schema(description = "属性ID")
+        private String property; //要聚合对字段
+
+        @Schema(description = "别名,默认和property一致")
+        private String alias; //别名
+
+        @Schema(description = "聚合方式,支持(count,sum,max,min,avg)", type = "string")
+        private Aggregation agg; //聚合函数
+
+        public String getAlias() {
+            if (StringUtils.isEmpty(alias)) {
+                return property;
+            }
+            return alias;
+        }
+    }
+
+    @Getter
+    @Setter
+    @Builder
+    @AllArgsConstructor
+    @NoArgsConstructor
+    class AggregationRequest {
+        //时间间隔
+        @Schema(description = "间隔,如: 1d", type = "string", defaultValue = "1d")
+        Interval interval = Interval.ofDays(1);
+        //时间格式
+        @Schema(defaultValue = "时间格式,如:yyyy-MM-dd", description = "yyyy-MM-dd")
+        String format = "yyyy-MM-dd";
+
+        @Schema(description = "时间从,如: 2020-09-01 00:00:00,支持表达式: now-1d")
+        Date from = new DateTime()
+            .plusMonths(-1)
+            .withHourOfDay(0)
+            .withMinuteOfHour(0)
+            .withSecondOfMinute(0)
+            .toDate();
+
+        @Schema(description = "时间到,如: 2020-09-30 00:00:00,支持表达式: now-1d")
+        Date to = new DateTime()
+            .withHourOfDay(23)
+            .withMinuteOfHour(59)
+            .withSecondOfMinute(59)
+            .toDate();
+
+        @Schema(description = "实例限制")
+        int limit = 30;
+
+        //过滤条件
+        @Schema(description = "过滤条件")
+        QueryParamEntity filter = QueryParamEntity.of();
+    }
+}

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

@@ -0,0 +1,12 @@
+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

@@ -0,0 +1,183 @@
+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.core.message.DeviceMessage;
+import org.jetlinks.core.metadata.ConfigMetadata;
+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.timeseries.query.AggregationData;
+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 {
+
+    String getId();
+
+    String getName();
+
+    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);
+
+    /**
+     * 查询指定的设备属性列表
+     *
+     * @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

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

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

@@ -0,0 +1,239 @@
+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.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.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.AggregationData;
+import org.jetlinks.community.timeseries.query.AggregationQueryParam;
+import org.jetlinks.community.timeseries.query.Group;
+import org.jetlinks.community.timeseries.query.TimeGroup;
+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;
+
+@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)
+                        .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) {
+
+        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 queryEachDeviceProperty(tp2.getT1().getId(), deviceId, propertiesMap, query);
+                }));
+    }
+
+
+    @Override
+    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
+                                                                @Nonnull DeviceDataService.AggregationRequest request,
+                                                                @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
+
+        return AggregationQueryParam.of()
+            .as(param -> {
+                for (DeviceDataService.DevicePropertyAggregation property : properties) {
+                    param.agg(property.getProperty(), property.getAlias(), property.getAgg());
+                }
+                return param;
+            })
+            .groupBy((Group) new TimeGroup(request.interval, "time", request.format))
+            .limit(request.limit)
+            .from(request.from)
+            .to(request.to)
+            .filter(request.filter)
+            .execute(timeSeriesManager.getService(getPropertyTimeSeriesMetric(productId))::aggregation)
+            .doOnNext(agg -> agg.values().remove("_time"))
+            ;
+    }
+
+    @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"));
+    }
+
+    @Override
+    protected Flux<Tuple2<String, TimeSeriesData>> convertProperties(String productId, DeviceMessage message, Map<String, Object> properties) {
+        return convertPropertiesForColumnPolicy(productId, message, properties);
+    }
+
+    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

@@ -0,0 +1,54 @@
+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;
+
+public abstract class TimeSeriesDeviceDataStoragePolicy extends AbstractDeviceDataStoragePolicy {
+
+
+    protected TimeSeriesManager timeSeriesManager;
+
+    public TimeSeriesDeviceDataStoragePolicy(DeviceRegistry registry,
+                                             TimeSeriesManager timeSeriesManager,
+                                             DeviceDataStorageProperties properties) {
+        super(registry, properties);
+        this.timeSeriesManager = timeSeriesManager;
+    }
+
+    protected Mono<Void> doSaveData(String metric, TimeSeriesData data) {
+        return timeSeriesManager
+            .getService(metric)
+            .commit(data);
+    }
+
+    protected Mono<Void> doSaveData(String metric, Flux<TimeSeriesData> data) {
+        return timeSeriesManager
+            .getService(metric)
+            .save(data);
+    }
+
+    protected <T> Flux<T> doQuery(String metric,
+                                  QueryParamEntity paramEntity,
+                                  Function<TimeSeriesData, T> mapper) {
+        return timeSeriesManager
+            .getService(metric)
+            .query(paramEntity)
+            .map(mapper);
+    }
+
+
+    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

@@ -0,0 +1,295 @@
+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.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.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.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;
+
+@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) {
+
+        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));
+                    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))
+                        ).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));
+
+        return AggregationQueryParam.of()
+            .as(param -> {
+                Arrays.stream(properties)
+                    .forEach(agg -> param.agg("numberValue", "value_" + agg.getAlias(), agg.getAgg()));
+                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", propertyAlias.values()))
+            //执行查询
+            .execute(timeSeriesManager.getService(getTimeSeriesMetric(productId))::aggregation)
+            //按时间分组,然后将返回的结果合并起来
+            .groupBy(agg -> agg.getString("time", ""))
+            .flatMap(group ->
+                {
+                    String time = group.key();
+                    return group
+                        //按属性分组
+                        .groupBy(agg -> agg.getString("property", ""))
+                        .flatMap(propsGroup -> {
+                            String property = propsGroup.key();
+                            return propsGroup
+                                .<Map<String, Object>>reduceWith(HashMap::new, (a, b) -> {
+                                    a.putIfAbsent("time", time);
+                                    a.putIfAbsent("_time", b.get("_time").orElseGet(Date::new));
+                                    b.get("value_" + property).ifPresent(v -> a.put(property, v));
+                                    return a;
+                                });
+                        })
+                        .<Map<String, Object>>reduceWith(HashMap::new, (a, b) -> {
+                            a.putAll(b);
+                            return a;
+                        });
+                }
+            )
+            .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"))
+            ;
+    }
+
+    @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"));
+    }
+
+    @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/timeseries/DeviceTimeSeriesMetadata.java

@@ -2,6 +2,9 @@ package org.jetlinks.community.device.timeseries;
 
 import org.jetlinks.core.metadata.EventMetadata;
 import org.jetlinks.community.timeseries.TimeSeriesMetadata;
+import org.jetlinks.core.metadata.PropertyMetadata;
+
+import java.util.List;
 
 /**
  * 设备相关时序数据库元数据定义
@@ -21,6 +24,9 @@ public interface DeviceTimeSeriesMetadata {
         return new DevicePropertiesTimeSeriesMetadata(productId);
     }
 
+    static TimeSeriesMetadata properties(String productId, List<PropertyMetadata> properties) {
+        return new FixedPropertiesTimeSeriesMetadata(productId, properties);
+    }
     /**
      * 获取设备事件时序数据元数据
      *

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/timeseries/FixedPropertiesTimeSeriesMetadata.java

@@ -0,0 +1,71 @@
+package org.jetlinks.community.device.timeseries;
+
+import org.jetlinks.core.metadata.PropertyMetadata;
+import org.jetlinks.core.metadata.SimplePropertyMetadata;
+import org.jetlinks.core.metadata.types.DateTimeType;
+import org.jetlinks.core.metadata.types.StringType;
+import org.jetlinks.community.timeseries.TimeSeriesMetadata;
+import org.jetlinks.community.timeseries.TimeSeriesMetric;
+
+import java.util.ArrayList;
+import java.util.List;
+
+class FixedPropertiesTimeSeriesMetadata implements TimeSeriesMetadata {
+
+    private final static List<PropertyMetadata> metadata = new ArrayList<>();
+
+    private final TimeSeriesMetric metric;
+
+    private final List<PropertyMetadata> fixed;
+
+    public FixedPropertiesTimeSeriesMetadata(String productId, List<PropertyMetadata> fixed) {
+        this.metric = DeviceTimeSeriesMetric.devicePropertyMetric(productId);
+        this.fixed = new ArrayList<>(fixed);
+        this.fixed.addAll(metadata);
+    }
+
+    static {
+
+        {
+            SimplePropertyMetadata property = new SimplePropertyMetadata();
+            property.setId("id");
+            property.setValueType(StringType.GLOBAL);
+            property.setName("id");
+            metadata.add(property);
+        }
+
+        {
+            SimplePropertyMetadata property = new SimplePropertyMetadata();
+            property.setId("deviceId");
+            property.setValueType(new StringType());
+            property.setName("设备ID");
+            metadata.add(property);
+        }
+
+        {
+            SimplePropertyMetadata property = new SimplePropertyMetadata();
+            property.setId("productId");
+            property.setValueType(new StringType());
+            property.setName("产品ID");
+            metadata.add(property);
+        }
+
+        {
+            SimplePropertyMetadata property = new SimplePropertyMetadata();
+            property.setId("createTime");
+            property.setValueType(DateTimeType.GLOBAL);
+            property.setName("创建时间");
+            metadata.add(property);
+        }
+    }
+
+    @Override
+    public TimeSeriesMetric getMetric() {
+        return metric;
+    }
+
+    @Override
+    public List<PropertyMetadata> getProperties() {
+        return new ArrayList<>(fixed);
+    }
+}