feat(场景联动): 添加设备数据执行动作，扩展数组条件，优化国际化 (#605)

jetlinks · Jan 23, 2025 · 27a9fd6 · 27a9fd6
1 parent ad6279e
commit 27a9fd6
Show file tree

Hide file tree

Showing 53 changed files with 2,332 additions and 145 deletions.
diff --git a/jetlinks-components/common-component/pom.xml b/jetlinks-components/common-component/pom.xml
@@ -19,6 +19,11 @@
             <version>${jetlinks.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.jetlinks</groupId>
+            <artifactId>jetlinks-supports</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.hswebframework.web</groupId>
             <artifactId>hsweb-authorization-api</artifactId>

diff --git a/...mon-component/src/main/java/org/jetlinks/community/configuration/CommonConfiguration.java b/...mon-component/src/main/java/org/jetlinks/community/configuration/CommonConfiguration.java
@@ -20,6 +20,8 @@
 import org.jetlinks.community.config.SimpleConfigManager;
 import org.jetlinks.community.config.entity.ConfigEntity;
 import org.jetlinks.community.dictionary.DictionaryJsonDeserializer;
+import org.jetlinks.community.reactorql.aggregation.InternalAggregationSupports;
+import org.jetlinks.community.reactorql.function.InternalFunctionSupport;
 import org.jetlinks.community.reference.DataReferenceManager;
 import org.jetlinks.community.reference.DataReferenceProvider;
 import org.jetlinks.community.reference.DefaultDataReferenceManager;
@@ -32,10 +34,12 @@
 import org.jetlinks.community.utils.TimeUtils;
 import org.jetlinks.core.event.EventBus;
 import org.jetlinks.core.metadata.DataType;
+import org.jetlinks.core.metadata.types.DataTypes;
 import org.jetlinks.core.rpc.RpcManager;
 import org.jetlinks.reactor.ql.feature.Feature;
 import org.jetlinks.reactor.ql.supports.DefaultReactorQLMetadata;
 import org.jetlinks.reactor.ql.utils.CastUtils;
+import org.jetlinks.supports.official.JetLinksDataTypeCodecs;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.config.BeanPostProcessor;
@@ -47,6 +51,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.core.ReactiveRedisOperations;
 import org.springframework.http.MediaType;
+import org.springframework.util.StringUtils;
 import org.springframework.util.unit.DataSize;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Hooks;
@@ -59,13 +64,17 @@
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
 import java.util.Date;
+import java.util.Map;
 
 @Configuration
 @SuppressWarnings("all")
 @EnableConfigurationProperties({ConfigScopeProperties.class})
 public class CommonConfiguration {
 
     static {
+        InternalAggregationSupports.register();
+        InternalFunctionSupport.register();
+
         BeanUtilsBean.getInstance().getConvertUtils().register(new Converter() {
             @Override
             public <T> T convert(Class<T> aClass, Object o) {
@@ -149,6 +158,23 @@ public <T> T convert(Class<T> type, Object value) {
             }
         }, EnumDict.class);
 
+        BeanUtilsBean.getInstance().getConvertUtils().register(new Converter() {
+            @Override
+            @Generated
+            public <T> T convert(Class<T> type, Object value) {
+                if (value instanceof Map) {
+                    Map<String, Object> map = ((Map) value);
+                    String typeId = (String) map.get("type");
+                    if (StringUtils.isEmpty(typeId)) {
+                        return null;
+                    }
+                    return (T) JetLinksDataTypeCodecs.decode(DataTypes.lookup(typeId).get(), map);
+                }
+                return null;
+
+            }
+        }, DataType.class);
+
         //捕获jvm错误,防止Flux被挂起
         Hooks.onOperatorError((err, val) -> {
             if (Exceptions.isJvmFatal(err)) {
@@ -180,6 +206,7 @@ public Object postProcessAfterInitialization(@Nonnull Object bean, @Nonnull Stri
     @Bean
     public Jackson2ObjectMapperBuilderCustomizer jackson2ObjectMapperBuilderCustomizer(){
         return builder->{
+            builder.deserializerByType(DataType.class, new DataTypeJSONDeserializer());
             builder.deserializerByType(Date.class,new SmartDateDeserializer());
             builder.deserializerByType(EnumDict.class, new DictionaryJsonDeserializer());
         };

diff --git a/...omponent/src/main/java/org/jetlinks/community/configuration/DataTypeJSONDeserializer.java b/...omponent/src/main/java/org/jetlinks/community/configuration/DataTypeJSONDeserializer.java
@@ -0,0 +1,26 @@
+package org.jetlinks.community.configuration;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import org.apache.commons.beanutils.BeanUtilsBean;
+import org.jetlinks.core.metadata.DataType;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ *
+ * @author zhangji 2025/1/23
+ * @since 2.3
+ */
+public class DataTypeJSONDeserializer extends JsonDeserializer<DataType> {
+    @Override
+    public DataType deserialize(JsonParser parser, DeserializationContext ctxt) throws IOException, JsonProcessingException {
+
+        Map<String,Object> map= ctxt.readValue(parser, Map.class);
+
+        return (DataType) BeanUtilsBean.getInstance().getConvertUtils().convert(map, DataType.class);
+    }
+}
diff --git a/...ponent/src/main/java/org/jetlinks/community/reactorql/aggregation/AggregationSupport.java b/...ponent/src/main/java/org/jetlinks/community/reactorql/aggregation/AggregationSupport.java
@@ -0,0 +1,35 @@
+package org.jetlinks.community.reactorql.aggregation;
+
+import org.hswebframework.ezorm.rdb.operator.builder.fragments.SqlFragments;
+import org.hswebframework.ezorm.rdb.operator.dml.FunctionColumn;
+import org.jetlinks.community.spi.Provider;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+/**
+ * 聚合函数支持.
+ *
+ * @author zhangji 2025/1/22
+ * @since 2.3
+ */
+public interface AggregationSupport extends Function<Publisher<?>, Mono<?>> {
+
+    Provider<AggregationSupport> supports = Provider.create(AggregationSupport.class);
+
+    String getId();
+
+    String getName();
+
+    SqlFragments createSql(FunctionColumn column);
+
+
+    static AggregationSupport getNow(String id) {
+        return AggregationSupport.supports
+            .get(id.toUpperCase())
+            .orElseGet(() -> AggregationSupport.supports
+                .get(id.toLowerCase())
+                .orElseGet(() -> AggregationSupport.supports.getNow(id)));
+    }
+}
diff --git a/...c/main/java/org/jetlinks/community/reactorql/aggregation/InternalAggregationSupports.java b/...c/main/java/org/jetlinks/community/reactorql/aggregation/InternalAggregationSupports.java
@@ -0,0 +1,82 @@
+package org.jetlinks.community.reactorql.aggregation;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.hswebframework.ezorm.rdb.operator.builder.fragments.BatchSqlFragments;
+import org.hswebframework.ezorm.rdb.operator.builder.fragments.SqlFragments;
+import org.hswebframework.ezorm.rdb.operator.dml.FunctionColumn;
+import org.jetlinks.reactor.ql.supports.agg.MapAggFeature;
+import org.jetlinks.reactor.ql.utils.CastUtils;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.math.MathFlux;
+
+import java.util.Comparator;
+import java.util.function.Function;
+
+import static org.jetlinks.reactor.ql.supports.DefaultReactorQLMetadata.addGlobal;
+
+/**
+ *
+ * @author zhangji 2025/1/22
+ * @since 2.3
+ */
+@AllArgsConstructor
+public enum InternalAggregationSupports implements AggregationSupport {
+    COUNT("总数", Flux::count, 0),
+    //去重计数
+    DISTINCT_COUNT("总数(去重)", flux -> flux.distinct().count(), 0) {
+        @Override
+        public SqlFragments createSql(FunctionColumn column) {
+            return new BatchSqlFragments().addSql("count(distinct ", column.getColumn(), ")");
+        }
+    },
+    MIN("最小值",
+        numberFlux -> MathFlux.min(numberFlux.map(CastUtils::castNumber), Comparator.comparing(Number::doubleValue)), null),
+    MAX("最大值", numberFlux -> MathFlux.max(numberFlux.map(CastUtils::castNumber), Comparator.comparing(Number::doubleValue)), null),
+    AVG("平均值", numberFlux -> MathFlux.averageDouble(numberFlux.map(CastUtils::castNumber), Number::doubleValue), null),
+    SUM("总和", numberFlux -> MathFlux.sumDouble(numberFlux.map(CastUtils::castNumber), Number::doubleValue), 0),
+
+    FIRST("第一个值", numberFlux -> numberFlux.take(1).singleOrEmpty(), null),
+    LAST("最后一个值", numberFlux -> numberFlux.takeLast(1).singleOrEmpty(), null),
+
+//    MEDIAN("中位数", numberFlux -> Mono.empty(), null),//中位数
+//    SPREAD("极差", numberFlux -> Mono.empty(), null),//差值
+//    STDDEV("标准差", numberFlux -> Mono.empty(), null),//标准差
+    ;
+
+    static {
+        for (InternalAggregationSupports value : values()) {
+            addGlobal(new MapAggFeature(value.getId(), value::apply));
+            AggregationSupport.supports.register(value.getId(), value);
+        }
+    }
+
+    public static void register(){
+
+    }
+
+    @Getter
+    private final String name;
+
+    private final Function<Flux<?>, Mono<?>> computer;
+    @Getter
+    private final Object defaultValue;
+
+    @Override
+    public SqlFragments createSql(FunctionColumn column) {
+        return new BatchSqlFragments()
+            .addSql(name() + "(").addSql(column.getColumn()).addSql(")");
+    }
+
+    @Override
+    public String getId() {
+        return name();
+    }
+
+    @Override
+    public Mono<?> apply(Publisher<?> publisher) {
+        return computer.apply(Flux.from(publisher));
+    }
+}
diff --git a/...ommon-component/src/main/java/org/jetlinks/community/reactorql/function/FunctionInfo.java b/...ommon-component/src/main/java/org/jetlinks/community/reactorql/function/FunctionInfo.java
@@ -0,0 +1,25 @@
+package org.jetlinks.community.reactorql.function;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.core.metadata.DataType;
+import org.jetlinks.core.metadata.PropertyMetadata;
+
+import java.util.List;
+
+/**
+ *
+ * @author zhangji 2025/1/22
+ * @since 2.3
+ */
+@Getter
+@Setter
+public class FunctionInfo {
+    private String id;
+
+    private String name;
+
+    private DataType outputType;
+
+    private List<PropertyMetadata> args;
+}
diff --git a/...on-component/src/main/java/org/jetlinks/community/reactorql/function/FunctionSupport.java b/...on-component/src/main/java/org/jetlinks/community/reactorql/function/FunctionSupport.java
@@ -0,0 +1,72 @@
+package org.jetlinks.community.reactorql.function;
+
+import org.hswebframework.ezorm.rdb.operator.builder.fragments.SqlFragments;
+import org.jetlinks.core.metadata.DataType;
+import org.jetlinks.community.spi.Provider;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * 函数支持,用于定义在可以在ReactorQL中使用的函数.
+ *
+ * @author zhangji 2025/1/22
+ * @since 2.3
+ */
+public interface FunctionSupport {
+
+    Provider<FunctionSupport> supports = Provider.create(FunctionSupport.class);
+
+    String getId();
+
+    String getName();
+
+    /**
+     * 是否支持列的数据类型
+     *
+     * @param type 数据类型
+     * @return 是否支持
+     */
+    boolean isSupported(DataType type);
+
+    /**
+     * 获取输出数据类型
+     *
+     * @return 输出数据类型
+     */
+    DataType getOutputType();
+
+    /**
+     * 创建SQL函数片段
+     *
+     * @param column 列名
+     * @param args   参数
+     * @return SQL函数片段
+     */
+    SqlFragments createSql(String column, Map<String, Object> args);
+
+    /**
+     * 查找支持的函数
+     *
+     * @param type 数据类型
+     * @return 函数信息
+     */
+    static List<FunctionInfo> lookup(DataType type) {
+        return supports
+            .getAll()
+            .stream()
+            .filter(support -> support.isSupported(type))
+            .map(FunctionSupport::toInfo)
+            .collect(Collectors.toList());
+    }
+
+
+    default FunctionInfo toInfo() {
+        FunctionInfo info = new FunctionInfo();
+        info.setId(getId());
+        info.setOutputType(getOutputType());
+        info.setName(getName());
+        return info;
+    }
+}