add 执行场景互联

test 地图
test 指令下发

jetlinks-components/rule-engine-component/src/main/java/org/jetlinks/community/rule/engine/executor/DeviceMessageSendTaskExecutorProvider.java

@@ -6,6 +6,7 @@ import lombok.Setter;
 import org.hswebframework.web.bean.FastBeanCopier;
 import org.hswebframework.web.id.IDGenerator;
 import org.hswebframework.web.utils.ExpressionUtils;
+import org.jetlinks.core.defaults.DefaultDeviceOperator;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceProductOperator;
 import org.jetlinks.core.device.DeviceRegistry;

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

@@ -1,13 +1,19 @@
 package org.jetlinks.community.device.web;
 
 import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.authorization.annotation.Authorize;
 import org.hswebframework.web.authorization.annotation.Resource;
+import org.hswebframework.web.crud.service.ReactiveCrudService;
+import org.hswebframework.web.crud.web.reactive.ReactiveServiceCrudController;
 import org.hswebframework.web.exception.BusinessException;
 import org.hswebframework.web.id.IDGenerator;
+import org.jetlinks.community.device.entity.DeviceInstanceEntity;
 import org.jetlinks.community.device.entity.DevicePropertiesEntity;
+import org.jetlinks.community.device.service.LocalDeviceInstanceService;
 import org.jetlinks.community.utils.ErrorUtils;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
@@ -32,16 +38,28 @@ import java.util.Map;
 import java.util.function.Function;
 
 @RestController
-@RequestMapping("/device")
+@RequestMapping("/device/message/task")
 @Slf4j
 @Authorize
 @Resource(id = "device-instance", name = "设备实例")
 @Tag(name = "设备指令API")
 @Deprecated
-public class DeviceMessageController {
+@AllArgsConstructor
+public class DeviceMessageController implements
+    ReactiveServiceCrudController<DeviceInstanceEntity, String> {
 
-    @Autowired
-    private DeviceRegistry registry;
+
+    private DeviceRegistry registry
+        ;
+
+    private final LocalDeviceInstanceService service;
+
+
+
+    @Override
+    public ReactiveCrudService<DeviceInstanceEntity, String> getService() {
+        return service;
+    }
 
     //获取设备属性
     @GetMapping("/{deviceId}/property/{property:.+}")

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

@@ -5,8 +5,10 @@ import lombok.AllArgsConstructor;
 import org.hswebframework.web.authorization.annotation.Authorize;
 import org.hswebframework.web.authorization.annotation.Resource;
 import org.jetlinks.community.device.service.LocalDeviceInstanceService;
+import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
 
 /**
  * @author lifang
@@ -23,4 +25,9 @@ import org.springframework.web.bind.annotation.RestController;
 @AllArgsConstructor
 public class GeoController {
     private final LocalDeviceInstanceService deviceInstanceService;
+
+    @PostMapping("/object/_search/geo.json")
+    public Mono<Object> search(){
+        return Mono.empty();
+    }
 }

jetlinks-manager/rule-engine-manager/src/main/java/org/jetlinks/community/rule/engine/device/RuleSceneTaskExecutorProvider.java

@@ -11,6 +11,7 @@ import org.jetlinks.core.event.EventBus;
 import org.jetlinks.core.event.Subscription;
 import org.jetlinks.core.message.DeviceMessage;
 import org.jetlinks.core.metadata.Jsonable;
+import org.jetlinks.reactor.ql.DefaultReactorQLRecord;
 import org.jetlinks.reactor.ql.ReactorQL;
 import org.jetlinks.reactor.ql.ReactorQLContext;
 import org.jetlinks.reactor.ql.ReactorQLRecord;
@@ -21,6 +22,7 @@ import org.jetlinks.rule.engine.api.task.Task;
 import org.jetlinks.rule.engine.api.task.TaskExecutor;
 import org.jetlinks.rule.engine.api.task.TaskExecutorProvider;
 import org.jetlinks.rule.engine.defaults.AbstractTaskExecutor;
+import org.jetlinks.supports.event.BrokerEventBus;
 import org.springframework.stereotype.Component;
 import org.springframework.util.StringUtils;
 import reactor.core.Disposable;
@@ -45,6 +47,7 @@ public class RuleSceneTaskExecutorProvider implements TaskExecutorProvider {
 
     private final Scheduler scheduler;
 
+
     @Override
     public String getExecutor() {
         return "rule_scene";
@@ -57,9 +60,9 @@ public class RuleSceneTaskExecutorProvider implements TaskExecutorProvider {
 
     class RuleSceneTaskExecutor extends AbstractTaskExecutor {
 
-        List<String> default_columns = Arrays.asList(
-            "timestamp", "deviceId"
-        );
+//        List<String> default_columns = Arrays.asList(
+//            "timestamp", "deviceId"
+//        );
 
         private RuleSceneEntity rule;
 
@@ -123,7 +126,8 @@ public class RuleSceneTaskExecutorProvider implements TaskExecutorProvider {
         }
 
         private ReactorQL createQL(RuleSceneEntity rule) {
-            List<String> columns = new ArrayList<>(default_columns);
+//            List<String> columns = new ArrayList<>(default_columns);
+            List<String> columns = new ArrayList<>();
             List<String> wheres = new ArrayList<>();
 
             List<RuleSceneEntity.Trigger> triggers = rule.getTriggers();
@@ -140,7 +144,6 @@ public class RuleSceneTaskExecutorProvider implements TaskExecutorProvider {
                     .ifPresent(expr -> wheres.add("(" + expr + ")"));
             }
             String sql = "select \n\t\t" + String.join("\n\t\t,", columns) + " \n\tfrom dual ";
-
             if (!wheres.isEmpty()) {
                 sql += "\n\twhere " + String.join("\n\t\t or ", wheres);
             }
@@ -152,71 +155,59 @@ public class RuleSceneTaskExecutorProvider implements TaskExecutorProvider {
 
         public Flux<Map<String, Object>> doSubscribe(EventBus eventBus) {
             Flux<Map<String, Object>> result =null;
+            List<Object> binds = new ArrayList<>();
             for (RuleSceneEntity.Trigger triggerMix : rule.getTriggers()) {
-                RuleSceneEntity.DeviceTrigger trigger = triggerMix.getDevice();
-                String topic = trigger.getType().getTopic(trigger.getProductId(), trigger.getDeviceId(), trigger.getModelId());
-                //订阅主题
-                Subscription subscription = Subscription.of(
-                    "rule_scene:" + topic,
-                    topic,
-                    Subscription.Feature.local
+            RuleSceneEntity.DeviceTrigger trigger = rule.getTriggers().get(0).getDevice();
+            String topic = trigger.getType().getTopic(trigger.getProductId(), trigger.getDeviceId(), trigger.getModelId());
+            binds.addAll(trigger.toFilterBinds());
+            //订阅主题
+            Subscription subscription = Subscription.of(
+                "rule_scene:" + topic,
+                topic,
+                Subscription.Feature.local
+            );
+            ReactorQLContext context = ReactorQLContext
+                .ofDatasource(ignore ->
+                    eventBus
+                        .subscribe(subscription, DeviceMessage.class)
+                        .map(Jsonable::toJson)
+                        .doOnNext(json -> {
+                            json.put("ruleSceneId", rule.getId());
+                            json.put("ruleSceneName", rule.getName());
+                        })
                 );
-                Flux<Map<String, Object>> resultFlux = eventBus
-                    .subscribe(subscription, DeviceMessage.class)
-                    .map(Jsonable::toJson)
-                    .doOnNext(json -> {
-                        json.put("productId", trigger.getProductId());
-                        json.put("ruleSceneId", rule.getId());
-                        json.put("ruleSceneName", rule.getName());
-                    })
-                    .map(JSONObject::getInnerMap);
-                ShakeLimit shakeLimit = trigger.getShakeLimit();
-                if (shakeLimit != null
-                    && shakeLimit.isEnabled()
-                    && shakeLimit.getTime() > 0) {
-                    int thresholdNumber = shakeLimit.getThreshold();
-                    Duration windowTime = Duration.ofSeconds(shakeLimit.getTime());
-                    resultFlux
-                        .as(flux ->
-                            StringUtils.hasText(trigger.getDeviceId())
-                                ? flux.window(windowTime, scheduler)//规则已经指定了固定的设备,直接开启时间窗口就行
-                                : flux //规则配置在设备产品上,则按设备ID分组后再开窗口
-                                .groupBy(map -> String.valueOf(map.get("deviceId")), Integer.MAX_VALUE)
-                                .flatMap(group -> group.window(windowTime, scheduler), Integer.MAX_VALUE))
-                        //处理每一组数据
-                        .flatMap(group -> group
-                            .index((index, data) -> Tuples.of(index + 1, data)) //给数据打上索引,索引号就是告警次数
-                            .filter(tp -> tp.getT1() >= thresholdNumber)//超过阈值告警
-                            .as(flux -> shakeLimit.isAlarmFirst() ? flux.take(1) : flux.takeLast(1))//取第一个或者最后一个
-                            .map(tp2 -> {
-                                tp2.getT2().put("totalScene", tp2.getT1());
-                                return tp2.getT2();
-                            }));
-                }
+
+            binds.forEach(context::bind);
+            Flux<Map<String, Object>> resultFlux = (ql == null ? ql = createQL(rule) : ql)
+                .start(context)
+                .map(ReactorQLRecord::asMap);
+
+            ShakeLimit shakeLimit = trigger.getShakeLimit();
+            if (shakeLimit != null
+                && shakeLimit.isEnabled()
+                && shakeLimit.getTime() > 0) {
+                int thresholdNumber = shakeLimit.getThreshold();
+                Duration windowTime = Duration.ofSeconds(shakeLimit.getTime());
+                resultFlux
+                    .as(flux ->
+                        StringUtils.hasText(trigger.getDeviceId())
+                            ? flux.window(windowTime, scheduler)//规则已经指定了固定的设备,直接开启时间窗口就行
+                            : flux //规则配置在设备产品上,则按设备ID分组后再开窗口
+                            .groupBy(map -> String.valueOf(map.get("deviceId")), Integer.MAX_VALUE)
+                            .flatMap(group -> group.window(windowTime, scheduler), Integer.MAX_VALUE))
+                    //处理每一组数据
+                    .flatMap(group -> group
+                        .index((index, data) -> Tuples.of(index + 1, data)) //给数据打上索引,索引号就是告警次数
+                        .filter(tp -> tp.getT1() >= thresholdNumber)//超过阈值告警
+                        .as(flux -> shakeLimit.isAlarmFirst() ? flux.take(1) : flux.takeLast(1))//取第一个或者最后一个
+                        .map(tp2 -> {
+                            tp2.getT2().put("totalScene", tp2.getT1());
+                            return tp2.getT2();
+                        }));
+            }
                 result=result==null?resultFlux:result.concatWith(resultFlux);
             }
             return result;
         }
     }
-
-    public static void main(String[] args) {
-        Flux<String> stringFlux = Flux.just("1")
-            .doOnNext(System.out::println);
-        Flux<String> stringFlux2 = Flux.just("2")
-            .doOnNext(System.out::println);
-        Flux<String> stringFlux3 = Flux.just("3")
-            .doOnNext(System.out::println);
-        Flux.empty().concatWith(stringFlux)
-//        Flux<Flux<String>> fluxFlux = Flux.fromStream(Stream.of(stringFlux, stringFlux2));
-        .concatWith(stringFlux2).concatWith(stringFlux3)
-            .doOnNext(s->{
-                System.out.println(s);
-            }).
-                subscribe();
-//        fluxFlux
-//            .publishOn(Schedulers.elastic())
-//            .flatMap(Function.identity())
-//            .then()
-//            .subscribe();
-    }
 }

jetlinks-manager/rule-engine-manager/src/main/java/org/jetlinks/community/rule/engine/service/RuleInstanceService.java

@@ -1,9 +1,11 @@
 package org.jetlinks.community.rule.engine.service;
 
+import cn.hutool.core.collection.CollectionUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.ezorm.core.param.QueryParam;
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.crud.service.GenericReactiveCrudService;
+import org.hswebframework.web.exception.NotFoundException;
 import org.jetlinks.community.elastic.search.service.ElasticSearchService;
 import org.jetlinks.community.rule.engine.entity.RuleEngineExecuteEventInfo;
 import org.jetlinks.community.rule.engine.entity.RuleEngineExecuteLogInfo;
@@ -20,6 +22,7 @@ import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 @Service
 @Slf4j
@@ -52,9 +55,12 @@ public class RuleInstanceService extends GenericReactiveCrudService<RuleInstance
             .then();
     }
 
-    public Mono<Void> execute(String id, RuleData ruleData) {
-        return this.ruleEngine.getTasks(id)
-            .map(task-> task.execute(ruleData))
+    public Mono<Void> execute(String taskId, RuleData ruleData) {
+        return this.ruleEngine.getTasks(taskId)
+            .switchIfEmpty(Mono.error(NotFoundException::new))
+            .filter(task -> CollectionUtil.isEmpty(task.getJob().getOutputs()))
+            .publishOn(Schedulers.parallel())
+            .flatMap(task-> task.execute(ruleData))
             .then();
     }
 

jetlinks-manager/rule-engine-manager/src/main/java/org/jetlinks/community/rule/engine/service/RuleSceneService.java

@@ -6,11 +6,16 @@ import org.jetlinks.community.rule.engine.entity.DeviceAlarmEntity;
 import org.jetlinks.community.rule.engine.entity.RuleSceneEntity;
 import org.jetlinks.community.rule.engine.enums.AlarmState;
 import org.jetlinks.community.rule.engine.enums.RuleSceneState;
+import org.jetlinks.rule.engine.api.RuleData;
 import org.reactivestreams.Publisher;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -59,7 +64,6 @@ public class RuleSceneService extends GenericReactiveCrudService<RuleSceneEntity
     }
 
     public Mono<Void> execute(String id) {
-
-        return null;
+        return instanceService.execute(id, RuleData.create(null));
     }
 }

jetlinks-manager/rule-engine-manager/src/main/java/org/jetlinks/community/rule/engine/web/RuleSceneController.java

@@ -57,4 +57,5 @@ public class RuleSceneController implements ReactiveServiceCrudController<RuleSc
     public Mono<Void> executeScene(@PathVariable String id) {
         return sceneService.execute(id);
     }
+
 }