Implementation of Generator node and improvements

@@ -15,9 +15,12 @@
  */
 package org.thingsboard.server.actors.ruleChain;
 
+import akka.actor.ActorContext;
+import akka.actor.ActorRef;
 import org.thingsboard.rule.engine.api.ListeningExecutor;
 import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.cluster.ServerAddress;
 import org.thingsboard.server.dao.alarm.AlarmService;
@@ -30,8 +33,10 @@ import org.thingsboard.server.dao.relation.RelationService;
 import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
+import scala.concurrent.duration.Duration;
 
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Created by ashvayka on 19.03.18.
@@ -61,7 +66,12 @@ class DefaultTbContext implements TbContext {
 
     @Override
     public void tellSelf(TbMsg msg, long delayMs) {
-        throw new RuntimeException("Not Implemented!");
+        //TODO: add persistence layer
+        scheduleMsgWithDelay(new RuleNodeToSelfMsg(msg), delayMs, nodeCtx.getSelfActor());
+    }
+
+    private void scheduleMsgWithDelay(Object msg, long delayInMs, ActorRef target) {
+        mainCtx.getScheduler().scheduleOnce(Duration.create(delayInMs, TimeUnit.MILLISECONDS), target, msg, mainCtx.getActorSystem().dispatcher(), nodeCtx.getSelfActor());
     }
 
     @Override
@@ -93,6 +103,11 @@ class DefaultTbContext implements TbContext {
     }
 
     @Override
+    public RuleNodeId getSelfId() {
+        return nodeCtx.getSelf().getId();
+    }
+
+    @Override
     public void tellNext(TbMsg msg, Set<String> relationTypes) {
         relationTypes.forEach(type -> tellNext(msg, type));
     }

@@ -47,12 +47,25 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa
             case RULE_TO_SELF_ERROR_MSG:
                 onRuleNodeToSelfErrorMsg((RuleNodeToSelfErrorMsg) msg);
                 break;
+            case RULE_TO_SELF_MSG:
+                onRuleNodeToSelfMsg((RuleNodeToSelfMsg) msg);
+                break;
             default:
                 return false;
         }
         return true;
     }
 
+    private void onRuleNodeToSelfMsg(RuleNodeToSelfMsg msg) {
+        logger.debug("[{}] Going to process rule msg: {}", id, msg.getMsg());
+        try {
+            processor.onRuleToSelfMsg(msg);
+            increaseMessagesProcessedCount();
+        } catch (Exception e) {
+            logAndPersist("onRuleMsg", e);
+        }
+    }
+
     private void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg msg) {
         logger.debug("[{}] Going to process rule msg: {}", id, msg.getMsg());
         try {

@@ -18,9 +18,10 @@ package org.thingsboard.server.actors.ruleChain;
 import akka.actor.ActorContext;
 import akka.actor.ActorRef;
 import akka.event.LoggingAdapter;
+import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbNode;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
-import org.thingsboard.rule.engine.api.TbNodeState;
+import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
 import org.thingsboard.server.common.data.id.RuleChainId;
@@ -31,6 +32,8 @@ import org.thingsboard.server.common.data.rule.RuleNode;
 import org.thingsboard.server.common.msg.cluster.ClusterEventMsg;
 import org.thingsboard.server.dao.rule.RuleChainService;
 
+import java.util.concurrent.ExecutionException;
+
 /**
  * @author Andrew Shvayka
  */
@@ -41,6 +44,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     private final RuleChainService service;
     private RuleNode ruleNode;
     private TbNode tbNode;
+    private TbContext defaultCtx;
 
     RuleNodeActorMessageProcessor(TenantId tenantId, RuleChainId ruleChainId, RuleNodeId ruleNodeId, ActorSystemContext systemContext
             , LoggingAdapter logger, ActorRef parent, ActorRef self) {
@@ -49,6 +53,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
         this.self = self;
         this.service = systemContext.getRuleChainService();
         this.ruleNode = systemContext.getRuleChainService().findRuleNodeById(entityId);
+        this.defaultCtx = new DefaultTbContext(systemContext, new RuleNodeCtx(tenantId, parent, self, ruleNode));
     }
 
     @Override
@@ -80,6 +85,14 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
 
     }
 
+    public void onRuleToSelfMsg(RuleNodeToSelfMsg msg) throws Exception {
+        checkActive();
+        if (ruleNode.isDebugMode()) {
+            systemContext.persistDebugInput(tenantId, entityId, msg.getMsg());
+        }
+        tbNode.onMsg(defaultCtx, msg.getMsg());
+    }
+
     void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg msg) throws Exception {
         checkActive();
         if (ruleNode.isDebugMode()) {
@@ -91,9 +104,8 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     private TbNode initComponent(RuleNode ruleNode) throws Exception {
         Class<?> componentClazz = Class.forName(ruleNode.getType());
         TbNode tbNode = (TbNode) (componentClazz.newInstance());
-        tbNode.init(new TbNodeConfiguration(ruleNode.getConfiguration()), new TbNodeState());
+        tbNode.init(defaultCtx, new TbNodeConfiguration(ruleNode.getConfiguration()));
         return tbNode;
     }
 
-
 }

+/**
+ * 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.server.actors.ruleChain;
+
+import lombok.Data;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.TbActorMsg;
+import org.thingsboard.server.common.msg.TbMsg;
+
+/**
+ * Created by ashvayka on 19.03.18.
+ */
+@Data
+final class RuleNodeToSelfMsg implements TbActorMsg {
+
+    private final TbMsg msg;
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.RULE_TO_SELF_MSG;
+    }
+
+}

@@ -54,4 +54,9 @@ public enum MsgType {
      */
     RULE_TO_SELF_ERROR_MSG,
 
+    /**
+     * Message that is sent by RuleActor implementation to RuleActor itself to process the message.
+     */
+    RULE_TO_SELF_MSG,
+
 }

@@ -15,8 +15,8 @@
  */
 package org.thingsboard.rule.engine.api;
 
-public interface NodeConfiguration {
+public interface NodeConfiguration<T extends NodeConfiguration> {
 
-    NodeConfiguration defaultConfiguration();
+    T defaultConfiguration();
 
 }

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.rule.engine.api;
 
+import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.cluster.ServerAddress;
 import org.thingsboard.server.dao.alarm.AlarmService;
@@ -55,6 +56,8 @@ public interface TbContext {
 
     void tellError(TbMsg msg, Throwable th);
 
+    RuleNodeId getSelfId();
+
     AttributesService getAttributesService();
 
     CustomerService getCustomerService();

@@ -24,7 +24,7 @@ import java.util.concurrent.ExecutionException;
  */
 public interface TbNode {
 
-    void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException;
+    void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException;
 
     void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException;
 

+/**
+ * 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.debug;
+
+import com.datastax.driver.core.utils.UUIDs;
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.rule.engine.TbNodeUtils;
+import org.thingsboard.rule.engine.api.ListeningExecutor;
+import org.thingsboard.rule.engine.api.RuleNode;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNode;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.rule.engine.filter.TbJsFilterNodeConfiguration;
+import org.thingsboard.rule.engine.js.NashornJsEngine;
+import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+
+import javax.script.Bindings;
+
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+
+import static org.thingsboard.rule.engine.DonAsynchron.withCallback;
+
+@Slf4j
+@RuleNode(
+        type = ComponentType.ACTION,
+        name = "generator",
+        configClazz = TbMsgGeneratorNodeConfiguration.class,
+        nodeDescription = "Periodically generates messages",
+        nodeDetails = "Generates messages with configurable period. ",
+        inEnabled = false
+)
+
+public class TbMsgGeneratorNode implements TbNode {
+
+    public static final String TB_MSG_GENERATOR_NODE_MSG = "TbMsgGeneratorNodeMsg";
+
+    private TbMsgGeneratorNodeConfiguration config;
+    private long delay;
+
+    @Override
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+        this.config = TbNodeUtils.convert(configuration, TbMsgGeneratorNodeConfiguration.class);
+        this.delay = TimeUnit.SECONDS.toMillis(config.getPeriodInSeconds());
+        ctx.tellSelf(newTickMsg(ctx), delay);
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) {
+        if (msg.getType().equals(TB_MSG_GENERATOR_NODE_MSG)) {
+            TbMsgMetaData metaData = new TbMsgMetaData();
+            if (config.getMsgMetaData() != null) {
+                config.getMsgMetaData().forEach(metaData::putValue);
+            }
+            ctx.tellNext(new TbMsg(UUIDs.timeBased(), config.getMsgType(), ctx.getSelfId(), metaData, config.getMsgBody().getBytes(StandardCharsets.UTF_8)));
+            ctx.tellSelf(newTickMsg(ctx), delay);
+        }
+    }
+
+    private TbMsg newTickMsg(TbContext ctx) {
+        return new TbMsg(UUIDs.timeBased(), TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), new byte[]{});
+    }
+
+    @Override
+    public void destroy() {
+    }
+}

+/**
+ * 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.debug;
+
+import lombok.Data;
+import org.thingsboard.rule.engine.api.NodeConfiguration;
+import java.util.Map;
+
+@Data
+public class TbMsgGeneratorNodeConfiguration implements NodeConfiguration<TbMsgGeneratorNodeConfiguration> {
+
+    private int msgCount;
+    private int periodInSeconds;
+    private String msgType;
+    private String msgBody;
+    private Map<String, String> msgMetaData;
+
+    @Override
+    public TbMsgGeneratorNodeConfiguration defaultConfiguration() {
+        TbMsgGeneratorNodeConfiguration configuration = new TbMsgGeneratorNodeConfiguration();
+        configuration.setMsgCount(0);
+        configuration.setPeriodInSeconds(1);
+        configuration.setMsgType("DebugMsg");
+        configuration.setMsgBody("{}");
+        return configuration;
+    }
+}

@@ -29,7 +29,7 @@ import static org.thingsboard.rule.engine.DonAsynchron.withCallback;
 @Slf4j
 @RuleNode(
         type = ComponentType.FILTER,
-        name = "script", relationTypes = {"True", "False", "Failure"},
+        name = "script", relationTypes = {"True", "False"},
         configClazz = TbJsFilterNodeConfiguration.class,
         nodeDescription = "Filter incoming messages using JS script",
         nodeDetails = "Evaluate incoming Message with configured JS condition. " +
@@ -45,7 +45,7 @@ public class TbJsFilterNode implements TbNode {
     private NashornJsEngine jsEngine;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbJsFilterNodeConfiguration.class);
         this.jsEngine = new NashornJsEngine(config.getJsScript(), "Filter");
     }

@@ -19,7 +19,7 @@ import lombok.Data;
 import org.thingsboard.rule.engine.api.NodeConfiguration;
 
 @Data
-public class TbJsFilterNodeConfiguration implements NodeConfiguration {
+public class TbJsFilterNodeConfiguration implements NodeConfiguration<TbJsFilterNodeConfiguration> {
 
     private String jsScript;
 

@@ -45,7 +45,7 @@ public class TbJsSwitchNode implements TbNode {
     private NashornJsEngine jsEngine;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbJsSwitchNodeConfiguration.class);
         this.jsEngine = new NashornJsEngine(config.getJsScript(), "Switch");
     }

@@ -22,7 +22,7 @@ import org.thingsboard.rule.engine.api.NodeConfiguration;
 import java.util.Set;
 
 @Data
-public class TbJsSwitchNodeConfiguration implements NodeConfiguration {
+public class TbJsSwitchNodeConfiguration implements NodeConfiguration<TbJsSwitchNodeConfiguration> {
 
     private String jsScript;
 

@@ -29,6 +29,7 @@ import org.thingsboard.server.common.msg.TbMsg;
         type = ComponentType.FILTER,
         name = "message type",
         configClazz = TbMsgTypeFilterNodeConfiguration.class,
+        relationTypes = {"True", "False"},
         nodeDescription = "Filter incoming messages by Message Type",
         nodeDetails = "Evaluate incoming Message with configured JS condition. " +
                 "If incoming MessageType is expected - send Message via <b>Success</b> chain, otherwise <b>Failure</b> chain is used.",
@@ -39,7 +40,7 @@ public class TbMsgTypeFilterNode implements TbNode {
     TbMsgTypeFilterNodeConfiguration config;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbMsgTypeFilterNodeConfiguration.class);
     }
 

@@ -26,7 +26,7 @@ import java.util.List;
  * Created by ashvayka on 19.01.18.
  */
 @Data
-public class TbMsgTypeFilterNodeConfiguration implements NodeConfiguration {
+public class TbMsgTypeFilterNodeConfiguration implements NodeConfiguration<TbMsgTypeFilterNodeConfiguration> {
 
     private List<String> messageTypes;
 

@@ -37,7 +37,7 @@ public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode
     private TbGetEntityAttrNodeConfiguration config;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbGetEntityAttrNodeConfiguration.class);
     }
 

@@ -49,7 +49,7 @@ public class TbGetAttributesNode implements TbNode {
     private TbGetAttributesNodeConfiguration config;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbGetAttributesNodeConfiguration.class);
     }
 

@@ -25,7 +25,7 @@ import java.util.List;
  * Created by ashvayka on 19.01.18.
  */
 @Data
-public class TbGetAttributesNodeConfiguration implements NodeConfiguration {
+public class TbGetAttributesNodeConfiguration implements NodeConfiguration<TbGetAttributesNodeConfiguration> {
 
     private List<String> clientAttributeNames;
     private List<String> sharedAttributeNames;

@@ -23,7 +23,7 @@ import java.util.Map;
 import java.util.Optional;
 
 @Data
-public class TbGetEntityAttrNodeConfiguration implements NodeConfiguration {
+public class TbGetEntityAttrNodeConfiguration implements NodeConfiguration<TbGetEntityAttrNodeConfiguration> {
 
     private Map<String, String> attrMapping;
     private boolean isTelemetry = false;

@@ -38,7 +38,7 @@ public class TbGetRelatedAttributeNode extends TbEntityGetAttrNode<EntityId> {
     private TbGetRelatedAttrNodeConfiguration config;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext context, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbGetRelatedAttrNodeConfiguration.class);
         setConfig(config);
     }

@@ -32,7 +32,7 @@ public abstract class TbAbstractTransformNode implements TbNode {
     private TbTransformNodeConfiguration config;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext context, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbTransformNodeConfiguration.class);
     }
 

@@ -49,7 +49,7 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode {
     private TbChangeOriginatorNodeConfiguration config;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbChangeOriginatorNodeConfiguration.class);
         validateConfig(config);
         setConfig(config);

@@ -38,7 +38,7 @@ public class TbTransformMsgNode extends TbAbstractTransformNode {
     private NashornJsEngine jsEngine;
 
     @Override
-    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbTransformMsgNodeConfiguration.class);
         this.jsEngine = new NashornJsEngine(config.getJsScript(), "Transform");
         setConfig(config);

@@ -144,7 +144,7 @@ public class TbJsFilterNodeTest {
         TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config));
 
         node = new TbJsFilterNode();
-        node.init(nodeConfiguration, null);
+        node.init(null, nodeConfiguration);
     }
 
     private void mockJsExecutor() {

@@ -107,7 +107,7 @@ public class TbJsSwitchNodeTest {
         TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config));
 
         node = new TbJsSwitchNode();
-        node.init(nodeConfiguration, null);
+        node.init(null, nodeConfiguration);
     }
 
     private void mockJsExecutor() {

@@ -88,7 +88,7 @@ public class TbGetCustomerAttributeNodeTest {
         TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config));
 
         node = new TbGetCustomerAttributeNode();
-        node.init(nodeConfiguration, null);
+        node.init(null, nodeConfiguration);
     }
 
     @Test
@@ -226,7 +226,7 @@ public class TbGetCustomerAttributeNodeTest {
         TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config));
 
         node = new TbGetCustomerAttributeNode();
-        node.init(nodeConfiguration, null);
+        node.init(null, nodeConfiguration);
 
 
         DeviceId deviceId = new DeviceId(UUIDs.timeBased());

@@ -119,6 +119,6 @@ public class TbChangeOriginatorNodeTest {
         TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config));
 
         node = new TbChangeOriginatorNode();
-        node.init(nodeConfiguration, null);
+        node.init(null, nodeConfiguration);
     }
 }
\ No newline at end of file

@@ -114,7 +114,7 @@ public class TbTransformMsgNodeTest {
         TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config));
 
         node = new TbTransformMsgNode();
-        node.init(nodeConfiguration, null);
+        node.init(null, nodeConfiguration);
     }
 
     private void mockJsExecutor() {