Merge branch 'develop/1.5' into develop/1.5-refactoring

@@ -23,7 +23,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 
 import com.fasterxml.jackson.databind.JsonNode;
 
-public class Customer extends ContactBased<CustomerId> implements HasName {
+public class Customer extends ContactBased<CustomerId> implements HasName, HasTenantId {
     
     private static final long serialVersionUID = -1599722990298929275L;
     

@@ -23,7 +23,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import com.fasterxml.jackson.databind.JsonNode;
 
 @EqualsAndHashCode(callSuper = true)
-public class Device extends SearchTextBasedWithAdditionalInfo<DeviceId> implements HasName {
+public class Device extends SearchTextBasedWithAdditionalInfo<DeviceId> implements HasName, HasTenantId, HasCustomerId {
 
     private static final long serialVersionUID = 2807343040519543363L;
 

+/**
+ * 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 org.thingsboard.server.common.data.id.CustomerId;
+
+public interface HasCustomerId {
+
+    CustomerId getCustomerId();
+}

+/**
+ * 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 org.thingsboard.server.common.data.id.TenantId;
+
+public interface HasTenantId {
+
+    TenantId getTenantId();
+}

@@ -25,7 +25,7 @@ import org.thingsboard.server.common.data.security.Authority;
 import com.fasterxml.jackson.databind.JsonNode;
 
 @EqualsAndHashCode(callSuper = true)
-public class User extends SearchTextBasedWithAdditionalInfo<UserId> implements HasName {
+public class User extends SearchTextBasedWithAdditionalInfo<UserId> implements HasName, HasTenantId, HasCustomerId {
 
     private static final long serialVersionUID = 8250339805336035966L;
 

@@ -22,6 +22,7 @@ import lombok.Builder;
 import lombok.Data;
 import org.thingsboard.server.common.data.BaseData;
 import org.thingsboard.server.common.data.HasName;
+import org.thingsboard.server.common.data.HasTenantId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 
@@ -31,7 +32,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 @Data
 @Builder
 @AllArgsConstructor
-public class Alarm extends BaseData<AlarmId> implements HasName {
+public class Alarm extends BaseData<AlarmId> implements HasName, HasTenantId {
 
     private TenantId tenantId;
     private String type;

@@ -17,16 +17,13 @@ package org.thingsboard.server.common.data.asset;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import lombok.EqualsAndHashCode;
-import org.thingsboard.server.common.data.HasAdditionalInfo;
-import org.thingsboard.server.common.data.HasName;
-import org.thingsboard.server.common.data.SearchTextBased;
-import org.thingsboard.server.common.data.SearchTextBasedWithAdditionalInfo;
+import org.thingsboard.server.common.data.*;
 import org.thingsboard.server.common.data.id.AssetId;
 import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.TenantId;
 
 @EqualsAndHashCode(callSuper = true)
-public class Asset extends SearchTextBasedWithAdditionalInfo<AssetId> implements HasName {
+public class Asset extends SearchTextBasedWithAdditionalInfo<AssetId> implements HasName, HasTenantId, HasCustomerId {
 
     private static final long serialVersionUID = 2807343040519543363L;
 

@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.EqualsAndHashCode;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.HasName;
+import org.thingsboard.server.common.data.HasTenantId;
 import org.thingsboard.server.common.data.SearchTextBasedWithAdditionalInfo;
 import org.thingsboard.server.common.data.id.PluginId;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -32,7 +33,7 @@ import java.io.IOException;
 
 @EqualsAndHashCode(callSuper = true)
 @Slf4j
-public class PluginMetaData extends SearchTextBasedWithAdditionalInfo<PluginId> implements HasName {
+public class PluginMetaData extends SearchTextBasedWithAdditionalInfo<PluginId> implements HasName, HasTenantId {
 
     private static final long serialVersionUID = 1L;
 

@@ -21,6 +21,7 @@ import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.HasName;
+import org.thingsboard.server.common.data.HasTenantId;
 import org.thingsboard.server.common.data.SearchTextBasedWithAdditionalInfo;
 import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
@@ -29,7 +30,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 @Data
 @EqualsAndHashCode(callSuper = true)
 @Slf4j
-public class RuleChain extends SearchTextBasedWithAdditionalInfo<RuleChainId> implements HasName {
+public class RuleChain extends SearchTextBasedWithAdditionalInfo<RuleChainId> implements HasName, HasTenantId {
 
     private static final long serialVersionUID = -5656679015121935465L;
 

@@ -23,6 +23,7 @@ import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.HasName;
+import org.thingsboard.server.common.data.HasTenantId;
 import org.thingsboard.server.common.data.SearchTextBasedWithAdditionalInfo;
 import org.thingsboard.server.common.data.id.RuleId;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -31,7 +32,7 @@ import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
 @Data
 @EqualsAndHashCode(callSuper = true)
 @Slf4j
-public class RuleMetaData extends SearchTextBasedWithAdditionalInfo<RuleId> implements HasName {
+public class RuleMetaData extends SearchTextBasedWithAdditionalInfo<RuleId> implements HasName, HasTenantId {
 
     private static final long serialVersionUID = -5656679015122935465L;
 

@@ -61,13 +61,14 @@ public class HostRequestIntervalRegistry {
     }
 
     public long tick(String clientHostId) {
+        IntervalCount intervalCount = hostCounts.computeIfAbsent(clientHostId, s -> new IntervalCount(intervalDurationMs));
+        long currentCount = intervalCount.resetIfExpiredAndTick();
         if (whiteList.contains(clientHostId)) {
             return 0;
         } else if (blackList.contains(clientHostId)) {
             return Long.MAX_VALUE;
         }
-        IntervalCount intervalCount = hostCounts.computeIfAbsent(clientHostId, s -> new IntervalCount(intervalDurationMs));
-        return intervalCount.resetIfExpiredAndTick();
+        return currentCount;
     }
 
     public void clean() {

@@ -17,7 +17,17 @@ package org.thingsboard.rule.engine.api;
 
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.cluster.ServerAddress;
+import org.thingsboard.server.dao.alarm.AlarmService;
+import org.thingsboard.server.dao.asset.AssetService;
 import org.thingsboard.server.dao.attributes.AttributesService;
+import org.thingsboard.server.dao.customer.CustomerService;
+import org.thingsboard.server.dao.device.DeviceService;
+import org.thingsboard.server.dao.plugin.PluginService;
+import org.thingsboard.server.dao.relation.RelationService;
+import org.thingsboard.server.dao.rule.RuleChainService;
+import org.thingsboard.server.dao.rule.RuleService;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.dao.user.UserService;
 
 import java.util.UUID;
 
@@ -40,6 +50,28 @@ public interface TbContext {
 
     void ack(TbMsg msg);
 
+    void tellError(TbMsg msg, Throwable th);
+
     AttributesService getAttributesService();
 
+    CustomerService getCustomerService();
+
+    UserService getUserService();
+
+    RuleService getRuleService();
+
+    PluginService getPluginService();
+
+    AssetService getAssetService();
+
+    DeviceService getDeviceService();
+
+    AlarmService getAlarmService();
+
+    RuleChainService getRuleChainService();
+
+    TimeseriesService getTimeseriesService();
+
+    RelationService getRelationService();
+
 }

@@ -22,6 +22,10 @@ import com.fasterxml.jackson.core.JsonProcessingException;
  */
 public class TbNodeException extends Exception {
 
+    public TbNodeException(String message) {
+        super(message);
+    }
+
     public TbNodeException(Exception e) {
         super(e);
     }

+/**
+ * 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;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import javax.annotation.Nullable;
+import java.util.function.Consumer;
+
+public class DonAsynchron {
+
+    public static  <T> void withCallback(ListenableFuture<T> future, Consumer<T> onSuccess, Consumer<Throwable> onFailure) {
+        Futures.addCallback(future, new FutureCallback<T>() {
+            @Override
+            public void onSuccess(@Nullable T result) {
+                try {
+                    onSuccess.accept(result);
+                } catch (Throwable th) {
+                    onFailure(th);
+                }
+
+            }
+
+            @Override
+            public void onFailure(Throwable t) {
+                onFailure.accept(t);
+            }
+        });
+    }
+}

+/**
+ * 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.base.Function;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.thingsboard.rule.engine.TbNodeUtils;
+import org.thingsboard.rule.engine.api.*;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.KvEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.thingsboard.rule.engine.DonAsynchron.withCallback;
+import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE;
+
+public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode {
+
+    private TbGetEntityAttrNodeConfiguration config;
+
+    @Override
+    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+        this.config = TbNodeUtils.convert(configuration, TbGetEntityAttrNodeConfiguration.class);
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) {
+        try {
+            withCallback(
+                    findEntityAsync(ctx, msg.getOriginator()),
+                    entityId -> withCallback(
+                            config.isTelemetry() ? getLatestTelemetry(ctx, entityId) : getAttributesAsync(ctx, entityId),
+                            attributes -> putAttributesAndTell(ctx, msg, attributes),
+                            t -> ctx.tellError(msg, t)
+                    ),
+                    t -> ctx.tellError(msg, t));
+        } catch (Throwable th) {
+            ctx.tellError(msg, th);
+        }
+    }
+
+    private ListenableFuture<List<KvEntry>> getAttributesAsync(TbContext ctx, EntityId entityId) {
+        ListenableFuture<List<AttributeKvEntry>> latest = ctx.getAttributesService().find(entityId, SERVER_SCOPE, config.getAttrMapping().keySet());
+        return Futures.transform(latest, (Function<? super List<AttributeKvEntry>, ? extends List<KvEntry>>) l ->
+                l.stream().map(i -> (KvEntry) i).collect(Collectors.toList()));
+    }
+
+    private ListenableFuture<List<KvEntry>> getLatestTelemetry(TbContext ctx, EntityId entityId) {
+        ListenableFuture<List<TsKvEntry>> latest = ctx.getTimeseriesService().findLatest(entityId, config.getAttrMapping().keySet());
+        return Futures.transform(latest, (Function<? super List<TsKvEntry>, ? extends List<KvEntry>>) l ->
+                l.stream().map(i -> (KvEntry) i).collect(Collectors.toList()));
+    }
+
+
+    private void putAttributesAndTell(TbContext ctx, TbMsg msg, List<KvEntry> attributes) {
+        attributes.forEach(r -> {
+            String attrName = config.getAttrMapping().get(r.getKey());
+            msg.getMetaData().putValue(attrName, r.getValueAsString());
+        });
+        ctx.tellNext(msg);
+    }
+
+    @Override
+    public void destroy() {
+
+    }
+
+    protected abstract ListenableFuture<T> findEntityAsync(TbContext ctx, EntityId originator);
+
+}

@@ -15,23 +15,27 @@
  */
 package org.thingsboard.rule.engine.metadata;
 
+import com.google.common.util.concurrent.AsyncFunction;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.rule.engine.TbNodeUtils;
 import org.thingsboard.rule.engine.api.*;
-import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.attributes.AttributesService;
 
 import java.util.List;
 
+import static org.thingsboard.rule.engine.DonAsynchron.withCallback;
+import static org.thingsboard.server.common.data.DataConstants.*;
+
 /**
  * Created by ashvayka on 19.01.18.
  */
 @Slf4j
 public class TbGetAttributesNode implements TbNode {
 
-    TbGetAttributesNodeConfiguration config;
+    private TbGetAttributesNodeConfiguration config;
 
     @Override
     public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
@@ -40,26 +44,25 @@ public class TbGetAttributesNode implements TbNode {
 
     @Override
     public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
-        try {
-            //TODO: refactor this to work async and fetch attributes from cache.
-            AttributesService service = ctx.getAttributesService();
-            fetchAttributes(msg, service, config.getClientAttributeNames(), DataConstants.CLIENT_SCOPE, "cs.");
-            fetchAttributes(msg, service, config.getServerAttributeNames(), DataConstants.SERVER_SCOPE, "ss.");
-            fetchAttributes(msg, service, config.getSharedAttributeNames(), DataConstants.SHARED_SCOPE, "shared.");
-            ctx.tellNext(msg);
-        } catch (Exception e) {
-            log.warn("[{}][{}] Failed to fetch attributes", msg.getOriginator(), msg.getId(), e);
-            throw new TbNodeException(e);
-        }
+        ListenableFuture<List<Void>> future = Futures.allAsList(
+                putAttrAsync(ctx, msg, CLIENT_SCOPE, config.getClientAttributeNames(), "cs."),
+                putAttrAsync(ctx, msg, SHARED_SCOPE, config.getSharedAttributeNames(), "shared."),
+                putAttrAsync(ctx, msg, SERVER_SCOPE, config.getServerAttributeNames(), "ss."));
+
+        withCallback(future, i -> ctx.tellNext(msg), t -> ctx.tellError(msg, t));
     }
 
-    private void fetchAttributes(TbMsg msg, AttributesService service, List<String> attributeNames, String scope, String prefix) throws InterruptedException, java.util.concurrent.ExecutionException {
-        if (attributeNames != null && attributeNames.isEmpty()) {
-            List<AttributeKvEntry> attributes = service.find(msg.getOriginator(), scope, attributeNames).get();
-            attributes.forEach(attr -> msg.getMetaData().putValue(prefix + attr.getKey(), attr.getValueAsString()));
-        }
+    private ListenableFuture<Void> putAttributesAsync(TbMsg msg, List<AttributeKvEntry> attributes, String prefix) {
+        attributes.forEach(r -> msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString()));
+        return Futures.immediateFuture(null);
     }
 
+    private ListenableFuture<Void> putAttrAsync(TbContext ctx, TbMsg msg, String scope, List<String> attributes, String prefix) {
+        return Futures.transform(ctx.getAttributesService().find(msg.getOriginator(), scope, attributes),
+                (AsyncFunction<List<AttributeKvEntry>, Void>) i -> putAttributesAsync(msg, i, prefix));
+    }
+
+
     @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.metadata;
+
+import com.google.common.util.concurrent.AsyncFunction;
+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.HasCustomerId;
+import org.thingsboard.server.common.data.id.*;
+
+public class TbGetCustomerAttributeNode extends TbEntityGetAttrNode<CustomerId> {
+
+    @Override
+    protected ListenableFuture<CustomerId> findEntityAsync(TbContext ctx, EntityId originator) {
+
+        switch (originator.getEntityType()) {
+            case CUSTOMER:
+                return Futures.immediateFuture((CustomerId) originator);
+            case USER:
+                return getCustomerAsync(ctx.getUserService().findUserByIdAsync((UserId) originator));
+            case ASSET:
+                return getCustomerAsync(ctx.getAssetService().findAssetByIdAsync((AssetId) originator));
+            case DEVICE:
+                return getCustomerAsync(ctx.getDeviceService().findDeviceByIdAsync((DeviceId) originator));
+            default:
+                return Futures.immediateFailedFuture(new TbNodeException("Unexpected originator EntityType " + originator));
+        }
+    }
+
+    private <T extends HasCustomerId> ListenableFuture<CustomerId> getCustomerAsync(ListenableFuture<T> future) {
+        return Futures.transform(future, (AsyncFunction<HasCustomerId, CustomerId>) in -> {
+            return in != null ? Futures.immediateFuture(in.getCustomerId())
+                    : Futures.immediateFailedFuture(new IllegalStateException("Customer not found"));});
+    }
+
+}

+/**
+ * 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 java.util.Map;
+import java.util.Optional;
+
+@Data
+public class TbGetEntityAttrNodeConfiguration {
+
+    private Map<String, String> attrMapping;
+    private boolean isTelemetry = false;
+}

+/**
+ * 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.server.common.data.relation.EntitySearchDirection;
+
+@Data
+public class TbGetRelatedAttrNodeConfiguration {
+
+    private String relationType;
+    private EntitySearchDirection direction;
+}

+/**
+ * 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.AsyncFunction;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.commons.collections.CollectionUtils;
+import org.thingsboard.rule.engine.TbNodeUtils;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.rule.engine.api.TbNodeState;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.relation.EntityRelation;
+import org.thingsboard.server.common.data.relation.EntitySearchDirection;
+import org.thingsboard.server.dao.relation.RelationService;
+
+import java.util.List;
+
+import static org.thingsboard.server.common.data.relation.RelationTypeGroup.COMMON;
+
+public class TbGetRelatedAttributeNode extends TbEntityGetAttrNode<EntityId> {
+
+    private TbGetRelatedAttrNodeConfiguration config;
+
+    @Override
+    public void init(TbNodeConfiguration configuration, TbNodeState state) throws TbNodeException {
+        this.config = TbNodeUtils.convert(configuration, TbGetRelatedAttrNodeConfiguration.class);
+    }
+
+    @Override
+    protected ListenableFuture<EntityId> findEntityAsync(TbContext ctx, EntityId originator) {
+        RelationService relationService = ctx.getRelationService();
+        if (config.getDirection() == EntitySearchDirection.FROM) {
+            ListenableFuture<List<EntityRelation>> asyncRelation = relationService.findByFromAndTypeAsync(originator, config.getRelationType(), COMMON);
+            return Futures.transform(asyncRelation, (AsyncFunction<? super List<EntityRelation>, EntityId>)
+                    r -> CollectionUtils.isNotEmpty(r) ? Futures.immediateFuture(r.get(0).getTo())
+                            : Futures.immediateFailedFuture(new IllegalStateException("Relation not found")));
+        } else if (config.getDirection() == EntitySearchDirection.TO) {
+            ListenableFuture<List<EntityRelation>> asyncRelation = relationService.findByToAndTypeAsync(originator, config.getRelationType(), COMMON);
+            return Futures.transform(asyncRelation, (AsyncFunction<? super List<EntityRelation>, EntityId>)
+                    r -> CollectionUtils.isNotEmpty(r) ? Futures.immediateFuture(r.get(0).getFrom())
+                            : Futures.immediateFailedFuture(new IllegalStateException("Relation not found")));
+        }
+
+        return Futures.immediateFailedFuture(new IllegalStateException("Unknown direction"));
+    }
+}

+/**
+ * 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.AsyncFunction;
+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.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.HasTenantId;
+import org.thingsboard.server.common.data.alarm.AlarmId;
+import org.thingsboard.server.common.data.id.*;
+
+@Slf4j
+public class TbGetTenantAttributeNode extends TbEntityGetAttrNode<TenantId> {
+
+    @Override
+    protected ListenableFuture<TenantId> findEntityAsync(TbContext ctx, EntityId originator) {
+
+        switch (originator.getEntityType()) {
+            case TENANT:
+                return Futures.immediateFuture((TenantId) originator);
+            case CUSTOMER:
+                return getTenantAsync(ctx.getCustomerService().findCustomerByIdAsync((CustomerId) originator));
+            case USER:
+                return getTenantAsync(ctx.getUserService().findUserByIdAsync((UserId) originator));
+            case RULE:
+                return getTenantAsync(ctx.getRuleService().findRuleByIdAsync((RuleId) originator));
+            case PLUGIN:
+                return getTenantAsync(ctx.getPluginService().findPluginByIdAsync((PluginId) originator));
+            case ASSET:
+                return getTenantAsync(ctx.getAssetService().findAssetByIdAsync((AssetId) originator));
+            case DEVICE:
+                return getTenantAsync(ctx.getDeviceService().findDeviceByIdAsync((DeviceId) originator));
+            case ALARM:
+                return getTenantAsync(ctx.getAlarmService().findAlarmByIdAsync((AlarmId) originator));
+            case RULE_CHAIN:
+                return getTenantAsync(ctx.getRuleChainService().findRuleChainByIdAsync((RuleChainId) originator));
+            default:
+                return Futures.immediateFailedFuture(new TbNodeException("Unexpected originator EntityType " + originator));
+        }
+    }
+
+    private <T extends HasTenantId> ListenableFuture<TenantId> getTenantAsync(ListenableFuture<T> future) {
+        return Futures.transform(future, (AsyncFunction<HasTenantId, TenantId>) in -> {
+            return in != null ? Futures.immediateFuture(in.getTenantId())
+                : Futures.immediateFailedFuture(new IllegalStateException("Tenant not found"));});
+    }
+
+}

+/**
+ * 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.datastax.driver.core.utils.UUIDs;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.asset.Asset;
+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.UserId;
+import org.thingsboard.server.common.data.kv.*;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.dao.asset.AssetService;
+import org.thingsboard.server.dao.attributes.AttributesService;
+import org.thingsboard.server.dao.device.DeviceService;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.dao.user.UserService;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TbGetCustomerAttributeNodeTest {
+
+    private TbGetCustomerAttributeNode node;
+
+    @Mock
+    private TbContext ctx;
+
+    @Mock
+    private AttributesService attributesService;
+    @Mock
+    private TimeseriesService timeseriesService;
+    @Mock
+    private UserService userService;
+    @Mock
+    private AssetService assetService;
+    @Mock
+    private DeviceService deviceService;
+
+    private TbMsg msg;
+
+    @Before
+    public void init() throws TbNodeException {
+        TbGetEntityAttrNodeConfiguration config = new TbGetEntityAttrNodeConfiguration();
+        Map<String, String> attrMapping = new HashMap<>();
+        attrMapping.putIfAbsent("temperature", "tempo");
+        config.setAttrMapping(attrMapping);
+        config.setTelemetry(false);
+        ObjectMapper mapper = new ObjectMapper();
+        TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration();
+        nodeConfiguration.setData(mapper.valueToTree(config));
+
+        node = new TbGetCustomerAttributeNode();
+        node.init(nodeConfiguration, null);
+    }
+
+    @Test
+    public void errorThrownIfCannotLoadAttributes() {
+        UserId userId = new UserId(UUIDs.timeBased());
+        CustomerId customerId = new CustomerId(UUIDs.timeBased());
+        User user = new User();
+        user.setCustomerId(customerId);
+
+        msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), new byte[4]);
+
+        when(ctx.getUserService()).thenReturn(userService);
+        when(userService.findUserByIdAsync(userId)).thenReturn(Futures.immediateFuture(user));
+
+        when(ctx.getAttributesService()).thenReturn(attributesService);
+        when(attributesService.find(customerId, SERVER_SCOPE, Collections.singleton("temperature")))
+                .thenThrow(new IllegalStateException("something wrong"));
+
+        node.onMsg(ctx, msg);
+        final ArgumentCaptor<Throwable> captor = ArgumentCaptor.forClass(Throwable.class);
+        verify(ctx).tellError(same(msg), captor.capture());
+
+        Throwable value = captor.getValue();
+        assertEquals("something wrong", value.getMessage());
+        assertTrue(msg.getMetaData().getData().isEmpty());
+    }
+
+    @Test
+    public void errorThrownIfCannotLoadAttributesAsync() {
+        UserId userId = new UserId(UUIDs.timeBased());
+        CustomerId customerId = new CustomerId(UUIDs.timeBased());
+        User user = new User();
+        user.setCustomerId(customerId);
+
+        msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), new byte[4]);
+
+        when(ctx.getUserService()).thenReturn(userService);
+        when(userService.findUserByIdAsync(userId)).thenReturn(Futures.immediateFuture(user));
+
+        when(ctx.getAttributesService()).thenReturn(attributesService);
+        when(attributesService.find(customerId, SERVER_SCOPE, Collections.singleton("temperature")))
+                .thenReturn(Futures.immediateFailedFuture(new IllegalStateException("something wrong")));
+
+        node.onMsg(ctx, msg);
+        final ArgumentCaptor<Throwable> captor = ArgumentCaptor.forClass(Throwable.class);
+        verify(ctx).tellError(same(msg), captor.capture());
+
+        Throwable value = captor.getValue();
+        assertEquals("something wrong", value.getMessage());
+        assertTrue(msg.getMetaData().getData().isEmpty());
+    }
+
+    @Test
+    public void errorThrownIfCustomerCannotBeFound() {
+        UserId userId = new UserId(UUIDs.timeBased());
+        CustomerId customerId = new CustomerId(UUIDs.timeBased());
+        User user = new User();
+        user.setCustomerId(customerId);
+
+        msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), new byte[4]);
+
+        when(ctx.getUserService()).thenReturn(userService);
+        when(userService.findUserByIdAsync(userId)).thenReturn(Futures.immediateFuture(null));
+
+        node.onMsg(ctx, msg);
+        final ArgumentCaptor<Throwable> captor = ArgumentCaptor.forClass(Throwable.class);
+        verify(ctx).tellError(same(msg), captor.capture());
+
+        Throwable value = captor.getValue();
+        assertEquals(IllegalStateException.class, value.getClass());
+        assertEquals("Customer not found", value.getMessage());
+        assertTrue(msg.getMetaData().getData().isEmpty());
+    }
+
+    @Test
+    public void customerAttributeAddedInMetadata() {
+        CustomerId customerId = new CustomerId(UUIDs.timeBased());
+        msg = new TbMsg(UUIDs.timeBased(), "CUSTOMER", customerId, new TbMsgMetaData(), new byte[4]);
+        entityAttributeFetched(customerId);
+    }
+
+    @Test
+    public void usersCustomerAttributesFetched() {
+        UserId userId = new UserId(UUIDs.timeBased());
+        CustomerId customerId = new CustomerId(UUIDs.timeBased());
+        User user = new User();
+        user.setCustomerId(customerId);
+
+        msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), new byte[4]);
+
+        when(ctx.getUserService()).thenReturn(userService);
+        when(userService.findUserByIdAsync(userId)).thenReturn(Futures.immediateFuture(user));
+
+        entityAttributeFetched(customerId);
+    }
+
+    @Test
+    public void assetsCustomerAttributesFetched() {
+        AssetId assetId = new AssetId(UUIDs.timeBased());
+        CustomerId customerId = new CustomerId(UUIDs.timeBased());
+        Asset asset = new Asset();
+        asset.setCustomerId(customerId);
+
+        msg = new TbMsg(UUIDs.timeBased(), "USER", assetId, new TbMsgMetaData(), new byte[4]);
+
+        when(ctx.getAssetService()).thenReturn(assetService);
+        when(assetService.findAssetByIdAsync(assetId)).thenReturn(Futures.immediateFuture(asset));
+
+        entityAttributeFetched(customerId);
+    }
+
+    @Test
+    public void deviceCustomerAttributesFetched() {
+        DeviceId deviceId = new DeviceId(UUIDs.timeBased());
+        CustomerId customerId = new CustomerId(UUIDs.timeBased());
+        Device device = new Device();
+        device.setCustomerId(customerId);
+
+        msg = new TbMsg(UUIDs.timeBased(), "USER", deviceId, new TbMsgMetaData(), new byte[4]);
+
+        when(ctx.getDeviceService()).thenReturn(deviceService);
+        when(deviceService.findDeviceByIdAsync(deviceId)).thenReturn(Futures.immediateFuture(device));
+
+        entityAttributeFetched(customerId);
+    }
+
+    @Test
+    public void deviceCustomerTelemetryFetched() throws TbNodeException {
+        TbGetEntityAttrNodeConfiguration config = new TbGetEntityAttrNodeConfiguration();
+        Map<String, String> attrMapping = new HashMap<>();
+        attrMapping.putIfAbsent("temperature", "tempo");
+        config.setAttrMapping(attrMapping);
+        config.setTelemetry(true);
+        ObjectMapper mapper = new ObjectMapper();
+        TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration();
+        nodeConfiguration.setData(mapper.valueToTree(config));
+
+        node = new TbGetCustomerAttributeNode();
+        node.init(nodeConfiguration, null);
+
+
+        DeviceId deviceId = new DeviceId(UUIDs.timeBased());
+        CustomerId customerId = new CustomerId(UUIDs.timeBased());
+        Device device = new Device();
+        device.setCustomerId(customerId);
+
+        msg = new TbMsg(UUIDs.timeBased(), "USER", deviceId, new TbMsgMetaData(), new byte[4]);
+
+        when(ctx.getDeviceService()).thenReturn(deviceService);
+        when(deviceService.findDeviceByIdAsync(deviceId)).thenReturn(Futures.immediateFuture(device));
+
+        List<TsKvEntry> timeseries = Lists.newArrayList(new BasicTsKvEntry(1L, new StringDataEntry("temperature", "highest")));
+
+        when(ctx.getTimeseriesService()).thenReturn(timeseriesService);
+        when(timeseriesService.findLatest(customerId, Collections.singleton("temperature")))
+                .thenReturn(Futures.immediateFuture(timeseries));
+
+        node.onMsg(ctx, msg);
+        verify(ctx).tellNext(msg);
+        assertEquals(msg.getMetaData().getValue("tempo"), "highest");
+    }
+
+    private void entityAttributeFetched(CustomerId customerId) {
+        List<AttributeKvEntry> attributes = Lists.newArrayList(new BaseAttributeKvEntry(new StringDataEntry("temperature", "high"), 1L));
+
+        when(ctx.getAttributesService()).thenReturn(attributesService);
+        when(attributesService.find(customerId, SERVER_SCOPE, Collections.singleton("temperature")))
+                .thenReturn(Futures.immediateFuture(attributes));
+
+        node.onMsg(ctx, msg);
+        verify(ctx).tellNext(msg);
+        assertEquals(msg.getMetaData().getValue("tempo"), "high");
+    }
+}
\ No newline at end of file