Merge remote-tracking branch 'origin/master'

README.md

@@ -7,7 +7,7 @@
 JetLinks 基于Java8,Spring Boot 2.x,WebFlux,Netty,Vert.x,Reactor等开发, 
 是一个开箱即用,可二次开发的企业级物联网基础平台。平台实现了物联网相关的众多基础功能,
 能帮助你快速建立物联网相关业务系统。
-
+ 
 
 ## 核心特性
 

jetlinks-components/dashboard-component/src/main/java/org/jetlinks/community/dashboard/measurements/SystemMonitor.java

@@ -78,7 +78,9 @@ public enum SystemMonitor {
 
     @SneakyThrows
     public static double getValue(String id) {
-        return items.getOrDefault(id, zero).call();
+        double val = items.getOrDefault(id, zero).call();
+
+        return Double.isNaN(val) ? 0 : val;
     }
 
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/DefaultElasticSearchService.java

@@ -153,8 +153,8 @@ public class DefaultElasticSearchService implements ElasticSearchService {
         //这里的警告都输出到控制台,输入到slf4j可能会造成日志递归.
         FluxUtils.bufferRate(
             Flux.<Buffer>create(sink -> this.sink = sink),
-            1000,
-            2000,
+            Integer.getInteger("elasticsearch.flush.rate", 1000),
+            Integer.getInteger("elasticsearch.buffer.size", 2000),
             Duration.ofSeconds(3))
             .onBackpressureBuffer(512,
                 drop -> System.err.println("无法处理更多索引请求!"),

jetlinks-components/logging-component/src/main/java/org/jetlinks/community/logging/logback/SystemLoggingAppender.java

@@ -59,7 +59,7 @@ public class SystemLoggingAppender extends UnsynchronizedAppenderBase<ILoggingEv
             String gitLocation = null;
             String mavenModule = null;
             try {
-                Class clazz = Class.forName(element.getClassName());
+                Class<?> clazz = Class.forName(element.getClassName());
                 ModuleUtils.ModuleInfo moduleInfo = ModuleUtils.getModuleByClass(clazz);
                 if (!StringUtils.isEmpty(moduleInfo.getGitRepository())) {
                     StringBuilder javaSb = new StringBuilder();
@@ -73,8 +73,7 @@ public class SystemLoggingAppender extends UnsynchronizedAppenderBase<ILoggingEv
                     gitLocation = javaSb.toString();
                 }
                 mavenModule = moduleInfo.getArtifactId();
-            } catch (Exception e) {
-                log.warn("记录系统日志时，加载类:{}错误。{}", element.getClassName(), e);
+            } catch (Exception ignore) {
             }
             Map<String, String> context = new HashMap<>(staticContext);
             Map<String, String> mdc = MDC.getCopyOfContextMap();

jetlinks-components/network-component/tcp-component/src/test/java/org/jetlinks/community/network/tcp/server/TcpServerProviderTest.java

@@ -24,11 +24,12 @@ class TcpServerProviderTest {
     @BeforeAll
     static void init() {
         TcpServerProperties properties = TcpServerProperties.builder()
-                .id("test")
-                .options(new NetServerOptions().setPort(8080))
-                .parserType(PayloadParserType.FIXED_LENGTH)
-                .parserConfiguration(Collections.singletonMap("size", 5))
-                .build();
+            .id("test")
+            .port(8080)
+            .options(new NetServerOptions())
+            .parserType(PayloadParserType.FIXED_LENGTH)
+            .parserConfiguration(Collections.singletonMap("size", 5))
+            .build();
 
         TcpServerProvider provider = new TcpServerProvider((id) -> Mono.empty(), Vertx.vertx(), new DefaultPayloadParserBuilder());
 
@@ -40,30 +41,30 @@ class TcpServerProviderTest {
     void test() {
 
         Vertx.vertx().createNetClient()
-                .connect(8080, "localhost", handle -> {
-                    if (handle.succeeded()) {
-                        //模拟粘包，同时发送2个包
-                        handle.result().write("hellohello", r -> {
-                            if (r.succeeded()) {
-                                log.info("tcp客户端消息发送成功");
-                            } else {
-                                log.error("tcp客户端消息发送错误", r.cause());
-                            }
-                        });
-                    } else {
-                        log.error("创建tcp客户端错误", handle.cause());
-                    }
-                });
+            .connect(8080, "localhost", handle -> {
+                if (handle.succeeded()) {
+                    //模拟粘包，同时发送2个包
+                    handle.result().write("hellohello", r -> {
+                        if (r.succeeded()) {
+                            log.info("tcp客户端消息发送成功");
+                        } else {
+                            log.error("tcp客户端消息发送错误", r.cause());
+                        }
+                    });
+                } else {
+                    log.error("创建tcp客户端错误", handle.cause());
+                }
+            });
 
 
         tcpServer.handleConnection()
-                .flatMap(TcpClient::subscribe)
-                .map(TcpMessage::getPayload)
-                .map(payload -> payload.toString(StandardCharsets.UTF_8))
-                .take(2)
-                .as(StepVerifier::create)
-                .expectNext("hello", "hello")//收到2个完整的包
-                .verifyComplete();
+            .flatMap(TcpClient::subscribe)
+            .map(TcpMessage::getPayload)
+            .map(payload -> payload.toString(StandardCharsets.UTF_8))
+            .take(2)
+            .as(StepVerifier::create)
+            .expectNext("hello", "hello")//收到2个完整的包
+            .verifyComplete();
     }
 
 

jetlinks-components/notify-component/notify-core/pom.xml

@@ -35,6 +35,12 @@
             <artifactId>common-component</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.jetlinks.community</groupId>
+            <artifactId>gateway-component</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
 
     </dependencies>
 

jetlinks-components/notify-component/notify-core/src/main/java/org/jetlinks/community/notify/DefaultNotifierManager.java

@@ -1,6 +1,7 @@
 package org.jetlinks.community.notify;
 
 import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.community.gateway.MessageGateway;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.stereotype.Component;
@@ -21,8 +22,11 @@ public class DefaultNotifierManager implements NotifierManager, BeanPostProcesso
 
     private NotifyConfigManager configManager;
 
-    public DefaultNotifierManager(NotifyConfigManager manager) {
+    private MessageGateway messageGateway;
+
+    public DefaultNotifierManager(NotifyConfigManager manager, MessageGateway messageGateway) {
         this.configManager = manager;
+        this.messageGateway = messageGateway;
     }
 
     protected Mono<NotifierProperties> getProperties(NotifyType notifyType,
@@ -42,6 +46,8 @@ public class DefaultNotifierManager implements NotifierManager, BeanPostProcesso
             .flatMap(map -> Mono.justOrEmpty(map.get(properties.getProvider())))
             .switchIfEmpty(Mono.error(new UnsupportedOperationException("不支持的服务商:" + properties.getProvider())))
             .flatMap(notifierProvider -> notifierProvider.createNotifier(properties))
+            //转成代理,把通知事件发送到消息网关中.
+            .map(notifier -> new NotifierEventDispatcher<>(messageGateway, notifier))
             .flatMap(notifier -> Mono.justOrEmpty(notifiers.put(properties.getId(), notifier))
                 .flatMap(Notifier::close)//如果存在旧的通知器则关掉之
                 .onErrorContinue((err, obj) -> log.error(err.getMessage(), err))//忽略异常

jetlinks-components/notify-component/notify-core/src/main/java/org/jetlinks/community/notify/Notifier.java

@@ -17,6 +17,11 @@ import java.util.function.Consumer;
  */
 public interface Notifier<T extends Template> {
 
+    /**
+     * @return 通知器ID
+     */
+    String getNotifierId();
+
     /**
      * 获取通知类型,如: 语音通知
      *
@@ -73,4 +78,7 @@ public interface Notifier<T extends Template> {
     @Nonnull
     Mono<Void> close();
 
+    default <R extends Notifier<T>> R unwrap(Class<R> type) {
+        return type.cast(this);
+    }
 }

jetlinks-components/notify-component/notify-core/src/main/java/org/jetlinks/community/notify/NotifierEventDispatcher.java

@@ -0,0 +1,27 @@
+package org.jetlinks.community.notify;
+
+import org.jetlinks.community.gateway.MessageGateway;
+import org.jetlinks.community.notify.event.NotifierEvent;
+import org.jetlinks.community.notify.template.Template;
+import reactor.core.publisher.Mono;
+
+public class NotifierEventDispatcher<T extends Template> extends NotifierProxy<T> {
+
+    private final MessageGateway gateway;
+
+    public NotifierEventDispatcher(MessageGateway gateway, Notifier<T> target) {
+        super(target);
+        this.gateway = gateway;
+    }
+
+    @Override
+    protected Mono<Void> onEvent(NotifierEvent event) {
+        // /notify/{notifierId}/success
+
+        return gateway
+            .publish(String.join("/", "/notify", event.getNotifierId(), event.isSuccess() ? "success" : "error"), event.toSerializable())
+            .then();
+    }
+
+
+}

jetlinks-components/notify-component/notify-core/src/main/java/org/jetlinks/community/notify/NotifierProxy.java

@@ -0,0 +1,106 @@
+package org.jetlinks.community.notify;
+
+import lombok.AllArgsConstructor;
+import org.jetlinks.core.Values;
+import org.jetlinks.community.notify.event.NotifierEvent;
+import org.jetlinks.community.notify.template.Template;
+import reactor.core.publisher.Mono;
+
+import javax.annotation.Nonnull;
+
+@AllArgsConstructor
+public abstract class NotifierProxy<T extends Template> implements Notifier<T> {
+
+    private final Notifier<T> target;
+
+    @Override
+    public String getNotifierId() {
+        return target.getNotifierId();
+    }
+
+    @Nonnull
+    @Override
+    public NotifyType getType() {
+        return target.getType();
+    }
+
+    @Nonnull
+    @Override
+    public Provider getProvider() {
+        return target.getProvider();
+    }
+
+    @Override
+    public <R extends Notifier<T>> R unwrap(Class<R> type) {
+        return target.unwrap(type);
+    }
+
+    @Nonnull
+    @Override
+    public Mono<Void> send(@Nonnull String templateId, Values context) {
+        return target
+            .send(templateId, context)
+            .switchIfEmpty(Mono.defer(() -> onSuccess(templateId, context)))
+            .onErrorResume(err -> onError(templateId, context, err).then(Mono.error(err)));
+    }
+
+    @Nonnull
+    @Override
+    public Mono<Void> send(@Nonnull T template, @Nonnull Values context) {
+        return target.send(template, context)
+            .switchIfEmpty(Mono.defer(() -> onSuccess(template, context)))
+            .onErrorResume(err -> onError(template, context, err).then(Mono.error(err)));
+    }
+
+    protected Mono<Void> onError(T template, Values ctx, Throwable error) {
+        return onEvent(NotifierEvent.builder()
+            .cause(error)
+            .context(ctx.getAllValues())
+            .notifierId(getNotifierId())
+            .notifyType(getType())
+            .provider(getProvider())
+            .template(template)
+            .build());
+    }
+
+    protected Mono<Void> onError(String templateId, Values ctx, Throwable error) {
+        return onEvent(NotifierEvent.builder()
+            .cause(error)
+            .context(ctx.getAllValues())
+            .notifierId(getNotifierId())
+            .notifyType(getType())
+            .provider(getProvider())
+            .templateId(templateId)
+            .build());
+    }
+
+    protected Mono<Void> onSuccess(String templateId, Values ctx) {
+        return onEvent(NotifierEvent.builder()
+            .success(true)
+            .context(ctx.getAllValues())
+            .notifierId(getNotifierId())
+            .notifyType(getType())
+            .provider(getProvider())
+            .templateId(templateId)
+            .build());
+    }
+
+    protected Mono<Void> onSuccess(T template, Values ctx) {
+        return onEvent(NotifierEvent.builder()
+            .success(true)
+            .context(ctx.getAllValues())
+            .notifierId(getNotifierId())
+            .notifyType(getType())
+            .provider(getProvider())
+            .template(template)
+            .build());
+    }
+
+    protected abstract Mono<Void> onEvent(NotifierEvent event);
+
+    @Nonnull
+    @Override
+    public Mono<Void> close() {
+        return target.close();
+    }
+}

jetlinks-components/notify-component/notify-core/src/main/java/org/jetlinks/community/notify/event/NotifierEvent.java

@@ -0,0 +1,56 @@
+package org.jetlinks.community.notify.event;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+import org.hswebframework.utils.StringUtils;
+import org.jetlinks.community.notify.NotifyType;
+import org.jetlinks.community.notify.Provider;
+import org.jetlinks.community.notify.template.Template;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Map;
+
+@Getter
+@Setter
+@Builder
+public class NotifierEvent {
+
+    private boolean success;
+
+    @Nullable
+    private Throwable cause;
+
+    @Nonnull
+    private String notifierId;
+
+    @Nonnull
+    private NotifyType notifyType;
+
+    @Nonnull
+    private Provider provider;
+
+    @Nullable
+    private String templateId;
+
+    @Nullable
+    private Template template;
+
+    @Nonnull
+    private Map<String, Object> context;
+
+    public SerializableNotifierEvent toSerializable() {
+        return SerializableNotifierEvent.builder()
+            .success(success)
+            .notifierId(notifierId)
+            .notifyType(notifyType.getId())
+            .provider(provider.getId())
+            .templateId(templateId)
+            .template(template)
+            .context(context)
+            .cause(cause != null ? StringUtils.throwable2String(cause) : "")
+            .errorType(cause != null ? cause.getClass().getName() : null)
+            .build();
+    }
+}

jetlinks-components/notify-component/notify-core/src/main/java/org/jetlinks/community/notify/event/SerializableNotifierEvent.java

@@ -0,0 +1,42 @@
+package org.jetlinks.community.notify.event;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.community.notify.template.Template;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Map;
+
+@Getter
+@Setter
+@Builder
+public class SerializableNotifierEvent {
+
+    private boolean success;
+
+    @Nullable
+    private String errorType;
+
+    @Nullable
+    private String cause;
+
+    @Nonnull
+    private String notifierId;
+
+    @Nonnull
+    private String notifyType;
+
+    @Nonnull
+    private String provider;
+
+    @Nullable
+    private String templateId;
+
+    @Nullable
+    private Template template;
+
+    @Nonnull
+    private Map<String,Object> context;
+}

jetlinks-components/notify-component/notify-dingtalk/src/main/java/org/jetlinks/community/notify/dingtalk/DingTalkNotifier.java

@@ -1,5 +1,6 @@
 package org.jetlinks.community.notify.dingtalk;
 
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.exception.BusinessException;
 import org.jetlinks.core.Values;
@@ -22,24 +23,27 @@ import java.util.concurrent.atomic.AtomicReference;
 @Slf4j
 public class DingTalkNotifier extends AbstractNotifier<DingTalkMessageTemplate> {
 
-    private AtomicReference<String> accessToken = new AtomicReference<>();
+    private final AtomicReference<String> accessToken = new AtomicReference<>();
 
     private long refreshTokenTime;
 
-    private long tokenTimeOut = Duration.ofSeconds(7000).toMillis();
+    private final long tokenTimeOut = Duration.ofSeconds(7000).toMillis();
 
-    private WebClient client;
+    private final WebClient client;
 
     private static final String tokenApi = "https://oapi.dingtalk.com/gettoken";
 
     private static final String notify = "https://oapi.dingtalk.com/topapi/message/corpconversation/asyncsend_v2";
 
-    private DingTalkProperties properties;
+    private final DingTalkProperties properties;
+    @Getter
+    private final String notifierId;
 
-    public DingTalkNotifier(WebClient client, DingTalkProperties properties, TemplateManager templateManager) {
+    public DingTalkNotifier(String id,WebClient client, DingTalkProperties properties, TemplateManager templateManager) {
         super(templateManager);
         this.client = client;
         this.properties = properties;
+        this.notifierId = id;
     }
 
     @Nonnull

jetlinks-components/notify-component/notify-dingtalk/src/main/java/org/jetlinks/community/notify/dingtalk/DingTalkNotifierProvider.java

@@ -30,15 +30,15 @@ public class DingTalkNotifierProvider implements NotifierProvider, TemplateProvi
     }
 
     public static final DefaultConfigMetadata notifierConfig = new DefaultConfigMetadata("通知配置", "")
-            .add("appKey", "appKey", "", new StringType().expand(ConfigMetadataConstants.required.value(true)))
-            .add("appSecret", "appSecret", "", new StringType());
+        .add("appKey", "appKey", "", new StringType().expand(ConfigMetadataConstants.required.value(true)))
+        .add("appSecret", "appSecret", "", new StringType());
 
     public static final DefaultConfigMetadata templateConfig = new DefaultConfigMetadata("模版配置", "")
-            .add("agentId", "应用ID", "", new StringType().expand(ConfigMetadataConstants.required.value(true)))
-            .add("userIdList", "收信人ID", "与部门ID不能同时为空", new StringType())
-            .add("departmentIdList", "收信部门ID", "与收信人ID不能同时为空", new StringType())
-            .add("toAllUser", "全部用户", "推送到全部用户", new BooleanType())
-            .add("message", "内容", "最大不超过500字", new StringType().expand(ConfigMetadataConstants.maxLength.value(500L)));
+        .add("agentId", "应用ID", "", new StringType().expand(ConfigMetadataConstants.required.value(true)))
+        .add("userIdList", "收信人ID", "与部门ID不能同时为空", new StringType())
+        .add("departmentIdList", "收信部门ID", "与收信人ID不能同时为空", new StringType())
+        .add("toAllUser", "全部用户", "推送到全部用户", new BooleanType())
+        .add("message", "内容", "最大不超过500字", new StringType().expand(ConfigMetadataConstants.maxLength.value(500L)));
 
     @Nonnull
     @Override
@@ -64,7 +64,7 @@ public class DingTalkNotifierProvider implements NotifierProvider, TemplateProvi
     public Mono<DingTalkNotifier> createNotifier(@Nonnull NotifierProperties properties) {
         return Mono.defer(() -> {
             DingTalkProperties dingTalkProperties = FastBeanCopier.copy(properties.getConfiguration(), new DingTalkProperties());
-            return Mono.just(new DingTalkNotifier(client, ValidatorUtils.tryValidate(dingTalkProperties), templateManager));
+            return Mono.just(new DingTalkNotifier(properties.getId(), client, ValidatorUtils.tryValidate(dingTalkProperties), templateManager));
         });
     }
 

jetlinks-components/notify-component/notify-dingtalk/src/test/java/org/jetlinks/community/notify/dingtalk/DingTalkNotifierTest.java

@@ -27,7 +27,7 @@ class DingTalkNotifierTest {
         messageTemplate.setMessage("test"+System.currentTimeMillis());
         messageTemplate.setUserIdList("0458215455697857");
 
-        DingTalkNotifier notifier=new DingTalkNotifier(
+        DingTalkNotifier notifier=new DingTalkNotifier("test",
                 WebClient.builder().build(),properties,null
         );
 

jetlinks-components/notify-component/notify-email/src/main/java/org/jetlinks/community/notify/email/embedded/DefaultEmailNotifier.java

@@ -59,10 +59,15 @@ public class DefaultEmailNotifier extends AbstractNotifier<EmailTemplate> {
     @Setter
     private String sender;
 
+    @Getter
+    private final String notifierId;
+
+
     public static Scheduler scheduler = Schedulers.newElastic("email-notifier");
 
     public DefaultEmailNotifier(NotifierProperties properties, TemplateManager templateManager) {
         super(templateManager);
+        notifierId = properties.getId();
 
         DefaultEmailProperties emailProperties = new JSONObject(properties.getConfiguration())
             .toJavaObject(DefaultEmailProperties.class);

jetlinks-components/notify-component/notify-sms/src/main/java/org/jetlinks/community/notify/sms/provider/Hy2046SmsSenderProvider.java

@@ -1,6 +1,7 @@
 package org.jetlinks.community.notify.sms.provider;
 
 import com.alibaba.fastjson.JSON;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.jetlinks.core.Values;
@@ -84,7 +85,7 @@ public class Hy2046SmsSenderProvider implements NotifierProvider, TemplateProvid
             Assert.hasText(userId, "短信配置错误,缺少userId");
             Assert.hasText(username, "短信配置错误,缺少username");
             Assert.hasText(password, "短信配置错误,缺少password");
-            return Mono.just(new Hy2046SmsSender(userId, username, password));
+            return Mono.just(new Hy2046SmsSender(properties.getId(),userId, username, password));
         });
     }
 
@@ -103,10 +104,13 @@ public class Hy2046SmsSenderProvider implements NotifierProvider, TemplateProvid
         String userId;
         String username;
         String password;
+        @Getter
+        private final String notifierId;
 
-        public Hy2046SmsSender(String userId, String username, String password) {
+        public Hy2046SmsSender(String id,String userId, String username, String password) {
             super(templateManager);
             this.userId = userId;
+            this.notifierId = id;
             this.username = username;
             this.password = password;
         }

jetlinks-components/notify-component/notify-sms/src/main/java/org/jetlinks/community/notify/sms/provider/TestSmsProvider.java

@@ -69,6 +69,11 @@ public class TestSmsProvider extends AbstractNotifier<PlainTextSmsTemplate> impl
         return Mono.just(this);
     }
 
+    @Override
+    public String getNotifierId() {
+        return "test-sms-notify";
+    }
+
     @Override
     public String getId() {
         return "test";

jetlinks-components/notify-component/notify-voice/src/main/java/org/jetlinks/community/notify/voice/aliyun/AliyunVoiceNotifier.java

@@ -9,6 +9,7 @@ import com.aliyuncs.IAcsClient;
 import com.aliyuncs.http.MethodType;
 import com.aliyuncs.profile.DefaultProfile;
 import com.aliyuncs.profile.IClientProfile;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.exception.BusinessException;
 import org.hswebframework.web.logger.ReactiveLogger;
@@ -25,14 +26,17 @@ import java.util.Objects;
 @Slf4j
 public class AliyunVoiceNotifier extends AbstractNotifier<AliyunVoiceTemplate> {
 
-    private IAcsClient client;
+    private final IAcsClient client;
     private String domain = "dyvmsapi.aliyuncs.com";
     private String regionId = "cn-hangzhou";
-    private int connectTimeout = 1000;
-    private int readTimeout = 5000;
+    private final int connectTimeout = 1000;
+    private final int readTimeout = 5000;
+    @Getter
+    private String notifierId;
 
     public AliyunVoiceNotifier(NotifierProperties profile, TemplateManager templateManager) {
         super(templateManager);
+        this.notifierId = profile.getId();
         Map<String, Object> config = profile.getConfiguration();
         DefaultProfile defaultProfile = DefaultProfile.getProfile(
                 this.regionId = (String) Objects.requireNonNull(config.get("regionId"), "regionId不能为空"),

jetlinks-components/notify-component/notify-wechat/src/main/java/org/jetlinks/community/notify/wechat/WechatNotifierProvider.java

@@ -1,6 +1,7 @@
 package org.jetlinks.community.notify.wechat;
 
 import com.alibaba.fastjson.JSON;
+import lombok.Getter;
 import org.hswebframework.web.bean.FastBeanCopier;
 import org.hswebframework.web.validator.ValidatorUtils;
 import org.jetlinks.core.metadata.ConfigMetadata;
@@ -23,7 +24,6 @@ public class WechatNotifierProvider implements NotifierProvider, TemplateProvide
     private WebClient client = WebClient.create();
 
     private final TemplateManager templateManager;
-
     public WechatNotifierProvider(TemplateManager templateManager) {
         this.templateManager = templateManager;
     }
@@ -61,7 +61,7 @@ public class WechatNotifierProvider implements NotifierProvider, TemplateProvide
     public Mono<WeixinCorpNotifier> createNotifier(@Nonnull NotifierProperties properties) {
         return Mono.defer(() -> {
             WechatCorpProperties wechatCorpProperties = FastBeanCopier.copy(properties.getConfiguration(), new WechatCorpProperties());
-            return Mono.just(new WeixinCorpNotifier(client, ValidatorUtils.tryValidate(wechatCorpProperties), templateManager));
+            return Mono.just(new WeixinCorpNotifier(properties.getId(),client, ValidatorUtils.tryValidate(wechatCorpProperties), templateManager));
         });
     }
 

jetlinks-components/notify-component/notify-wechat/src/main/java/org/jetlinks/community/notify/wechat/WeixinCorpNotifier.java

@@ -1,5 +1,6 @@
 package org.jetlinks.community.notify.wechat;
 
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.exception.BusinessException;
 import org.jetlinks.core.Values;
@@ -22,24 +23,27 @@ import java.util.concurrent.atomic.AtomicReference;
 @Slf4j
 public class WeixinCorpNotifier extends AbstractNotifier<WechatMessageTemplate> {
 
-    private AtomicReference<String> accessToken = new AtomicReference<>();
+    private final AtomicReference<String> accessToken = new AtomicReference<>();
 
     private long refreshTokenTime;
 
-    private long tokenTimeOut = Duration.ofSeconds(7000).toMillis();
+    private final long tokenTimeOut = Duration.ofSeconds(7000).toMillis();
 
-    private WebClient client;
+    private final WebClient client;
 
     private static final String tokenApi = "https://qyapi.weixin.qq.com/cgi-bin/gettoken";
 
     private static final String notify = "https://qyapi.weixin.qq.com/cgi-bin/message/send";
 
-    private WechatCorpProperties properties;
+    private final WechatCorpProperties properties;
+    @Getter
+    private final String notifierId;
 
-    public WeixinCorpNotifier(WebClient client, WechatCorpProperties properties, TemplateManager templateManager) {
+    public WeixinCorpNotifier(String id,WebClient client, WechatCorpProperties properties, TemplateManager templateManager) {
         super(templateManager);
         this.client = client;
         this.properties = properties;
+        this.notifierId = id;
     }
 
     @Nonnull

jetlinks-components/notify-component/notify-wechat/src/test/java/org/jetlinks/community/notify/wechat/WeixinCorpNotifierTest.java

@@ -26,7 +26,7 @@ class WeixinCorpNotifierTest {
         messageTemplate.setMessage("test"+System.currentTimeMillis());
         messageTemplate.setToUser("userId");
 
-        WeixinCorpNotifier notifier=new WeixinCorpNotifier(
+        WeixinCorpNotifier notifier=new WeixinCorpNotifier("test",
                 WebClient.builder().build(),properties,null
         );
 

jetlinks-components/timeseries-component/src/main/java/org/jetlinks/community/timeseries/micrometer/TimeSeriesMeterRegistry.java

@@ -47,7 +47,7 @@ public class TimeSeriesMeterRegistry extends StepMeterRegistry {
     public void start(ThreadFactory threadFactory) {
         super.start(threadFactory);
         timeSeriesManager.registerMetadata(MeterTimeSeriesMetadata.of(metric,keys))
-            .doOnError(e -> log.error("register metric metadata error", e))
+            .doOnError(e -> log.error("register metric [{}] metadata error", metric.getId(), e))
             .subscribe((r) -> log.error("register metric [{}] metadata success", metric.getId()));
     }
 

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

@@ -0,0 +1,17 @@
+package org.jetlinks.community.device.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.jetlinks.community.device.enums.DeviceState;
+
+@Getter
+@Setter
+@AllArgsConstructor(staticName = "of")
+@NoArgsConstructor
+public class DeviceStateInfo {
+    private String deviceId;
+
+    private DeviceState state;
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/message/DeviceBatchOperationSubscriptionProvider.java

@@ -0,0 +1,90 @@
+package org.jetlinks.community.device.message;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import org.hswebframework.web.api.crud.entity.QueryParamEntity;
+import org.jetlinks.community.device.entity.DeviceInstanceEntity;
+import org.jetlinks.community.device.service.LocalDeviceInstanceService;
+import org.jetlinks.community.gateway.external.SubscribeRequest;
+import org.jetlinks.community.gateway.external.SubscriptionProvider;
+import org.jetlinks.supports.utils.MqttTopicUtils;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+import java.util.Map;
+
+@Component
+public class DeviceBatchOperationSubscriptionProvider implements SubscriptionProvider {
+
+    private final LocalDeviceInstanceService instanceService;
+
+    public DeviceBatchOperationSubscriptionProvider(LocalDeviceInstanceService instanceService) {
+        this.instanceService = instanceService;
+    }
+
+    @Override
+    public String id() {
+        return "device-batch-operator";
+    }
+
+    @Override
+    public String name() {
+        return "设备批量操作";
+    }
+
+    @Override
+    public String[] getTopicPattern() {
+        return new String[]{
+            "/device-batch/*"
+        };
+    }
+
+    @Override
+    public Flux<?> subscribe(SubscribeRequest request) {
+        String topic = request.getTopic();
+
+        @SuppressWarnings("all")
+        QueryParamEntity queryParamEntity = request.get("query")
+            .map(json -> {
+                if (json instanceof Map) {
+                    return new JSONObject(((Map<String, Object>) json));
+                } else {
+                    return JSON.parseObject(String.valueOf(json));
+                }
+            }).map(json -> json.toJavaObject(QueryParamEntity.class))
+            .orElseGet(QueryParamEntity::new);
+
+
+        Map<String, String> var = MqttTopicUtils.getPathVariables("/device-batch/{type}", topic);
+        String type = var.get("type");
+
+        switch (type) {
+            case "state-sync":
+                return handleStateSync(queryParamEntity);
+            case "deploy":
+                return handleDeploy(queryParamEntity);
+
+            default:
+                return Flux.error(new IllegalArgumentException("不支持的类型:" + type));
+        }
+
+    }
+
+    private Flux<?> handleDeploy(QueryParamEntity queryParamEntity) {
+
+        return instanceService
+            .query(queryParamEntity.noPaging().includes("id"))
+            .as(instanceService::deploy);
+    }
+
+    private Flux<?> handleStateSync(QueryParamEntity queryParamEntity) {
+
+        return instanceService.query(queryParamEntity.noPaging().includes("id"))
+            .map(DeviceInstanceEntity::getId)
+            .buffer(200)
+            .publishOn(Schedulers.single())
+            .concatMap(flux -> instanceService.syncStateBatch(Flux.just(flux), true))
+            .flatMap(Flux::fromIterable);
+    }
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/message/DeviceCurrentStateSubscriptionProvider.java

@@ -0,0 +1,57 @@
+package org.jetlinks.community.device.message;
+
+import lombok.AllArgsConstructor;
+import org.jetlinks.community.device.entity.DeviceInstanceEntity;
+import org.jetlinks.community.device.service.LocalDeviceInstanceService;
+import org.jetlinks.community.gateway.external.SubscribeRequest;
+import org.jetlinks.community.gateway.external.SubscriptionProvider;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+@Component
+@AllArgsConstructor
+public class DeviceCurrentStateSubscriptionProvider implements SubscriptionProvider {
+
+    private final LocalDeviceInstanceService instanceService;
+
+    @Override
+    public String id() {
+        return "device-state-subscriber";
+    }
+
+    @Override
+    public String name() {
+        return "设备当前状态消息";
+    }
+
+    @Override
+    public String[] getTopicPattern() {
+        return new String[]{
+            "/device-current-state"
+        };
+    }
+
+    @Override
+    @SuppressWarnings("all")
+    public Flux<Map<String, Object>> subscribe(SubscribeRequest request) {
+        List<String> deviceId = request.get("deviceId")
+            .map(List.class::cast)
+            .orElseThrow(() -> new IllegalArgumentException("deviceId不能为空"));
+
+        return Flux
+            .fromIterable(deviceId)
+            .buffer(200)
+            .concatMap(buf -> {
+                return instanceService.createQuery()
+                    .select(DeviceInstanceEntity::getId, DeviceInstanceEntity::getState)
+                    .in(DeviceInstanceEntity::getId, buf)
+                    .fetch();
+            })
+            .map(instance -> Collections.singletonMap(instance.getId(), instance.getState().name()));
+
+    }
+}

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

@@ -352,7 +352,7 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
             .flatMap(message -> Mono.justOrEmpty(DeviceMessageUtils.convert(message))
                 .map(DeviceMessage::getDeviceId)), 800, 200, Duration.ofSeconds(2))
             .publishOn(Schedulers.parallel())
-            .concatMap(list -> syncStateBatch(Flux.just(list), false).reduce(Math::addExact))
+            .concatMap(list -> syncStateBatch(Flux.just(list), false).map(List::size))
             .onErrorContinue((err, obj) -> log.error(err.getMessage(), err))
             .subscribe((i) -> log.info("同步设备状态成功:{}", i));
     }
@@ -364,7 +364,7 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
             .switchIfEmpty(Mono.error(NotFoundException::new));
     }
 
-    public Flux<Integer> syncStateBatch(Flux<List<String>> batch, boolean force) {
+    public Flux<List<DeviceStateInfo>> syncStateBatch(Flux<List<String>> batch, boolean force) {
 
         return batch
             .concatMap(list -> Flux.fromIterable(list)
@@ -384,6 +384,7 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                 .collect(Collectors.groupingBy(Tuple2::getT1))
                 .flatMapIterable(Map::entrySet)
                 .flatMap(group -> {
+                    List<String> deviceId=group.getValue().stream().map(Tuple3::getT2).collect(Collectors.toList());
                     DeviceState state = DeviceState.of(group.getKey());
                     return Mono.zip(
                         //批量修改设备状态
@@ -391,10 +392,9 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                             .createUpdate()
                             .set(DeviceInstanceEntity::getState, state)
                             .where()
-                            .in(DeviceInstanceEntity::getId, group.getValue().stream().map(Tuple3::getT2).collect(Collectors.toList()))
+                            .in(DeviceInstanceEntity::getId,deviceId)
                             .execute()
-                            .thenReturn(group.getValue().size())//mysql下可能不会返回更新数量
-                        ,
+                            .thenReturn(group.getValue().size()),
                         //修改子设备状态
                         Flux.fromIterable(group.getValue())
                             .filter(Tuple3::getT3)
@@ -408,8 +408,8 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                                     .where()
                                     .in(DeviceInstanceEntity::getParentId, parents)
                                     .execute())
-                            .defaultIfEmpty(0),
-                        Math::addExact);
+                            .defaultIfEmpty(0))
+                        .thenReturn(deviceId.stream().map(id->DeviceStateInfo.of(id,state)).collect(Collectors.toList()));
                 }));
     }
 

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

@@ -168,7 +168,7 @@ public class DeviceInstanceController implements
             .map(DeviceInstanceEntity::getId)
             .buffer(200)
             .publishOn(Schedulers.single())
-            .concatMap(flux -> service.syncStateBatch(Flux.just(flux), true))
+            .concatMap(flux -> service.syncStateBatch(Flux.just(flux), true).map(List::size))
             .defaultIfEmpty(0);
     }
 

jetlinks-manager/notify-manager/src/main/java/org/jetlinks/community/notify/manager/entity/NotifyHistoryEntity.java

@@ -0,0 +1,89 @@
+package org.jetlinks.community.notify.manager.entity;
+
+import com.alibaba.fastjson.JSON;
+import lombok.Getter;
+import lombok.Setter;
+import org.hswebframework.ezorm.rdb.mapping.annotation.ColumnType;
+import org.hswebframework.ezorm.rdb.mapping.annotation.DefaultValue;
+import org.hswebframework.ezorm.rdb.mapping.annotation.EnumCodec;
+import org.hswebframework.ezorm.rdb.mapping.annotation.JsonCodec;
+import org.hswebframework.web.api.crud.entity.GenericEntity;
+import org.hswebframework.web.bean.FastBeanCopier;
+import org.hswebframework.web.crud.generator.Generators;
+import org.jetlinks.community.notify.event.SerializableNotifierEvent;
+import org.jetlinks.community.notify.manager.enums.NotifyState;
+
+import javax.persistence.Column;
+import javax.persistence.Index;
+import javax.persistence.Table;
+import java.sql.JDBCType;
+import java.util.Date;
+import java.util.Map;
+
+@Table(name = "notify_history", indexes = {
+    @Index(name = "idx_nt_his_notifier_id", columnList = "notifier_id")
+})
+@Getter
+@Setter
+public class NotifyHistoryEntity extends GenericEntity<String> {
+
+    private static final long serialVersionUID = -6849794470754667710L;
+
+    @Column(length = 32, nullable = false, updatable = false)
+    private String notifierId;
+
+    @Column(nullable = false)
+    @DefaultValue("success")
+    @EnumCodec
+    @ColumnType(javaType = String.class)
+    private NotifyState state;
+
+    @Column(length = 1024)
+    private String errorType;
+
+    @Column
+    @ColumnType(jdbcType = JDBCType.CLOB, javaType = String.class)
+    private String errorStack;
+
+    @Column(length = 32, nullable = false)
+    @DefaultValue("-")
+    private String templateId;
+
+    @Column
+    @ColumnType(jdbcType = JDBCType.CLOB, javaType = String.class)
+    private String template;
+
+    @Column
+    @ColumnType(jdbcType = JDBCType.CLOB, javaType = String.class)
+    @JsonCodec
+    private Map<String, Object> context;
+
+    @Column(length = 32, nullable = false)
+    private String provider;
+
+    @Column(length = 32, nullable = false)
+    private String notifyType;
+
+    @Column
+    @DefaultValue(generator = Generators.CURRENT_TIME)
+    private Date notifyTime;
+
+    @Column
+    @DefaultValue("0")
+    private Integer retryTimes;
+
+    public static NotifyHistoryEntity of(SerializableNotifierEvent event) {
+       NotifyHistoryEntity entity = FastBeanCopier.copy(event, new NotifyHistoryEntity());
+        if (null != event.getTemplate()) {
+            entity.setTemplate(JSON.toJSONString(event.getTemplate()));
+        }
+        if (event.isSuccess()) {
+            entity.setState(NotifyState.success);
+        } else {
+            entity.setErrorStack(event.getCause());
+            entity.setState(NotifyState.error);
+        }
+        return entity;
+    }
+
+}

jetlinks-manager/notify-manager/src/main/java/org/jetlinks/community/notify/manager/enums/NotifyState.java

@@ -0,0 +1,20 @@
+package org.jetlinks.community.notify.manager.enums;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.hswebframework.web.dict.EnumDict;
+
+@Getter
+@AllArgsConstructor
+public enum NotifyState implements EnumDict<String> {
+
+    success("成功"),
+    error("失败");
+
+    private final String text;
+
+    @Override
+    public String getValue() {
+        return name();
+    }
+}

jetlinks-manager/notify-manager/src/main/java/org/jetlinks/community/notify/manager/service/NotifyHistoryService.java

@@ -0,0 +1,22 @@
+package org.jetlinks.community.notify.manager.service;
+
+import org.hswebframework.web.crud.service.GenericReactiveCrudService;
+import org.jetlinks.community.gateway.annotation.Subscribe;
+import org.jetlinks.community.notify.event.SerializableNotifierEvent;
+import org.jetlinks.community.notify.manager.entity.NotifyHistoryEntity;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import reactor.core.publisher.Mono;
+
+@Service
+public class NotifyHistoryService extends GenericReactiveCrudService<NotifyHistoryEntity, String> {
+
+
+    @Subscribe("/notify/**")
+    @Transactional(propagation = Propagation.NEVER)
+    public Mono<Void> handleNotify(SerializableNotifierEvent event) {
+        return insert(Mono.just(NotifyHistoryEntity.of(event))).then();
+    }
+
+}

jetlinks-manager/notify-manager/src/main/java/org/jetlinks/community/notify/manager/web/NotifierHistoryController.java

@@ -0,0 +1,25 @@
+package org.jetlinks.community.notify.manager.web;
+
+import org.hswebframework.web.authorization.annotation.Resource;
+import org.hswebframework.web.crud.web.reactive.ReactiveServiceQueryController;
+import org.jetlinks.community.notify.manager.entity.NotifyHistoryEntity;
+import org.jetlinks.community.notify.manager.service.NotifyHistoryService;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/notify/history")
+@Resource(id = "notifier", name = "通知管理")
+public class NotifierHistoryController implements ReactiveServiceQueryController<NotifyHistoryEntity, String> {
+
+    private final NotifyHistoryService historyService;
+
+    public NotifierHistoryController(NotifyHistoryService historyService) {
+        this.historyService = historyService;
+    }
+
+    @Override
+    public NotifyHistoryService getService() {
+        return historyService;
+    }
+}

jetlinks-manager/rule-engine-manager/src/main/java/org/jetlinks/community/rule/engine/entity/RuleModelEntity.java

@@ -43,10 +43,10 @@ public class RuleModelEntity extends GenericEntity<String> implements RecordCrea
     @Column(name = "version",nullable = false)
     private Integer version;
 
-    @Column(name = "creator_id")
+    @Column(name = "creator_id",updatable = false)
     private String creatorId;
 
-    @Column(name = "create_time")
+    @Column(name = "create_time",updatable = false)
     private Long createTime;
 
     @Column(name = "modifier_id")

jetlinks-standalone/src/main/resources/application-embedded.yml

@@ -5,10 +5,10 @@ spring:
     embedded:
       enabled: true  # 使用内置的redis,不建议在生产环境中使用.
       host: 127.0.0.1
-      port: 6379
+      port: 6370
       data-path: ./data/redis
     host: 127.0.0.1
-    port: 6379
+    port: 6370
     lettuce:
       pool:
         max-active: 1024
@@ -26,11 +26,11 @@ elasticsearch:
   embedded:
     enabled: true # 为true时使用内嵌的elasticsearch,不建议在生产环境中使用
     data-path: ./data/elasticsearch
-    port: 9200
+    port: 9201
     host: 0.0.0.0
   client:
     host: localhost
-    port: 9200
+    port: 9201
     max-conn-total: 128
     connect-timeout: 5000
     socket-timeout: 5000

pom.xml

@@ -19,7 +19,8 @@
         <spring.boot.version>2.2.5.RELEASE</spring.boot.version>
         <java.version>1.8</java.version>
         <project.build.jdk>${java.version}</project.build.jdk>
-        <hsweb.framework.version>4.0.1</hsweb.framework.version>
+        <hsweb.framework.version>4.0.3-SNAPSHOT</hsweb.framework.version>
+        <easyorm.version>4.0.3-SNAPSHOT</easyorm.version>
         <hsweb.expands.version>3.0.2</hsweb.expands.version>
         <jetlinks.version>1.0.3-SNAPSHOT</jetlinks.version>
         <r2dbc.version>Arabba-RELEASE</r2dbc.version>
@@ -258,7 +259,22 @@
                 <artifactId>elasticsearch-rest-high-level-client</artifactId>
                 <version>${elasticsearch.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.hswebframework</groupId>
+                <artifactId>hsweb-easy-orm-core</artifactId>
+                <version>${easyorm.version}</version>
+            </dependency>
 
+            <dependency>
+                <groupId>org.hswebframework</groupId>
+                <artifactId>hsweb-easy-orm-rdb</artifactId>
+                <version>${easyorm.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.hswebframework</groupId>
+                <artifactId>hsweb-easy-orm-elasticsearch</artifactId>
+                <version>${easyorm.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>