Enrich Rule Node with Tenant/Customer/Self attributes

common/data/src/main/java/org/thingsboard/server/common/data/Customer.java

@@ -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;
     

common/data/src/main/java/org/thingsboard/server/common/data/Device.java

@@ -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;
 

common/data/src/main/java/org/thingsboard/server/common/data/HasCustomerId.java

@@ -0,0 +1,23 @@
+/**
+ * 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();
+}

common/data/src/main/java/org/thingsboard/server/common/data/HasTenantId.java

@@ -0,0 +1,23 @@
+/**
+ * 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();
+}

common/data/src/main/java/org/thingsboard/server/common/data/User.java

@@ -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;
 

common/data/src/main/java/org/thingsboard/server/common/data/alarm/Alarm.java

@@ -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;

common/data/src/main/java/org/thingsboard/server/common/data/asset/Asset.java

@@ -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.*;
-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.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;
 

common/data/src/main/java/org/thingsboard/server/common/data/plugin/PluginMetaData.java

@@ -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;
 

common/data/src/main/java/org/thingsboard/server/common/data/rule/RuleChain.java

@@ -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;
 

common/data/src/main/java/org/thingsboard/server/common/data/rule/RuleMetaData.java

@@ -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;
 

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

@@ -17,7 +17,15 @@ 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.rule.RuleChainService;
+import org.thingsboard.server.dao.rule.RuleService;
+import org.thingsboard.server.dao.user.UserService;
 
 import java.util.UUID;
 
@@ -40,6 +48,24 @@ public interface TbContext {
 
     void ack(UUID 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();
+
 }

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

@@ -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);
     }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/DonAsynchron.java

@@ -0,0 +1,40 @@
+/**
+ * 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) {
+                onSuccess.accept(result);
+            }
+
+            @Override
+            public void onFailure(Throwable t) {
+                onFailure.accept(t);
+            }
+        });
+    }
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbEntityGetAttrNode.java

@@ -0,0 +1,66 @@
+/**
+ * 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.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.msg.TbMsg;
+
+import java.util.List;
+
+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) {
+        withCallback(
+                findEntityAsync(ctx, msg.getOriginator()),
+                entityId -> withCallback(
+                        ctx.getAttributesService().find(entityId, SERVER_SCOPE, config.getAttrMapping().keySet()),
+                        attributes -> putAttributesAndTell(ctx, msg, attributes),
+                        t -> ctx.tellError(msg, t)
+                ),
+                t -> ctx.tellError(msg, t));
+    }
+
+    private void putAttributesAndTell(TbContext ctx, TbMsg msg, List<AttributeKvEntry> 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);
+
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetAttributesNode.java

@@ -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 {
+        ListenableFuture<List<Void>> future = Futures.successfulAsList(
-            //TODO: refactor this to work async and fetch attributes from cache.
+                putAttrAsync(ctx, msg, CLIENT_SCOPE, config.getClientAttributeNames(), "cs."),
-            AttributesService service = ctx.getAttributesService();
+                putAttrAsync(ctx, msg, SHARED_SCOPE, config.getSharedAttributeNames(), "shared."),
-            fetchAttributes(msg, service, config.getClientAttributeNames(), DataConstants.CLIENT_SCOPE, "cs.");
+                putAttrAsync(ctx, msg, SERVER_SCOPE, config.getServerAttributeNames(), "ss."));
-            fetchAttributes(msg, service, config.getServerAttributeNames(), DataConstants.SERVER_SCOPE, "ss.");
+
-            fetchAttributes(msg, service, config.getSharedAttributeNames(), DataConstants.SHARED_SCOPE, "shared.");
+        withCallback(future, i -> ctx.tellNext(msg), t -> ctx.tellError(msg, t));
-            ctx.tellNext(msg);
-        } catch (Exception e) {
-            log.warn("[{}][{}] Failed to fetch attributes", msg.getOriginator(), msg.getId(), e);
-            throw new TbNodeException(e);
-        }
     }
 
-    private void fetchAttributes(TbMsg msg, AttributesService service, List<String> attributeNames, String scope, String prefix) throws InterruptedException, java.util.concurrent.ExecutionException {
+    private ListenableFuture<Void> putAttributesAsync(TbMsg msg, List<AttributeKvEntry> attributes, String prefix) {
-        if (attributeNames != null && attributeNames.isEmpty()) {
+        attributes.forEach(r -> msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString()));
-            List<AttributeKvEntry> attributes = service.find(msg.getOriginator(), scope, attributeNames).get();
+        return Futures.immediateFuture(null);
-            attributes.forEach(attr -> msg.getMetaData().putValue(prefix + attr.getKey(), attr.getValueAsString()));
-        }
     }
 
+    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() {
 

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetCustomerAttributeNode.java

@@ -0,0 +1,49 @@
+/**
+ * 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 -> Futures.immediateFuture(in.getCustomerId()));
+    }
+
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetEntityAttrNodeConfiguration.java

@@ -0,0 +1,27 @@
+/**
+ * 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;
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTenantAttributeNode.java

@@ -0,0 +1,62 @@
+/**
+ * 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 -> Futures.immediateFuture(in.getTenantId()));
+    }
+
+}

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/TbGetCustomerAttributeNodeTest.java

@@ -0,0 +1,44 @@
+/**
+ * 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 org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import static org.junit.Assert.*;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TbGetCustomerAttributeNodeTest {
+
+    //todo-vp: return to this after fixing compilation problem with protobuff
+//    private TbGetCustomerAttributeNode node;
+//
+//    @Mock
+//    private TbContext ctx;
+//
+//    private TbMsg msg;
+//
+//    @Test
+//    public void customerAttributeAddedInMetadata() {
+//        msg = new TbMsg();
+//        node.onMsg(ctx, );
+//    }
+
+}