Msg Count Node

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -56,6 +56,7 @@ import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
+import org.thingsboard.server.kafka.TbNodeIdProvider;
 import org.thingsboard.server.service.cluster.discovery.DiscoveryService;
 import org.thingsboard.server.service.cluster.routing.ClusterRoutingService;
 import org.thingsboard.server.service.cluster.rpc.ClusterRpcService;
@@ -276,6 +277,10 @@ public class ActorSystemContext {
     @Setter
     private ActorSystem actorSystem;
 
+    @Autowired
+    @Getter
+    private TbNodeIdProvider nodeIdProvider;
+
     @Getter
     @Setter
     private ActorRef appActor;

application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java

@@ -161,6 +161,11 @@ class DefaultTbContext implements TbContext {
         return new RuleNodeJsScriptEngine(mainCtx.getJsSandbox(), nodeCtx.getSelf().getId(), script, argNames);
     }
 
+    @Override
+    public String getNodeId() {
+        return mainCtx.getNodeIdProvider().getNodeId();
+    }
+
     @Override
     public AttributesService getAttributesService() {
         return mainCtx.getAttributesService();

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java

@@ -34,6 +34,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
 
 import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
 
 /**
  * Created by ashvayka on 13.01.18.
@@ -98,4 +99,5 @@ public interface TbContext {
 
     ScriptEngine createJsScriptEngine(String script, String... argNames);
 
+    String getNodeId();
 }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java

@@ -0,0 +1,101 @@
+/**
+ * Copyright © 2016-2018 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.action;
+
+import com.datastax.driver.core.utils.UUIDs;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.rule.engine.api.*;
+import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgDataType;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.session.SessionMsgType;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
+
+@Slf4j
+@RuleNode(
+        type = ComponentType.ACTION,
+        name = "message count",
+        configClazz = TbMsgCountNodeConfiguration.class,
+        nodeDescription = "Count incoming messages",
+        nodeDetails = "Count incoming messages for specified interval and produces POST_TELEMETRY_REQUEST msg with messages count",
+        icon = "functions",
+        uiResources = {"static/rulenode/rulenode-core-config.js"},
+        configDirective = "tbActionNodeMsgCountConfig"
+)
+public class TbMsgCountNode implements TbNode {
+
+    private static final String TB_MSG_COUNT_NODE_MSG = "TbMsgCountNodeMsg";
+
+    private AtomicLong messagesProcessed = new AtomicLong(0);
+    private final Gson gson = new Gson();
+    private UUID nextTickId;
+    private long delay;
+    private String telemetryPrefix;
+    private long lastScheduledTs;
+
+    @Override
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+        TbMsgCountNodeConfiguration config = TbNodeUtils.convert(configuration, TbMsgCountNodeConfiguration.class);
+        this.delay = TimeUnit.SECONDS.toMillis(config.getInterval());
+        this.telemetryPrefix = config.getTelemetryPrefix();
+        scheduleTickMsg(ctx);
+
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) {
+        if (msg.getType().equals(TB_MSG_COUNT_NODE_MSG) && msg.getId().equals(nextTickId)) {
+            JsonObject telemetryJson = new JsonObject();
+            telemetryJson.addProperty(this.telemetryPrefix + "_" + ctx.getNodeId(), messagesProcessed.longValue());
+
+            messagesProcessed = new AtomicLong(0);
+
+            TbMsgMetaData metaData = new TbMsgMetaData();
+            metaData.putValue("delta", Long.toString(System.currentTimeMillis() - lastScheduledTs + delay));
+
+            TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), ctx.getTenantId(), metaData, TbMsgDataType.JSON, gson.toJson(telemetryJson), null, null, 0L);
+            ctx.tellNext(tbMsg, SUCCESS);
+            scheduleTickMsg(ctx);
+        } else {
+            messagesProcessed.incrementAndGet();
+        }
+    }
+
+    private void scheduleTickMsg(TbContext ctx) {
+        long curTs = System.currentTimeMillis();
+        if (lastScheduledTs == 0L) {
+            lastScheduledTs = curTs;
+        }
+        lastScheduledTs = lastScheduledTs + delay;
+        long curDelay = Math.max(0L, (lastScheduledTs - curTs));
+        TbMsg tickMsg = ctx.newMsg(TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), "");
+        nextTickId = tickMsg.getId();
+        ctx.tellSelf(tickMsg, curDelay);
+    }
+
+    @Override
+    public void destroy() {
+    }
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNodeConfiguration.java

@@ -0,0 +1,34 @@
+/**
+ * Copyright © 2016-2018 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.action;
+
+import lombok.Data;
+import org.thingsboard.rule.engine.api.NodeConfiguration;
+
+@Data
+public class TbMsgCountNodeConfiguration implements NodeConfiguration<TbMsgCountNodeConfiguration> {
+
+    private String telemetryPrefix;
+    private int interval;
+
+    @Override
+    public TbMsgCountNodeConfiguration defaultConfiguration() {
+        TbMsgCountNodeConfiguration configuration = new TbMsgCountNodeConfiguration();
+        configuration.setInterval(1);
+        configuration.setTelemetryPrefix("messageCount");
+        return configuration;
+    }
+}

ui/src/app/api/telemetry-websocket.service.js

@@ -141,7 +141,7 @@ function TelemetryWebsocketService($rootScope, $websocket, $timeout, $window, $m
     }
 
     function onClose (closeEvent) {
-        if (closeEvent && closeEvent.code > 1000) {
+        if (closeEvent && closeEvent.code > 1000 && closeEvent.code !== 1006) {
             showWsError(closeEvent.code, closeEvent.reason);
         }
         isOpening = false;