Check Relations Node and Originator Fields

@@ -43,6 +43,7 @@ import org.thingsboard.server.dao.customer.CustomerService;
 import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.dao.relation.RelationService;
 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.service.script.RuleNodeJsScriptEngine;
@@ -167,6 +168,11 @@ class DefaultTbContext implements TbContext {
     }
 
     @Override
+    public TenantService getTenantService() {
+        return mainCtx.getTenantService();
+    }
+
+    @Override
     public UserService getUserService() {
         return mainCtx.getUserService();
     }

+/**
+ * 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.common.data;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import org.thingsboard.server.common.data.id.EntityId;
+
+/**
+ * Created by ashvayka on 01.06.18.
+ */
+@Data
+@AllArgsConstructor
+public class EntityFieldsData {
+
+    public static final String DEFAULT = "default";
+
+    private final EntityId entityId;
+    private final String name;
+    private final String type;
+
+    public EntityFieldsData(EntityId entityId, String name) {
+        this(entityId, name, DEFAULT);
+    }
+}

@@ -28,6 +28,7 @@ import org.thingsboard.server.dao.customer.CustomerService;
 import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.dao.relation.RelationService;
 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;
 
@@ -62,6 +63,8 @@ public interface TbContext {
 
     CustomerService getCustomerService();
 
+    TenantService getTenantService();
+
     UserService getUserService();
 
     AssetService getAssetService();

+/**
+ * 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.filter;
+
+import lombok.extern.slf4j.Slf4j;
+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.api.util.TbNodeUtils;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.EntityIdFactory;
+import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.data.relation.EntitySearchDirection;
+import org.thingsboard.server.common.data.relation.RelationTypeGroup;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import javax.management.relation.RelationType;
+
+import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
+import static org.thingsboard.rule.engine.api.util.DonAsynchron.withCallback;
+
+/**
+ * Created by ashvayka on 19.01.18.
+ */
+@Slf4j
+@RuleNode(
+        type = ComponentType.FILTER,
+        name = "check relation",
+        configClazz = TbCheckRelationNodeConfiguration.class,
+        relationTypes = {"True", "False"},
+        nodeDescription = "Checks the relation from the selected entity to originator of the message by type and direction",
+        nodeDetails = "If incoming MessageType is expected - send Message via <b>True</b> chain, otherwise <b>False</b> chain is used.")
+public class TbCheckRelationNode implements TbNode {
+
+    private TbCheckRelationNodeConfiguration config;
+
+    @Override
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+        this.config = TbNodeUtils.convert(configuration, TbCheckRelationNodeConfiguration.class);
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
+        EntityId from;
+        EntityId to;
+        if (EntitySearchDirection.FROM.name().equals(config.getDirection())) {
+            from = EntityIdFactory.getByTypeAndId(config.getEntityType(), config.getEntityId());
+            to = msg.getOriginator();
+        } else {
+            to = EntityIdFactory.getByTypeAndId(config.getEntityType(), config.getEntityId());
+            from = msg.getOriginator();
+        }
+        withCallback(ctx.getRelationService().checkRelation(from, to, config.getRelationType(), RelationTypeGroup.COMMON),
+                filterResult -> ctx.tellNext(msg, filterResult ? "True" : "False"), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
+    }
+
+    @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.filter;
+
+import lombok.Data;
+import org.thingsboard.rule.engine.api.NodeConfiguration;
+import org.thingsboard.server.common.data.relation.EntitySearchDirection;
+import org.thingsboard.server.common.msg.session.SessionMsgType;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Created by ashvayka on 19.01.18.
+ */
+@Data
+public class TbCheckRelationNodeConfiguration implements NodeConfiguration<TbCheckRelationNodeConfiguration> {
+
+    private String direction;
+    private String entityId;
+    private String entityType;
+    private String relationType;
+
+    @Override
+    public TbCheckRelationNodeConfiguration defaultConfiguration() {
+        TbCheckRelationNodeConfiguration configuration = new TbCheckRelationNodeConfiguration();
+        configuration.setDirection(EntitySearchDirection.FROM.name());
+        configuration.setRelationType("Contains");
+        return configuration;
+    }
+}

@@ -51,7 +51,7 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
     public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
         try {
             withCallback(
-                    findEntityAsync(ctx, msg.getOriginator()),
+                    findEntityIdAsync(ctx, msg.getOriginator()),
                     entityId -> safePutAttributes(ctx, msg, entityId),
                     t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
         } catch (Throwable th) {
@@ -112,5 +112,5 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
 
     }
 
-    protected abstract ListenableFuture<T> findEntityAsync(TbContext ctx, EntityId originator);
+    protected abstract ListenableFuture<T> findEntityIdAsync(TbContext ctx, EntityId originator);
 }

@@ -48,7 +48,7 @@ public class TbGetAttributesNode extends TbAbstractGetAttributesNode<TbGetAttrib
     }
 
     @Override
-    protected ListenableFuture<EntityId> findEntityAsync(TbContext ctx, EntityId originator) {
+    protected ListenableFuture<EntityId> findEntityIdAsync(TbContext ctx, EntityId originator) {
         return Futures.immediateFuture(originator);
     }
 }

@@ -46,7 +46,7 @@ public class TbGetDeviceAttrNode extends TbAbstractGetAttributesNode<TbGetDevice
     }
 
     @Override
-    protected ListenableFuture<DeviceId> findEntityAsync(TbContext ctx, EntityId originator) {
+    protected ListenableFuture<DeviceId> findEntityIdAsync(TbContext ctx, EntityId originator) {
         return EntitiesRelatedDeviceIdAsyncLoader.findDeviceAsync(ctx, originator, config.getDeviceRelationsQuery());
     }
 }

+/**
+ * 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.metadata;
+
+import lombok.Data;
+import org.thingsboard.rule.engine.api.NodeConfiguration;
+
+@Data
+public class TbGetOriginatorFieldsConfiguration implements NodeConfiguration<TbGetOriginatorFieldsConfiguration> {
+
+    private boolean fetchName;
+    private String nameMetadataKey;
+    private boolean fetchType;
+    private String typeMetadataKey;
+
+    @Override
+    public TbGetOriginatorFieldsConfiguration defaultConfiguration() {
+        TbGetOriginatorFieldsConfiguration configuration = new TbGetOriginatorFieldsConfiguration();
+        configuration.setFetchName(true);
+        configuration.setNameMetadataKey("entityName");
+        configuration.setFetchType(true);
+        configuration.setTypeMetadataKey("entityType");
+        return configuration;
+    }
+}

+/**
+ * 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.metadata;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import lombok.extern.slf4j.Slf4j;
+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.api.util.TbNodeUtils;
+import org.thingsboard.rule.engine.util.EntitiesFieldsAsyncLoader;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
+import static org.thingsboard.rule.engine.api.util.DonAsynchron.withCallback;
+
+/**
+ * Created by ashvayka on 19.01.18.
+ */
+@Slf4j
+@RuleNode(type = ComponentType.ENRICHMENT,
+        name = "entity fields",
+        configClazz = TbGetOriginatorFieldsConfiguration.class,
+        nodeDescription = "Add Message Originator Name and Type into Message Metadata",
+        nodeDetails = "If originator is Asset, Device or Alarm, both name and type are added. In all other cases type will always be \"default\"")
+public class TbGetOriginatorFieldsNode implements TbNode {
+
+    private TbGetOriginatorFieldsConfiguration config;
+
+    @Override
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+        config = TbNodeUtils.convert(configuration, TbGetOriginatorFieldsConfiguration.class);
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
+        try {
+            withCallback(putEntityFields(ctx, msg.getOriginator(), msg),
+                    i -> ctx.tellNext(msg, SUCCESS), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
+        } catch (Throwable th) {
+            ctx.tellFailure(msg, th);
+        }
+    }
+
+    private ListenableFuture<Void> putEntityFields(TbContext ctx, EntityId entityId, TbMsg msg) {
+        if (!config.isFetchName() && !config.isFetchType()) {
+            return Futures.immediateFuture(null);
+        } else {
+            return Futures.transform(EntitiesFieldsAsyncLoader.findAsync(ctx, entityId),
+                    data -> {
+                        if (config.isFetchName()) {
+                            msg.getMetaData().putValue(config.getNameMetadataKey(), data.getName());
+                        }
+                        if (config.isFetchType()) {
+                            msg.getMetaData().putValue(config.getTypeMetadataKey(), data.getType());
+                        }
+                        return null;
+                    }
+            );
+        }
+    }
+
+    @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.util;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.BaseData;
+import org.thingsboard.server.common.data.EntityFieldsData;
+import org.thingsboard.server.common.data.alarm.AlarmId;
+import org.thingsboard.server.common.data.id.AssetId;
+import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.id.UserId;
+
+import java.util.function.Function;
+
+public class EntitiesFieldsAsyncLoader {
+
+    public static ListenableFuture<EntityFieldsData> findAsync(TbContext ctx, EntityId original) {
+        switch (original.getEntityType()) {
+            case TENANT:
+                return getAsync(ctx.getTenantService().findTenantByIdAsync((TenantId) original),
+                        t -> new EntityFieldsData(t.getId(), t.getName()));
+            case CUSTOMER:
+                return getAsync(ctx.getCustomerService().findCustomerByIdAsync((CustomerId) original),
+                        t -> new EntityFieldsData(t.getId(), t.getName()));
+            case USER:
+                return getAsync(ctx.getUserService().findUserByIdAsync((UserId) original),
+                        t -> new EntityFieldsData(t.getId(), t.getName()));
+            case ASSET:
+                return getAsync(ctx.getAssetService().findAssetByIdAsync((AssetId) original),
+                        t -> new EntityFieldsData(t.getId(), t.getName(), t.getType()));
+            case DEVICE:
+                return getAsync(ctx.getDeviceService().findDeviceByIdAsync((DeviceId) original),
+                        t -> new EntityFieldsData(t.getId(), t.getName(), t.getType()));
+            case ALARM:
+                return getAsync(ctx.getAlarmService().findAlarmByIdAsync((AlarmId) original),
+                        t -> new EntityFieldsData(t.getId(), t.getName(), t.getType()));
+            case RULE_CHAIN:
+                return getAsync(ctx.getRuleChainService().findRuleChainByIdAsync((RuleChainId) original),
+                        t -> new EntityFieldsData(t.getId(), t.getName()));
+            default:
+                return Futures.immediateFailedFuture(new TbNodeException("Unexpected original EntityType " + original));
+        }
+    }
+
+    private static <T extends BaseData> ListenableFuture<EntityFieldsData> getAsync(
+            ListenableFuture<T> future, Function<T, EntityFieldsData> converter) {
+        return Futures.transformAsync(future, in -> in != null ?
+                Futures.immediateFuture(converter.apply(in))
+                : Futures.immediateFailedFuture(new RuntimeException("Entity not found!")));
+    }
+}