diff --git a/application/pom.xml b/application/pom.xml
index 3693fe6dbe..96c9ab7b39 100644
--- a/application/pom.xml
+++ b/application/pom.xml
@@ -286,7 +286,6 @@
com.jayway.jsonpath
json-path
- test
com.jayway.jsonpath
diff --git a/application/src/main/data/upgrade/3.4.1/schema_update.sql b/application/src/main/data/upgrade/3.4.1/schema_update.sql
new file mode 100644
index 0000000000..79cde8167f
--- /dev/null
+++ b/application/src/main/data/upgrade/3.4.1/schema_update.sql
@@ -0,0 +1,74 @@
+--
+-- Copyright © 2016-2022 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.
+--
+
+DO
+$$
+ DECLARE table_partition RECORD;
+ BEGIN
+ -- in case of running the upgrade script a second time:
+ IF NOT (SELECT exists(SELECT FROM pg_tables WHERE tablename = 'old_audit_log')) THEN
+ ALTER TABLE audit_log RENAME TO old_audit_log;
+ ALTER INDEX IF EXISTS idx_audit_log_tenant_id_and_created_time RENAME TO idx_old_audit_log_tenant_id_and_created_time;
+
+ FOR table_partition IN SELECT tablename AS name, split_part(tablename, '_', 3) AS partition_ts
+ FROM pg_tables WHERE tablename LIKE 'audit_log_%'
+ LOOP
+ EXECUTE format('ALTER TABLE %s RENAME TO old_audit_log_%s', table_partition.name, table_partition.partition_ts);
+ END LOOP;
+ ELSE
+ RAISE NOTICE 'Table old_audit_log already exists, leaving as is';
+ END IF;
+ END;
+$$;
+
+CREATE TABLE IF NOT EXISTS audit_log (
+ id uuid NOT NULL,
+ created_time bigint NOT NULL,
+ tenant_id uuid,
+ customer_id uuid,
+ entity_id uuid,
+ entity_type varchar(255),
+ entity_name varchar(255),
+ user_id uuid,
+ user_name varchar(255),
+ action_type varchar(255),
+ action_data varchar(1000000),
+ action_status varchar(255),
+ action_failure_details varchar(1000000)
+) PARTITION BY RANGE (created_time);
+CREATE INDEX IF NOT EXISTS idx_audit_log_tenant_id_and_created_time ON audit_log(tenant_id, created_time DESC);
+
+CREATE OR REPLACE PROCEDURE migrate_audit_logs(IN start_time_ms BIGINT, IN end_time_ms BIGINT, IN partition_size_ms BIGINT)
+ LANGUAGE plpgsql AS
+$$
+DECLARE
+ p RECORD;
+ partition_end_ts BIGINT;
+BEGIN
+ FOR p IN SELECT DISTINCT (created_time - created_time % partition_size_ms) AS partition_ts FROM old_audit_log
+ WHERE created_time >= start_time_ms AND created_time < end_time_ms
+ LOOP
+ partition_end_ts = p.partition_ts + partition_size_ms;
+ RAISE NOTICE '[audit_log] Partition to create : [%-%]', p.partition_ts, partition_end_ts;
+ EXECUTE format('CREATE TABLE IF NOT EXISTS audit_log_%s PARTITION OF audit_log ' ||
+ 'FOR VALUES FROM ( %s ) TO ( %s )', p.partition_ts, p.partition_ts, partition_end_ts);
+ END LOOP;
+
+ INSERT INTO audit_log
+ SELECT * FROM old_audit_log
+ WHERE created_time >= start_time_ms AND created_time < end_time_ms;
+END;
+$$;
diff --git a/application/src/main/data/upgrade/3.4.1/schema_update_after.sql b/application/src/main/data/upgrade/3.4.1/schema_update_after.sql
new file mode 100644
index 0000000000..78df4d2fd2
--- /dev/null
+++ b/application/src/main/data/upgrade/3.4.1/schema_update_after.sql
@@ -0,0 +1,21 @@
+--
+-- Copyright © 2016-2022 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.
+--
+
+DROP PROCEDURE IF EXISTS update_asset_profiles;
+
+ALTER TABLE asset ALTER COLUMN asset_profile_id SET NOT NULL;
+ALTER TABLE asset DROP CONSTRAINT IF EXISTS fk_asset_profile;
+ALTER TABLE asset ADD CONSTRAINT fk_asset_profile FOREIGN KEY (asset_profile_id) REFERENCES asset_profile(id);
diff --git a/application/src/main/data/upgrade/3.4.1/schema_update_before.sql b/application/src/main/data/upgrade/3.4.1/schema_update_before.sql
new file mode 100644
index 0000000000..59566e42b5
--- /dev/null
+++ b/application/src/main/data/upgrade/3.4.1/schema_update_before.sql
@@ -0,0 +1,45 @@
+--
+-- Copyright © 2016-2022 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.
+--
+
+CREATE TABLE IF NOT EXISTS asset_profile (
+ id uuid NOT NULL CONSTRAINT asset_profile_pkey PRIMARY KEY,
+ created_time bigint NOT NULL,
+ name varchar(255),
+ image varchar(1000000),
+ description varchar,
+ search_text varchar(255),
+ is_default boolean,
+ tenant_id uuid,
+ default_rule_chain_id uuid,
+ default_dashboard_id uuid,
+ default_queue_name varchar(255),
+ external_id uuid,
+ CONSTRAINT asset_profile_name_unq_key UNIQUE (tenant_id, name),
+ CONSTRAINT asset_profile_external_id_unq_key UNIQUE (tenant_id, external_id),
+ CONSTRAINT fk_default_rule_chain_asset_profile FOREIGN KEY (default_rule_chain_id) REFERENCES rule_chain(id),
+ CONSTRAINT fk_default_dashboard_asset_profile FOREIGN KEY (default_dashboard_id) REFERENCES dashboard(id)
+ );
+
+CREATE OR REPLACE PROCEDURE update_asset_profiles()
+ LANGUAGE plpgsql AS
+$$
+BEGIN
+UPDATE asset as a SET asset_profile_id = p.id
+ FROM
+ (SELECT id, tenant_id, name from asset_profile) as p
+WHERE a.asset_profile_id IS NULL AND p.tenant_id = a.tenant_id AND a.type = p.name;
+END;
+$$;
diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
index b4c308db86..f4e7eb2463 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
@@ -56,6 +56,7 @@ import org.thingsboard.server.dao.cassandra.CassandraCluster;
import org.thingsboard.server.dao.customer.CustomerService;
import org.thingsboard.server.dao.dashboard.DashboardService;
import org.thingsboard.server.dao.device.ClaimDevicesService;
+import org.thingsboard.server.dao.device.DeviceCredentialsService;
import org.thingsboard.server.dao.device.DeviceService;
import org.thingsboard.server.dao.edge.EdgeEventService;
import org.thingsboard.server.dao.edge.EdgeService;
@@ -86,6 +87,7 @@ import org.thingsboard.server.service.executors.DbCallbackExecutorService;
import org.thingsboard.server.service.executors.ExternalCallExecutorService;
import org.thingsboard.server.service.executors.SharedEventLoopGroupService;
import org.thingsboard.server.service.mail.MailExecutorService;
+import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.service.rpc.TbCoreDeviceRpcService;
import org.thingsboard.server.service.rpc.TbRpcService;
@@ -174,6 +176,10 @@ public class ActorSystemContext {
@Getter
private DeviceService deviceService;
+ @Autowired
+ @Getter
+ private DeviceCredentialsService deviceCredentialsService;
+
@Autowired
@Getter
private TbTenantProfileCache tenantProfileCache;
@@ -182,6 +188,10 @@ public class ActorSystemContext {
@Getter
private TbDeviceProfileCache deviceProfileCache;
+ @Autowired
+ @Getter
+ private TbAssetProfileCache assetProfileCache;
+
@Autowired
@Getter
private AssetService assetService;
diff --git a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
index 2817b1e359..eeb6d8e82e 100644
--- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
@@ -36,7 +36,7 @@ import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.aware.TenantAwareMsg;
-import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
@@ -105,7 +105,9 @@ public class AppActor extends ContextAwareActor {
onToDeviceActorMsg((TenantAwareMsg) msg, true);
break;
case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
- onToTenantActorMsg((EdgeEventUpdateMsg) msg);
+ case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG:
+ case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG:
+ onToEdgeSessionMsg((EdgeSessionMsg) msg);
break;
case SESSION_TIMEOUT_MSG:
ctx.broadcastToChildrenByType(msg, EntityType.TENANT);
@@ -193,7 +195,7 @@ public class AppActor extends ContextAwareActor {
() -> new TenantActor.ActorCreator(systemContext, tenantId));
}
- private void onToTenantActorMsg(EdgeEventUpdateMsg msg) {
+ private void onToEdgeSessionMsg(EdgeSessionMsg msg) {
TbActorRef target = null;
if (ModelConstants.SYSTEM_TENANT.equals(msg.getTenantId())) {
log.warn("Message has system tenant id: {}", msg);
@@ -203,7 +205,7 @@ public class AppActor extends ContextAwareActor {
if (target != null) {
target.tellWithHighPriority(msg);
} else {
- log.debug("[{}] Invalid edge event update msg: {}", msg.getTenantId(), msg);
+ log.debug("[{}] Invalid edge session msg: {}", msg.getTenantId(), msg);
}
}
diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
index 92fdfc6e48..62d4829085 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
@@ -17,11 +17,15 @@ package org.thingsboard.server.actors.ruleChain;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.channel.EventLoopGroup;
import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ListeningExecutor;
import org.thingsboard.rule.engine.api.MailService;
import org.thingsboard.rule.engine.api.RuleEngineAlarmService;
+import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache;
import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache;
import org.thingsboard.rule.engine.api.RuleEngineRpcService;
import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
@@ -38,10 +42,13 @@ import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.HasRuleEngineProfile;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.alarm.Alarm;
import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+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.EdgeId;
@@ -49,6 +56,7 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.rule.RuleNode;
@@ -64,6 +72,7 @@ import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.cassandra.CassandraCluster;
import org.thingsboard.server.dao.customer.CustomerService;
import org.thingsboard.server.dao.dashboard.DashboardService;
+import org.thingsboard.server.dao.device.DeviceCredentialsService;
import org.thingsboard.server.dao.device.DeviceService;
import org.thingsboard.server.dao.edge.EdgeEventService;
import org.thingsboard.server.dao.edge.EdgeService;
@@ -84,6 +93,7 @@ import org.thingsboard.server.queue.TbQueueMsgMetadata;
import org.thingsboard.server.service.script.RuleNodeJsScriptEngine;
import java.util.Collections;
+import java.util.List;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
@@ -326,38 +336,62 @@ class DefaultTbContext implements TbContext {
}
public TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId) {
- RuleChainId ruleChainId = null;
- String queueName = null;
+ DeviceProfile deviceProfile = null;
if (device.getDeviceProfileId() != null) {
- DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId());
- if (deviceProfile == null) {
- log.warn("[{}] Device profile is null!", device.getDeviceProfileId());
- } else {
- ruleChainId = deviceProfile.getDefaultRuleChainId();
- queueName = deviceProfile.getDefaultQueueName();
- }
+ deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId());
}
- return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueName, ruleChainId);
+ return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, deviceProfile);
}
public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) {
- return entityActionMsg(asset, asset.getId(), ruleNodeId, DataConstants.ENTITY_CREATED);
+ AssetProfile assetProfile = null;
+ if (asset.getAssetProfileId() != null) {
+ assetProfile = mainCtx.getAssetProfileCache().find(asset.getAssetProfileId());
+ }
+ return entityActionMsg(asset, asset.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, assetProfile);
}
public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action) {
- RuleChainId ruleChainId = null;
- String queueName = null;
+ HasRuleEngineProfile profile = null;
if (EntityType.DEVICE.equals(alarm.getOriginator().getEntityType())) {
DeviceId deviceId = new DeviceId(alarm.getOriginator().getId());
- DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
- if (deviceProfile == null) {
- log.warn("[{}] Device profile is null!", deviceId);
- } else {
- ruleChainId = deviceProfile.getDefaultRuleChainId();
- queueName = deviceProfile.getDefaultQueueName();
- }
+ profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
+ } else if (EntityType.ASSET.equals(alarm.getOriginator().getEntityType())) {
+ AssetId assetId = new AssetId(alarm.getOriginator().getId());
+ profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId);
+ }
+ return entityActionMsg(alarm, alarm.getOriginator(), ruleNodeId, action, profile);
+ }
+
+ public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) {
+ ObjectNode entityNode = JacksonUtil.newObjectNode();
+ if (attributes != null) {
+ attributes.forEach(attributeKvEntry -> JacksonUtil.addKvEntry(entityNode, attributeKvEntry));
+ }
+ return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode));
+ }
+
+ public TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys) {
+ ObjectNode entityNode = JacksonUtil.newObjectNode();
+ ArrayNode attrsArrayNode = entityNode.putArray("attributes");
+ if (keys != null) {
+ keys.forEach(attrsArrayNode::add);
+ }
+ return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_DELETED, JacksonUtil.toString(entityNode));
+ }
+
+ private TbMsg attributesActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, String action, String msgData) {
+ TbMsgMetaData tbMsgMetaData = getActionMetaData(ruleNodeId);
+ tbMsgMetaData.putValue("scope", scope);
+ HasRuleEngineProfile profile = null;
+ if (EntityType.DEVICE.equals(originator.getEntityType())) {
+ DeviceId deviceId = new DeviceId(originator.getId());
+ profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
+ } else if (EntityType.ASSET.equals(originator.getEntityType())) {
+ AssetId assetId = new AssetId(originator.getId());
+ profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId);
}
- return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, queueName, ruleChainId);
+ return entityActionMsg(originator, tbMsgMetaData, msgData, action, profile);
}
@Override
@@ -366,17 +400,27 @@ class DefaultTbContext implements TbContext {
}
public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action) {
- return entityActionMsg(entity, id, ruleNodeId, action, null, null);
+ return entityActionMsg(entity, id, ruleNodeId, action, null);
}
- public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, String queueName, RuleChainId ruleChainId) {
+ public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, K profile) {
try {
- return TbMsg.newMsg(queueName, action, id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), ruleChainId, null);
+ return entityActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, profile);
} catch (JsonProcessingException | IllegalArgumentException e) {
throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e);
}
}
+ private TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, K profile) {
+ String defaultQueueName = null;
+ RuleChainId defaultRuleChainId = null;
+ if (profile != null) {
+ defaultQueueName = profile.getDefaultQueueName();
+ defaultRuleChainId = profile.getDefaultRuleChainId();
+ }
+ return TbMsg.newMsg(defaultQueueName, action, id, msgMetaData, msgData, defaultRuleChainId, null);
+ }
+
@Override
public RuleNodeId getSelfId() {
return nodeCtx.getSelf().getId();
@@ -478,6 +522,11 @@ class DefaultTbContext implements TbContext {
return mainCtx.getDeviceService();
}
+ @Override
+ public DeviceCredentialsService getDeviceCredentialsService() {
+ return mainCtx.getDeviceCredentialsService();
+ }
+
@Override
public TbClusterService getClusterService() {
return mainCtx.getClusterService();
@@ -533,6 +582,11 @@ class DefaultTbContext implements TbContext {
return mainCtx.getDeviceProfileCache();
}
+ @Override
+ public RuleEngineAssetProfileCache getAssetProfileCache() {
+ return mainCtx.getAssetProfileCache();
+ }
+
@Override
public EdgeService getEdgeService() {
return mainCtx.getEdgeService();
@@ -647,9 +701,15 @@ class DefaultTbContext implements TbContext {
mainCtx.getDeviceProfileCache().addListener(getTenantId(), getSelfId(), profileListener, deviceListener);
}
+ @Override
+ public void addAssetProfileListeners(Consumer profileListener, BiConsumer assetListener) {
+ mainCtx.getAssetProfileCache().addListener(getTenantId(), getSelfId(), profileListener, assetListener);
+ }
+
@Override
public void removeListeners() {
mainCtx.getDeviceProfileCache().removeListener(getTenantId(), getSelfId());
+ mainCtx.getAssetProfileCache().removeListener(getTenantId(), getSelfId());
mainCtx.getTenantProfileCache().removeListener(getTenantId(), getSelfId());
}
diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
index c22b44d7cf..5718f581ad 100644
--- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
@@ -47,7 +47,7 @@ import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.aware.DeviceAwareMsg;
import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
-import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
@@ -167,7 +167,9 @@ public class TenantActor extends RuleChainManagerActor {
onRuleChainMsg((RuleChainAwareMsg) msg);
break;
case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
- onToEdgeSessionMsg((EdgeEventUpdateMsg) msg);
+ case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG:
+ case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG:
+ onToEdgeSessionMsg((EdgeSessionMsg) msg);
break;
default:
return false;
@@ -271,9 +273,8 @@ public class TenantActor extends RuleChainManagerActor {
() -> new DeviceActorCreator(systemContext, tenantId, deviceId));
}
- private void onToEdgeSessionMsg(EdgeEventUpdateMsg msg) {
- log.trace("[{}] onToEdgeSessionMsg [{}]", msg.getTenantId(), msg);
- systemContext.getEdgeRpcService().onEdgeEvent(tenantId, msg.getEdgeId());
+ private void onToEdgeSessionMsg(EdgeSessionMsg msg) {
+ systemContext.getEdgeRpcService().onToEdgeSessionMsg(tenantId, msg);
}
private ApiUsageState getApiUsageState() {
diff --git a/application/src/main/java/org/thingsboard/server/controller/AssetController.java b/application/src/main/java/org/thingsboard/server/controller/AssetController.java
index 75e77d4719..8450c67ea3 100644
--- a/application/src/main/java/org/thingsboard/server/controller/AssetController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/AssetController.java
@@ -41,18 +41,19 @@ import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.AssetId;
+import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.page.TimePageLink;
+import org.thingsboard.server.common.data.sync.ie.importing.csv.BulkImportRequest;
+import org.thingsboard.server.common.data.sync.ie.importing.csv.BulkImportResult;
import org.thingsboard.server.dao.exception.IncorrectParameterException;
import org.thingsboard.server.dao.model.ModelConstants;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.asset.AssetBulkImportService;
-import org.thingsboard.server.common.data.sync.ie.importing.csv.BulkImportRequest;
-import org.thingsboard.server.common.data.sync.ie.importing.csv.BulkImportResult;
import org.thingsboard.server.service.entitiy.asset.TbAssetService;
import org.thingsboard.server.service.security.model.SecurityUser;
import org.thingsboard.server.service.security.permission.Operation;
@@ -65,6 +66,7 @@ import java.util.stream.Collectors;
import static org.thingsboard.server.controller.ControllerConstants.ASSET_ID_PARAM_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.ASSET_INFO_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.ASSET_NAME_DESCRIPTION;
+import static org.thingsboard.server.controller.ControllerConstants.ASSET_PROFILE_ID_PARAM_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.ASSET_SORT_PROPERTY_ALLOWABLE_VALUES;
import static org.thingsboard.server.controller.ControllerConstants.ASSET_TEXT_SEARCH_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.ASSET_TYPE_DESCRIPTION;
@@ -257,6 +259,8 @@ public class AssetController extends BaseController {
@RequestParam int page,
@ApiParam(value = ASSET_TYPE_DESCRIPTION)
@RequestParam(required = false) String type,
+ @ApiParam(value = ASSET_PROFILE_ID_PARAM_DESCRIPTION)
+ @RequestParam(required = false) String assetProfileId,
@ApiParam(value = ASSET_TEXT_SEARCH_DESCRIPTION)
@RequestParam(required = false) String textSearch,
@ApiParam(value = SORT_PROPERTY_DESCRIPTION, allowableValues = ASSET_SORT_PROPERTY_ALLOWABLE_VALUES)
@@ -268,6 +272,9 @@ public class AssetController extends BaseController {
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
if (type != null && type.trim().length() > 0) {
return checkNotNull(assetService.findAssetInfosByTenantIdAndType(tenantId, type, pageLink));
+ } else if (assetProfileId != null && assetProfileId.length() > 0) {
+ AssetProfileId profileId = new AssetProfileId(toUUID(assetProfileId));
+ return checkNotNull(assetService.findAssetInfosByTenantIdAndAssetProfileId(tenantId, profileId, pageLink));
} else {
return checkNotNull(assetService.findAssetInfosByTenantId(tenantId, pageLink));
}
@@ -345,6 +352,8 @@ public class AssetController extends BaseController {
@RequestParam int page,
@ApiParam(value = ASSET_TYPE_DESCRIPTION)
@RequestParam(required = false) String type,
+ @ApiParam(value = ASSET_PROFILE_ID_PARAM_DESCRIPTION)
+ @RequestParam(required = false) String assetProfileId,
@ApiParam(value = ASSET_TEXT_SEARCH_DESCRIPTION)
@RequestParam(required = false) String textSearch,
@ApiParam(value = SORT_PROPERTY_DESCRIPTION, allowableValues = ASSET_SORT_PROPERTY_ALLOWABLE_VALUES)
@@ -359,6 +368,9 @@ public class AssetController extends BaseController {
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
if (type != null && type.trim().length() > 0) {
return checkNotNull(assetService.findAssetInfosByTenantIdAndCustomerIdAndType(tenantId, customerId, type, pageLink));
+ } else if (assetProfileId != null && assetProfileId.length() > 0) {
+ AssetProfileId profileId = new AssetProfileId(toUUID(assetProfileId));
+ return checkNotNull(assetService.findAssetInfosByTenantIdAndCustomerIdAndAssetProfileId(tenantId, customerId, profileId, pageLink));
} else {
return checkNotNull(assetService.findAssetInfosByTenantIdAndCustomerId(tenantId, customerId, pageLink));
}
diff --git a/application/src/main/java/org/thingsboard/server/controller/AssetProfileController.java b/application/src/main/java/org/thingsboard/server/controller/AssetProfileController.java
new file mode 100644
index 0000000000..2e84dd6634
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/controller/AssetProfileController.java
@@ -0,0 +1,227 @@
+/**
+ * Copyright © 2016-2022 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.controller;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+import org.thingsboard.server.common.data.asset.AssetProfileInfo;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+import org.thingsboard.server.common.data.id.AssetProfileId;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.queue.util.TbCoreComponent;
+import org.thingsboard.server.service.entitiy.asset.profile.TbAssetProfileService;
+import org.thingsboard.server.service.security.permission.Operation;
+import org.thingsboard.server.service.security.permission.Resource;
+
+import static org.thingsboard.server.controller.ControllerConstants.ASSET_PROFILE_ID;
+import static org.thingsboard.server.controller.ControllerConstants.ASSET_PROFILE_ID_PARAM_DESCRIPTION;
+import static org.thingsboard.server.controller.ControllerConstants.ASSET_PROFILE_INFO_DESCRIPTION;
+import static org.thingsboard.server.controller.ControllerConstants.ASSET_PROFILE_SORT_PROPERTY_ALLOWABLE_VALUES;
+import static org.thingsboard.server.controller.ControllerConstants.ASSET_PROFILE_TEXT_SEARCH_DESCRIPTION;
+import static org.thingsboard.server.controller.ControllerConstants.NEW_LINE;
+import static org.thingsboard.server.controller.ControllerConstants.PAGE_DATA_PARAMETERS;
+import static org.thingsboard.server.controller.ControllerConstants.PAGE_NUMBER_DESCRIPTION;
+import static org.thingsboard.server.controller.ControllerConstants.PAGE_SIZE_DESCRIPTION;
+import static org.thingsboard.server.controller.ControllerConstants.SORT_ORDER_ALLOWABLE_VALUES;
+import static org.thingsboard.server.controller.ControllerConstants.SORT_ORDER_DESCRIPTION;
+import static org.thingsboard.server.controller.ControllerConstants.SORT_PROPERTY_DESCRIPTION;
+import static org.thingsboard.server.controller.ControllerConstants.TENANT_AUTHORITY_PARAGRAPH;
+import static org.thingsboard.server.controller.ControllerConstants.TENANT_OR_CUSTOMER_AUTHORITY_PARAGRAPH;
+import static org.thingsboard.server.controller.ControllerConstants.UUID_WIKI_LINK;
+
+@RestController
+@TbCoreComponent
+@RequestMapping("/api")
+@RequiredArgsConstructor
+@Slf4j
+public class AssetProfileController extends BaseController {
+
+ private final TbAssetProfileService tbAssetProfileService;
+
+ @ApiOperation(value = "Get Asset Profile (getAssetProfileById)",
+ notes = "Fetch the Asset Profile object based on the provided Asset Profile Id. " +
+ "The server checks that the asset profile is owned by the same tenant. " + TENANT_AUTHORITY_PARAGRAPH,
+ produces = "application/json")
+ @PreAuthorize("hasAnyAuthority('TENANT_ADMIN')")
+ @RequestMapping(value = "/assetProfile/{assetProfileId}", method = RequestMethod.GET)
+ @ResponseBody
+ public AssetProfile getAssetProfileById(
+ @ApiParam(value = ASSET_PROFILE_ID_PARAM_DESCRIPTION)
+ @PathVariable(ASSET_PROFILE_ID) String strAssetProfileId) throws ThingsboardException {
+ checkParameter(ASSET_PROFILE_ID, strAssetProfileId);
+ try {
+ AssetProfileId assetProfileId = new AssetProfileId(toUUID(strAssetProfileId));
+ return checkAssetProfileId(assetProfileId, Operation.READ);
+ } catch (Exception e) {
+ throw handleException(e);
+ }
+ }
+
+ @ApiOperation(value = "Get Asset Profile Info (getAssetProfileInfoById)",
+ notes = "Fetch the Asset Profile Info object based on the provided Asset Profile Id. "
+ + ASSET_PROFILE_INFO_DESCRIPTION + TENANT_OR_CUSTOMER_AUTHORITY_PARAGRAPH,
+ produces = "application/json")
+ @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
+ @RequestMapping(value = "/assetProfileInfo/{assetProfileId}", method = RequestMethod.GET)
+ @ResponseBody
+ public AssetProfileInfo getAssetProfileInfoById(
+ @ApiParam(value = ASSET_PROFILE_ID_PARAM_DESCRIPTION)
+ @PathVariable(ASSET_PROFILE_ID) String strAssetProfileId) throws ThingsboardException {
+ checkParameter(ASSET_PROFILE_ID, strAssetProfileId);
+ try {
+ AssetProfileId assetProfileId = new AssetProfileId(toUUID(strAssetProfileId));
+ return new AssetProfileInfo(checkAssetProfileId(assetProfileId, Operation.READ));
+ } catch (Exception e) {
+ throw handleException(e);
+ }
+ }
+
+ @ApiOperation(value = "Get Default Asset Profile (getDefaultAssetProfileInfo)",
+ notes = "Fetch the Default Asset Profile Info object. " +
+ ASSET_PROFILE_INFO_DESCRIPTION + TENANT_OR_CUSTOMER_AUTHORITY_PARAGRAPH,
+ produces = "application/json")
+ @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
+ @RequestMapping(value = "/assetProfileInfo/default", method = RequestMethod.GET)
+ @ResponseBody
+ public AssetProfileInfo getDefaultAssetProfileInfo() throws ThingsboardException {
+ try {
+ return checkNotNull(assetProfileService.findDefaultAssetProfileInfo(getTenantId()));
+ } catch (Exception e) {
+ throw handleException(e);
+ }
+ }
+
+ @ApiOperation(value = "Create Or Update Asset Profile (saveAssetProfile)",
+ notes = "Create or update the Asset Profile. When creating asset profile, platform generates asset profile id as " + UUID_WIKI_LINK +
+ "The newly created asset profile id will be present in the response. " +
+ "Specify existing asset profile id to update the asset profile. " +
+ "Referencing non-existing asset profile Id will cause 'Not Found' error. " + NEW_LINE +
+ "Asset profile name is unique in the scope of tenant. Only one 'default' asset profile may exist in scope of tenant. " +
+ "Remove 'id', 'tenantId' from the request body example (below) to create new Asset Profile entity. " +
+ TENANT_AUTHORITY_PARAGRAPH,
+ produces = "application/json",
+ consumes = "application/json")
+ @PreAuthorize("hasAuthority('TENANT_ADMIN')")
+ @RequestMapping(value = "/assetProfile", method = RequestMethod.POST)
+ @ResponseBody
+ public AssetProfile saveAssetProfile(
+ @ApiParam(value = "A JSON value representing the asset profile.")
+ @RequestBody AssetProfile assetProfile) throws Exception {
+ assetProfile.setTenantId(getTenantId());
+ checkEntity(assetProfile.getId(), assetProfile, Resource.ASSET_PROFILE);
+ return tbAssetProfileService.save(assetProfile, getCurrentUser());
+ }
+
+ @ApiOperation(value = "Delete asset profile (deleteAssetProfile)",
+ notes = "Deletes the asset profile. Referencing non-existing asset profile Id will cause an error. " +
+ "Can't delete the asset profile if it is referenced by existing assets." + TENANT_AUTHORITY_PARAGRAPH,
+ produces = "application/json")
+ @PreAuthorize("hasAuthority('TENANT_ADMIN')")
+ @RequestMapping(value = "/assetProfile/{assetProfileId}", method = RequestMethod.DELETE)
+ @ResponseStatus(value = HttpStatus.OK)
+ public void deleteAssetProfile(
+ @ApiParam(value = ASSET_PROFILE_ID_PARAM_DESCRIPTION)
+ @PathVariable(ASSET_PROFILE_ID) String strAssetProfileId) throws ThingsboardException {
+ checkParameter(ASSET_PROFILE_ID, strAssetProfileId);
+ AssetProfileId assetProfileId = new AssetProfileId(toUUID(strAssetProfileId));
+ AssetProfile assetProfile = checkAssetProfileId(assetProfileId, Operation.DELETE);
+ tbAssetProfileService.delete(assetProfile, getCurrentUser());
+ }
+
+ @ApiOperation(value = "Make Asset Profile Default (setDefaultAssetProfile)",
+ notes = "Marks asset profile as default within a tenant scope." + TENANT_AUTHORITY_PARAGRAPH,
+ produces = "application/json")
+ @PreAuthorize("hasAnyAuthority('TENANT_ADMIN')")
+ @RequestMapping(value = "/assetProfile/{assetProfileId}/default", method = RequestMethod.POST)
+ @ResponseBody
+ public AssetProfile setDefaultAssetProfile(
+ @ApiParam(value = ASSET_PROFILE_ID_PARAM_DESCRIPTION)
+ @PathVariable(ASSET_PROFILE_ID) String strAssetProfileId) throws ThingsboardException {
+ checkParameter(ASSET_PROFILE_ID, strAssetProfileId);
+ AssetProfileId assetProfileId = new AssetProfileId(toUUID(strAssetProfileId));
+ AssetProfile assetProfile = checkAssetProfileId(assetProfileId, Operation.WRITE);
+ AssetProfile previousDefaultAssetProfile = assetProfileService.findDefaultAssetProfile(getTenantId());
+ return tbAssetProfileService.setDefaultAssetProfile(assetProfile, previousDefaultAssetProfile, getCurrentUser());
+ }
+
+ @ApiOperation(value = "Get Asset Profiles (getAssetProfiles)",
+ notes = "Returns a page of asset profile objects owned by tenant. " +
+ PAGE_DATA_PARAMETERS + TENANT_AUTHORITY_PARAGRAPH,
+ produces = "application/json")
+ @PreAuthorize("hasAuthority('TENANT_ADMIN')")
+ @RequestMapping(value = "/assetProfiles", params = {"pageSize", "page"}, method = RequestMethod.GET)
+ @ResponseBody
+ public PageData getAssetProfiles(
+ @ApiParam(value = PAGE_SIZE_DESCRIPTION, required = true)
+ @RequestParam int pageSize,
+ @ApiParam(value = PAGE_NUMBER_DESCRIPTION, required = true)
+ @RequestParam int page,
+ @ApiParam(value = ASSET_PROFILE_TEXT_SEARCH_DESCRIPTION)
+ @RequestParam(required = false) String textSearch,
+ @ApiParam(value = SORT_PROPERTY_DESCRIPTION, allowableValues = ASSET_PROFILE_SORT_PROPERTY_ALLOWABLE_VALUES)
+ @RequestParam(required = false) String sortProperty,
+ @ApiParam(value = SORT_ORDER_DESCRIPTION, allowableValues = SORT_ORDER_ALLOWABLE_VALUES)
+ @RequestParam(required = false) String sortOrder) throws ThingsboardException {
+ try {
+ PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
+ return checkNotNull(assetProfileService.findAssetProfiles(getTenantId(), pageLink));
+ } catch (Exception e) {
+ throw handleException(e);
+ }
+ }
+
+ @ApiOperation(value = "Get Asset Profile infos (getAssetProfileInfos)",
+ notes = "Returns a page of asset profile info objects owned by tenant. " +
+ PAGE_DATA_PARAMETERS + ASSET_PROFILE_INFO_DESCRIPTION + TENANT_OR_CUSTOMER_AUTHORITY_PARAGRAPH,
+ produces = "application/json")
+ @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
+ @RequestMapping(value = "/assetProfileInfos", params = {"pageSize", "page"}, method = RequestMethod.GET)
+ @ResponseBody
+ public PageData getAssetProfileInfos(
+ @ApiParam(value = PAGE_SIZE_DESCRIPTION, required = true)
+ @RequestParam int pageSize,
+ @ApiParam(value = PAGE_NUMBER_DESCRIPTION, required = true)
+ @RequestParam int page,
+ @ApiParam(value = ASSET_PROFILE_TEXT_SEARCH_DESCRIPTION)
+ @RequestParam(required = false) String textSearch,
+ @ApiParam(value = SORT_PROPERTY_DESCRIPTION, allowableValues = ASSET_PROFILE_SORT_PROPERTY_ALLOWABLE_VALUES)
+ @RequestParam(required = false) String sortProperty,
+ @ApiParam(value = SORT_ORDER_DESCRIPTION, allowableValues = SORT_ORDER_ALLOWABLE_VALUES)
+ @RequestParam(required = false) String sortOrder) throws ThingsboardException {
+ try {
+ PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
+ return checkNotNull(assetProfileService.findAssetProfileInfos(getTenantId(), pageLink));
+ } catch (Exception e) {
+ throw handleException(e);
+ }
+ }
+}
diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java
index 1e9958cc25..ffc3fcf3cd 100644
--- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java
@@ -57,6 +57,7 @@ import org.thingsboard.server.common.data.alarm.Alarm;
import org.thingsboard.server.common.data.alarm.AlarmInfo;
import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.asset.AssetInfo;
+import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.edge.EdgeEventType;
@@ -65,6 +66,7 @@ import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.AlarmId;
import org.thingsboard.server.common.data.id.AssetId;
+import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DashboardId;
import org.thingsboard.server.common.data.id.DeviceId;
@@ -96,6 +98,7 @@ import org.thingsboard.server.common.data.rule.RuleChainType;
import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.common.data.widget.WidgetTypeDetails;
import org.thingsboard.server.common.data.widget.WidgetsBundle;
+import org.thingsboard.server.dao.asset.AssetProfileService;
import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.audit.AuditLogService;
@@ -132,6 +135,7 @@ import org.thingsboard.server.service.edge.EdgeNotificationService;
import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
import org.thingsboard.server.service.entitiy.TbNotificationEntityService;
import org.thingsboard.server.service.ota.OtaPackageStateService;
+import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.service.resource.TbResourceService;
import org.thingsboard.server.service.security.model.SecurityUser;
@@ -190,6 +194,9 @@ public abstract class BaseController {
@Autowired
protected AssetService assetService;
+ @Autowired
+ protected AssetProfileService assetProfileService;
+
@Autowired
protected AlarmSubscriptionService alarmService;
@@ -265,14 +272,14 @@ public abstract class BaseController {
@Autowired
protected TbDeviceProfileCache deviceProfileCache;
- @Autowired(required = false)
- protected EdgeService edgeService;
+ @Autowired
+ protected TbAssetProfileCache assetProfileCache;
@Autowired(required = false)
- protected EdgeNotificationService edgeNotificationService;
+ protected EdgeService edgeService;
@Autowired(required = false)
- protected EdgeRpcService edgeGrpcService;
+ protected EdgeRpcService edgeRpcService;
@Autowired
protected TbNotificationEntityService notificationEntityService;
@@ -548,6 +555,9 @@ public abstract class BaseController {
case ASSET:
checkAssetId(new AssetId(entityId.getId()), operation);
return;
+ case ASSET_PROFILE:
+ checkAssetProfileId(new AssetProfileId(entityId.getId()), operation);
+ return;
case DASHBOARD:
checkDashboardId(new DashboardId(entityId.getId()), operation);
return;
@@ -667,6 +677,18 @@ public abstract class BaseController {
}
}
+ AssetProfile checkAssetProfileId(AssetProfileId assetProfileId, Operation operation) throws ThingsboardException {
+ try {
+ validateId(assetProfileId, "Incorrect assetProfileId " + assetProfileId);
+ AssetProfile assetProfile = assetProfileService.findAssetProfileById(getCurrentUser().getTenantId(), assetProfileId);
+ checkNotNull(assetProfile, "Asset profile with id [" + assetProfileId + "] is not found");
+ accessControlService.checkPermission(getCurrentUser(), Resource.ASSET_PROFILE, operation, assetProfileId, assetProfile);
+ return assetProfile;
+ } catch (Exception e) {
+ throw handleException(e, false);
+ }
+ }
+
Alarm checkAlarmId(AlarmId alarmId, Operation operation) throws ThingsboardException {
try {
validateId(alarmId, "Incorrect alarmId " + alarmId);
diff --git a/application/src/main/java/org/thingsboard/server/controller/ControllerConstants.java b/application/src/main/java/org/thingsboard/server/controller/ControllerConstants.java
index d4199006d0..e00ccaa4bc 100644
--- a/application/src/main/java/org/thingsboard/server/controller/ControllerConstants.java
+++ b/application/src/main/java/org/thingsboard/server/controller/ControllerConstants.java
@@ -35,6 +35,8 @@ public class ControllerConstants {
protected static final String DEVICE_ID_PARAM_DESCRIPTION = "A string value representing the device id. For example, '784f394c-42b6-435a-983c-b7beff2784f9'";
protected static final String ENTITY_VIEW_ID_PARAM_DESCRIPTION = "A string value representing the entity view id. For example, '784f394c-42b6-435a-983c-b7beff2784f9'";
protected static final String DEVICE_PROFILE_ID_PARAM_DESCRIPTION = "A string value representing the device profile id. For example, '784f394c-42b6-435a-983c-b7beff2784f9'";
+
+ protected static final String ASSET_PROFILE_ID_PARAM_DESCRIPTION = "A string value representing the asset profile id. For example, '784f394c-42b6-435a-983c-b7beff2784f9'";
protected static final String TENANT_PROFILE_ID_PARAM_DESCRIPTION = "A string value representing the tenant profile id. For example, '784f394c-42b6-435a-983c-b7beff2784f9'";
protected static final String TENANT_ID_PARAM_DESCRIPTION = "A string value representing the tenant id. For example, '784f394c-42b6-435a-983c-b7beff2784f9'";
protected static final String EDGE_ID_PARAM_DESCRIPTION = "A string value representing the edge id. For example, '784f394c-42b6-435a-983c-b7beff2784f9'";
@@ -75,6 +77,8 @@ public class ControllerConstants {
protected static final String TENANT_PROFILE_TEXT_SEARCH_DESCRIPTION = "The case insensitive 'substring' filter based on the tenant profile name.";
protected static final String RULE_CHAIN_TEXT_SEARCH_DESCRIPTION = "The case insensitive 'substring' filter based on the rule chain name.";
protected static final String DEVICE_PROFILE_TEXT_SEARCH_DESCRIPTION = "The case insensitive 'substring' filter based on the device profile name.";
+
+ protected static final String ASSET_PROFILE_TEXT_SEARCH_DESCRIPTION = "The case insensitive 'substring' filter based on the asset profile name.";
protected static final String CUSTOMER_TEXT_SEARCH_DESCRIPTION = "The case insensitive 'substring' filter based on the customer title.";
protected static final String EDGE_TEXT_SEARCH_DESCRIPTION = "The case insensitive 'substring' filter based on the edge name.";
protected static final String EVENT_TEXT_SEARCH_DESCRIPTION = "The value is not used in searching.";
@@ -92,6 +96,8 @@ public class ControllerConstants {
protected static final String TENANT_PROFILE_INFO_SORT_PROPERTY_ALLOWABLE_VALUES = "id, name";
protected static final String TENANT_INFO_SORT_PROPERTY_ALLOWABLE_VALUES = "createdTime, tenantProfileName, title, email, country, state, city, address, address2, zip, phone, email";
protected static final String DEVICE_PROFILE_SORT_PROPERTY_ALLOWABLE_VALUES = "createdTime, name, type, transportType, description, isDefault";
+
+ protected static final String ASSET_PROFILE_SORT_PROPERTY_ALLOWABLE_VALUES = "createdTime, name, description, isDefault";
protected static final String ASSET_SORT_PROPERTY_ALLOWABLE_VALUES = "createdTime, name, type, label, customerTitle";
protected static final String ALARM_SORT_PROPERTY_ALLOWABLE_VALUES = "createdTime, startTs, endTs, type, ackTs, clearTs, severity, status";
protected static final String EVENT_SORT_PROPERTY_ALLOWABLE_VALUES = "ts, id";
@@ -110,6 +116,8 @@ public class ControllerConstants {
protected static final String RELATION_INFO_DESCRIPTION = "Relation Info is an extension of the default Relation object that contains information about the 'from' and 'to' entity names. ";
protected static final String EDGE_INFO_DESCRIPTION = "Edge Info is an extension of the default Edge object that contains information about the assigned customer name. ";
protected static final String DEVICE_PROFILE_INFO_DESCRIPTION = "Device Profile Info is a lightweight object that includes main information about Device Profile excluding the heavyweight configuration object. ";
+
+ protected static final String ASSET_PROFILE_INFO_DESCRIPTION = "Asset Profile Info is a lightweight object that includes main information about Asset Profile. ";
protected static final String QUEUE_SERVICE_TYPE_DESCRIPTION = "Service type (implemented only for the TB-RULE-ENGINE)";
protected static final String QUEUE_SERVICE_TYPE_ALLOWABLE_VALUES = "TB-RULE-ENGINE, TB-CORE, TB-TRANSPORT, JS-EXECUTOR";
protected static final String QUEUE_QUEUE_TEXT_SEARCH_DESCRIPTION = "The case insensitive 'substring' filter based on the queue name.";
@@ -1415,6 +1423,8 @@ public class ControllerConstants {
protected static final String DEVICE_PROFILE_ID = "deviceProfileId";
+ protected static final String ASSET_PROFILE_ID = "assetProfileId";
+
protected static final String MODEL_DESCRIPTION = "See the 'Model' tab for more details.";
protected static final String ENTITY_VIEW_DESCRIPTION = "Entity Views limit the degree of exposure of the Device or Asset telemetry and attributes to the Customers. " +
"Every Entity View references exactly one entity (device or asset) and defines telemetry and attribute keys that will be visible to the assigned Customer. " +
diff --git a/application/src/main/java/org/thingsboard/server/controller/DeviceProfileController.java b/application/src/main/java/org/thingsboard/server/controller/DeviceProfileController.java
index 0baf89e0dc..bd2a5a2ad9 100644
--- a/application/src/main/java/org/thingsboard/server/controller/DeviceProfileController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/DeviceProfileController.java
@@ -191,7 +191,7 @@ public class DeviceProfileController extends BaseController {
"Specify existing device profile id to update the device profile. " +
"Referencing non-existing device profile Id will cause 'Not Found' error. " + NEW_LINE +
"Device profile name is unique in the scope of tenant. Only one 'default' device profile may exist in scope of tenant." + DEVICE_PROFILE_DATA +
- "Remove 'id', 'tenantId' and optionally 'customerId' from the request body example (below) to create new Device Profile entity. " +
+ "Remove 'id', 'tenantId' from the request body example (below) to create new Device Profile entity. " +
TENANT_AUTHORITY_PARAGRAPH,
produces = "application/json",
consumes = "application/json")
diff --git a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java
index 59c9eb39c1..6a5dd44d0b 100644
--- a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java
@@ -22,6 +22,7 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
@@ -32,6 +33,7 @@ import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.context.request.async.DeferredResult;
import org.thingsboard.rule.engine.flow.TbRuleChainInputNode;
import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.EntitySubtype;
@@ -47,6 +49,8 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.rule.RuleChain;
+import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
+import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
import org.thingsboard.server.dao.exception.DataValidationException;
import org.thingsboard.server.dao.exception.IncorrectParameterException;
import org.thingsboard.server.dao.model.ModelConstants;
@@ -61,6 +65,7 @@ import org.thingsboard.server.service.security.permission.Resource;
import java.util.ArrayList;
import java.util.List;
+import java.util.UUID;
import java.util.stream.Collectors;
import static org.thingsboard.server.controller.ControllerConstants.CUSTOMER_ID_PARAM_DESCRIPTION;
@@ -529,24 +534,35 @@ public class EdgeController extends BaseController {
"All entities that are assigned to particular edge are going to be send to remote edge service." + TENANT_AUTHORITY_PARAGRAPH)
@PreAuthorize("hasAuthority('TENANT_ADMIN')")
@RequestMapping(value = "/edge/sync/{edgeId}", method = RequestMethod.POST)
- public void syncEdge(@ApiParam(value = EDGE_ID_PARAM_DESCRIPTION, required = true)
+ public DeferredResult syncEdge(@ApiParam(value = EDGE_ID_PARAM_DESCRIPTION, required = true)
@PathVariable("edgeId") String strEdgeId) throws ThingsboardException {
checkParameter("edgeId", strEdgeId);
try {
+ final DeferredResult response = new DeferredResult<>();
if (isEdgesEnabled()) {
EdgeId edgeId = new EdgeId(toUUID(strEdgeId));
edgeId = checkNotNull(edgeId);
SecurityUser user = getCurrentUser();
TenantId tenantId = user.getTenantId();
- edgeGrpcService.startSyncProcess(tenantId, edgeId);
+ ToEdgeSyncRequest request = new ToEdgeSyncRequest(UUID.randomUUID(), tenantId, edgeId);
+ edgeRpcService.processSyncRequest(request, fromEdgeSyncResponse -> reply(response, fromEdgeSyncResponse));
} else {
throw new ThingsboardException("Edges support disabled", ThingsboardErrorCode.GENERAL);
}
+ return response;
} catch (Exception e) {
throw handleException(e);
}
}
+ private void reply(DeferredResult response, FromEdgeSyncResponse fromEdgeSyncResponse) {
+ if (fromEdgeSyncResponse.isSuccess()) {
+ response.setResult(new ResponseEntity<>(HttpStatus.OK));
+ } else {
+ response.setErrorResult(new ThingsboardException("Edge is not connected", ThingsboardErrorCode.GENERAL));
+ }
+ }
+
@ApiOperation(value = "Find missing rule chains (findMissingToRelatedRuleChains)",
notes = "Returns list of rule chains ids that are not assigned to particular edge, but these rule chains are present in the already assigned rule chains to edge." + TENANT_AUTHORITY_PARAGRAPH)
@PreAuthorize("hasAuthority('TENANT_ADMIN')")
diff --git a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
index 2c3251834f..33a6d4bdd1 100644
--- a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
+++ b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
@@ -225,6 +225,10 @@ public class ThingsboardInstallService {
log.info("Upgrading ThingsBoard from version 3.4.0 to 3.4.1 ...");
databaseEntitiesUpgradeService.upgradeDatabase("3.4.0");
dataUpdateService.updateData("3.4.0");
+ case "3.4.1":
+ log.info("Upgrading ThingsBoard from version 3.4.1 to 3.4.2 ...");
+ databaseEntitiesUpgradeService.upgradeDatabase("3.4.1");
+ dataUpdateService.updateData("3.4.1");
log.info("Updating system data...");
systemDataLoaderService.updateSystemWidgets();
break;
diff --git a/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java b/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java
index a6cd3b735f..7057f903be 100644
--- a/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java
+++ b/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java
@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.EntityType;
@@ -34,9 +35,8 @@ import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
-import org.thingsboard.server.common.data.kv.DataType;
-import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgDataType;
import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -113,6 +113,15 @@ public class EntityActionService {
case UNASSIGNED_FROM_EDGE:
msgType = DataConstants.ENTITY_UNASSIGNED_FROM_EDGE;
break;
+ case RELATION_ADD_OR_UPDATE:
+ msgType = DataConstants.RELATION_ADD_OR_UPDATE;
+ break;
+ case RELATION_DELETED:
+ msgType = DataConstants.RELATION_DELETED;
+ break;
+ case RELATIONS_DELETED:
+ msgType = DataConstants.RELATIONS_DELETED;
+ break;
}
if (!StringUtils.isEmpty(msgType)) {
try {
@@ -170,7 +179,7 @@ public class EntityActionService {
metaData.putValue(DataConstants.SCOPE, scope);
if (attributes != null) {
for (AttributeKvEntry attr : attributes) {
- addKvEntry(entityNode, attr);
+ JacksonUtil.addKvEntry(entityNode, attr);
}
}
} else if (actionType == ActionType.ATTRIBUTES_DELETED) {
@@ -195,10 +204,12 @@ public class EntityActionService {
}
entityNode.put("startTs", extractParameter(Long.class, 1, additionalInfo));
entityNode.put("endTs", extractParameter(Long.class, 2, additionalInfo));
+ } else if (ActionType.RELATION_ADD_OR_UPDATE.equals(actionType) || ActionType.RELATION_DELETED.equals(actionType)) {
+ entityNode = json.valueToTree(extractParameter(EntityRelation.class, 0, additionalInfo));
}
}
TbMsg tbMsg = TbMsg.newMsg(msgType, entityId, customerId, metaData, TbMsgDataType.JSON, json.writeValueAsString(entityNode));
- if (tenantId.isNullUid()) {
+ if (tenantId == null || tenantId.isNullUid()) {
if (entity instanceof HasTenantId) {
tenantId = ((HasTenantId) entity).getTenantId();
}
@@ -246,27 +257,11 @@ public class EntityActionService {
element.put("ts", entry.getKey());
ObjectNode values = element.putObject("values");
for (TsKvEntry tsKvEntry : entry.getValue()) {
- addKvEntry(values, tsKvEntry);
+ JacksonUtil.addKvEntry(values, tsKvEntry);
}
result.add(element);
}
}
}
- private void addKvEntry(ObjectNode entityNode, KvEntry kvEntry) throws Exception {
- if (kvEntry.getDataType() == DataType.BOOLEAN) {
- kvEntry.getBooleanValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value));
- } else if (kvEntry.getDataType() == DataType.DOUBLE) {
- kvEntry.getDoubleValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value));
- } else if (kvEntry.getDataType() == DataType.LONG) {
- kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value));
- } else if (kvEntry.getDataType() == DataType.JSON) {
- if (kvEntry.getJsonValue().isPresent()) {
- entityNode.set(kvEntry.getKey(), json.readTree(kvEntry.getJsonValue().get()));
- }
- } else {
- entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString());
- }
- }
-
}
diff --git a/application/src/main/java/org/thingsboard/server/service/asset/AssetBulkImportService.java b/application/src/main/java/org/thingsboard/server/service/asset/AssetBulkImportService.java
index 47c9d905ad..5142311da9 100644
--- a/application/src/main/java/org/thingsboard/server/service/asset/AssetBulkImportService.java
+++ b/application/src/main/java/org/thingsboard/server/service/asset/AssetBulkImportService.java
@@ -21,9 +21,12 @@ import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.ie.importing.csv.BulkImportColumnType;
+import org.thingsboard.server.dao.asset.AssetProfileService;
import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.entitiy.asset.TbAssetService;
@@ -39,6 +42,7 @@ import java.util.Optional;
public class AssetBulkImportService extends AbstractBulkImportService {
private final AssetService assetService;
private final TbAssetService tbAssetService;
+ private final AssetProfileService assetProfileService;
@Override
protected void setEntityFields(Asset entity, Map fields) {
@@ -65,6 +69,13 @@ public class AssetBulkImportService extends AbstractBulkImportService {
@Override
@SneakyThrows
protected Asset saveEntity(SecurityUser user, Asset entity, Map fields) {
+ AssetProfile assetProfile;
+ if (StringUtils.isNotEmpty(entity.getType())) {
+ assetProfile = assetProfileService.findOrCreateAssetProfile(entity.getTenantId(), entity.getType());
+ } else {
+ assetProfile = assetProfileService.findDefaultAssetProfile(entity.getTenantId());
+ }
+ entity.setAssetProfileId(assetProfile.getId());
return tbAssetService.save(entity, user);
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
index 5fa58a5bb5..23fb8fd4be 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
@@ -134,6 +134,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
case ASSET:
case DEVICE:
case DEVICE_PROFILE:
+ case ASSET_PROFILE:
case ENTITY_VIEW:
case DASHBOARD:
case RULE_CHAIN:
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java
index 29ff49d626..8bc781b5fc 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java
@@ -21,6 +21,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import org.thingsboard.server.cluster.TbClusterService;
+import org.thingsboard.server.dao.asset.AssetProfileService;
import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.dashboard.DashboardService;
@@ -38,6 +39,7 @@ import org.thingsboard.server.service.edge.rpc.EdgeEventStorageSettings;
import org.thingsboard.server.service.edge.rpc.processor.AdminSettingsEdgeProcessor;
import org.thingsboard.server.service.edge.rpc.processor.AlarmEdgeProcessor;
import org.thingsboard.server.service.edge.rpc.processor.AssetEdgeProcessor;
+import org.thingsboard.server.service.edge.rpc.processor.AssetProfileEdgeProcessor;
import org.thingsboard.server.service.edge.rpc.processor.CustomerEdgeProcessor;
import org.thingsboard.server.service.edge.rpc.processor.DashboardEdgeProcessor;
import org.thingsboard.server.service.edge.rpc.processor.DeviceEdgeProcessor;
@@ -83,6 +85,9 @@ public class EdgeContextComponent {
@Autowired
private DeviceProfileService deviceProfileService;
+ @Autowired
+ private AssetProfileService assetProfileService;
+
@Autowired
private AttributesService attributesService;
@@ -113,6 +118,9 @@ public class EdgeContextComponent {
@Autowired
private DeviceProfileEdgeProcessor deviceProfileProcessor;
+ @Autowired
+ private AssetProfileEdgeProcessor assetProfileProcessor;
+
@Autowired
private DeviceEdgeProcessor deviceProcessor;
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
index c56fcd4ed9..4cb6fcdb0a 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
@@ -26,6 +26,7 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.ResourceUtils;
import org.thingsboard.server.common.data.edge.Edge;
@@ -34,6 +35,10 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
+import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
+import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
+import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc;
import org.thingsboard.server.gen.edge.v1.RequestMsg;
import org.thingsboard.server.gen.edge.v1.ResponseMsg;
@@ -59,6 +64,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
@Service
@Slf4j
@@ -71,6 +77,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private final Map sessionNewEvents = new HashMap<>();
private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>();
+ private final ConcurrentMap> localSyncEdgeRequests = new ConcurrentHashMap<>();
+
@Value("${edges.rpc.port}")
private int rpcPort;
@Value("${edges.rpc.ssl.enabled}")
@@ -98,12 +106,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
@Autowired
private TelemetrySubscriptionService tsSubService;
+ @Autowired
+ private TbClusterService clusterService;
+
private Server server;
private ScheduledExecutorService edgeEventProcessingExecutorService;
private ScheduledExecutorService sendDownlinkExecutorService;
+ private ScheduledExecutorService executorService;
+
@PostConstruct
public void init() {
log.info("Initializing Edge RPC service!");
@@ -129,8 +142,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
log.error("Failed to start Edge RPC server!", e);
throw new RuntimeException("Failed to start Edge RPC server!");
}
- this.edgeEventProcessingExecutorService = Executors.newScheduledThreadPool(schedulerPoolSize, ThingsBoardThreadFactory.forName("edge-scheduler"));
+ this.edgeEventProcessingExecutorService = Executors.newScheduledThreadPool(schedulerPoolSize, ThingsBoardThreadFactory.forName("edge-event-check-scheduler"));
this.sendDownlinkExecutorService = Executors.newScheduledThreadPool(sendSchedulerPoolSize, ThingsBoardThreadFactory.forName("edge-send-scheduler"));
+ this.executorService = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("edge-service"));
log.info("Edge RPC service initialized!");
}
@@ -153,6 +167,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
if (sendDownlinkExecutorService != null) {
sendDownlinkExecutorService.shutdownNow();
}
+ if (executorService != null) {
+ executorService.shutdownNow();
+ }
}
@Override
@@ -160,47 +177,76 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
return new EdgeGrpcSession(ctx, outputStream, this::onEdgeConnect, this::onEdgeDisconnect, sendDownlinkExecutorService).getInputStream();
}
+ @Override
+ public void onToEdgeSessionMsg(TenantId tenantId, EdgeSessionMsg msg) {
+ executorService.execute(() -> {
+ switch (msg.getMsgType()) {
+ case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
+ EdgeEventUpdateMsg edgeEventUpdateMsg = (EdgeEventUpdateMsg) msg;
+ log.trace("[{}] onToEdgeSessionMsg [{}]", edgeEventUpdateMsg.getTenantId(), msg);
+ onEdgeEvent(tenantId, edgeEventUpdateMsg.getEdgeId());
+ break;
+ case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG:
+ ToEdgeSyncRequest toEdgeSyncRequest = (ToEdgeSyncRequest) msg;
+ log.trace("[{}] toEdgeSyncRequest [{}]", toEdgeSyncRequest.getTenantId(), msg);
+ startSyncProcess(tenantId, toEdgeSyncRequest.getEdgeId(), toEdgeSyncRequest.getId());
+ break;
+ case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG:
+ FromEdgeSyncResponse fromEdgeSyncResponse = (FromEdgeSyncResponse) msg;
+ log.trace("[{}] fromEdgeSyncResponse [{}]", fromEdgeSyncResponse.getTenantId(), msg);
+ processSyncResponse(fromEdgeSyncResponse);
+ break;
+ }
+ });
+ }
+
@Override
public void updateEdge(TenantId tenantId, Edge edge) {
- EdgeGrpcSession session = sessions.get(edge.getId());
- if (session != null && session.isConnected()) {
- log.debug("[{}] Updating configuration for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
- session.onConfigurationUpdate(edge);
- } else {
- log.debug("[{}] Session doesn't exist for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
- }
+ executorService.execute(() -> {
+ EdgeGrpcSession session = sessions.get(edge.getId());
+ if (session != null && session.isConnected()) {
+ log.debug("[{}] Updating configuration for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
+ session.onConfigurationUpdate(edge);
+ } else {
+ log.debug("[{}] Session doesn't exist for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
+ }
+ });
}
@Override
public void deleteEdge(TenantId tenantId, EdgeId edgeId) {
+ executorService.execute(() -> {
+ EdgeGrpcSession session = sessions.get(edgeId);
+ if (session != null && session.isConnected()) {
+ log.info("[{}] Closing and removing session for edge [{}]", tenantId, edgeId);
+ session.close();
+ sessions.remove(edgeId);
+ final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
+ newEventLock.lock();
+ try {
+ sessionNewEvents.remove(edgeId);
+ } finally {
+ newEventLock.unlock();
+ }
+ cancelScheduleEdgeEventsCheck(edgeId);
+ }
+ });
+ }
+
+ private void onEdgeEvent(TenantId tenantId, EdgeId edgeId) {
EdgeGrpcSession session = sessions.get(edgeId);
if (session != null && session.isConnected()) {
- log.info("[{}] Closing and removing session for edge [{}]", tenantId, edgeId);
- session.close();
- sessions.remove(edgeId);
+ log.trace("[{}] onEdgeEvent [{}]", tenantId, edgeId.getId());
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock();
try {
- sessionNewEvents.remove(edgeId);
+ if (Boolean.FALSE.equals(sessionNewEvents.get(edgeId))) {
+ log.trace("[{}] set session new events flag to true [{}]", tenantId, edgeId.getId());
+ sessionNewEvents.put(edgeId, true);
+ }
} finally {
newEventLock.unlock();
}
- cancelScheduleEdgeEventsCheck(edgeId);
- }
- }
-
- @Override
- public void onEdgeEvent(TenantId tenantId, EdgeId edgeId) {
- log.trace("[{}] onEdgeEvent [{}]", tenantId, edgeId.getId());
- final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
- newEventLock.lock();
- try {
- if (Boolean.FALSE.equals(sessionNewEvents.get(edgeId))) {
- log.trace("[{}] set session new events flag to true [{}]", tenantId, edgeId.getId());
- sessionNewEvents.put(edgeId, true);
- }
- } finally {
- newEventLock.unlock();
}
}
@@ -220,14 +266,47 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
scheduleEdgeEventsCheck(edgeGrpcSession);
}
- @Override
- public void startSyncProcess(TenantId tenantId, EdgeId edgeId) {
+ private void startSyncProcess(TenantId tenantId, EdgeId edgeId, UUID requestId) {
EdgeGrpcSession session = sessions.get(edgeId);
- if (session != null && session.isConnected()) {
- session.startSyncProcess(tenantId, edgeId);
+ if (session != null) {
+ boolean success = false;
+ if (session.isConnected()) {
+ session.startSyncProcess(tenantId, edgeId);
+ success = true;
+ }
+ clusterService.pushEdgeSyncResponseToCore(new FromEdgeSyncResponse(requestId, tenantId, edgeId, success));
+ }
+ }
+
+ @Override
+ public void processSyncRequest(ToEdgeSyncRequest request, Consumer responseConsumer) {
+ log.trace("[{}][{}] Processing sync edge request [{}]", request.getTenantId(), request.getId(), request.getEdgeId());
+ UUID requestId = request.getId();
+ localSyncEdgeRequests.put(requestId, responseConsumer);
+ clusterService.pushEdgeSyncRequestToCore(request);
+ scheduleSyncRequestTimeout(request, requestId);
+ }
+
+ private void scheduleSyncRequestTimeout(ToEdgeSyncRequest request, UUID requestId) {
+ log.trace("[{}] scheduling sync edge request", requestId);
+ executorService.schedule(() -> {
+ log.trace("[{}] checking if sync edge request is not processed...", requestId);
+ Consumer consumer = localSyncEdgeRequests.remove(requestId);
+ if (consumer != null) {
+ log.trace("[{}] timeout for processing sync edge request.", requestId);
+ consumer.accept(new FromEdgeSyncResponse(requestId, request.getTenantId(), request.getEdgeId(), false));
+ }
+ }, 20, TimeUnit.SECONDS);
+ }
+
+ private void processSyncResponse(FromEdgeSyncResponse response) {
+ log.trace("[{}] Received response from sync service: [{}]", response.getId(), response);
+ UUID requestId = response.getId();
+ Consumer consumer = localSyncEdgeRequests.remove(requestId);
+ if (consumer != null) {
+ consumer.accept(response);
} else {
- log.error("[{}] Edge is not connected [{}]", tenantId, edgeId);
- throw new RuntimeException("Edge is not connected");
+ log.trace("[{}] Unknown or stale sync response received [{}]", requestId, response);
}
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
index 3f8950f281..5f1c03b35a 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
@@ -38,6 +38,7 @@ import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.gen.edge.v1.AlarmUpdateMsg;
+import org.thingsboard.server.gen.edge.v1.AssetProfileAssetsRequestMsg;
import org.thingsboard.server.gen.edge.v1.AttributesRequestMsg;
import org.thingsboard.server.gen.edge.v1.ConnectRequestMsg;
import org.thingsboard.server.gen.edge.v1.ConnectResponseCode;
@@ -513,6 +514,8 @@ public final class EdgeGrpcSession implements Closeable {
return ctx.getDeviceProcessor().processDeviceToEdge(edge, edgeEvent, msgType, action);
case DEVICE_PROFILE:
return ctx.getDeviceProfileProcessor().processDeviceProfileToEdge(edgeEvent, msgType, action);
+ case ASSET_PROFILE:
+ return ctx.getAssetProfileProcessor().processAssetProfileToEdge(edgeEvent, msgType, action);
case ASSET:
return ctx.getAssetProcessor().processAssetToEdge(edge, edgeEvent, msgType, action);
case ENTITY_VIEW:
@@ -634,6 +637,11 @@ public final class EdgeGrpcSession implements Closeable {
result.add(ctx.getEdgeRequestsService().processDeviceProfileDevicesRequestMsg(edge.getTenantId(), edge, deviceProfileDevicesRequestMsg));
}
}
+ if (uplinkMsg.getAssetProfileAssetsRequestMsgCount() > 0) {
+ for (AssetProfileAssetsRequestMsg assetProfileAssetsRequestMsg : uplinkMsg.getAssetProfileAssetsRequestMsgList()) {
+ result.add(ctx.getEdgeRequestsService().processAssetProfileAssetsRequestMsg(edge.getTenantId(), edge, assetProfileAssetsRequestMsg));
+ }
+ }
if (uplinkMsg.getWidgetBundleTypesRequestMsgCount() > 0) {
for (WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg : uplinkMsg.getWidgetBundleTypesRequestMsgList()) {
result.add(ctx.getEdgeRequestsService().processWidgetBundleTypesRequestMsg(edge.getTenantId(), edge, widgetBundleTypesRequestMsg));
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java
index ad65c15235..6c5515337e 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java
@@ -18,14 +18,19 @@ package org.thingsboard.server.service.edge.rpc;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
+import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
+import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
+
+import java.util.function.Consumer;
public interface EdgeRpcService {
+ void onToEdgeSessionMsg(TenantId tenantId, EdgeSessionMsg msg);
+
void updateEdge(TenantId tenantId, Edge edge);
void deleteEdge(TenantId tenantId, EdgeId edgeId);
- void onEdgeEvent(TenantId tenantId, EdgeId edgeId);
-
- void startSyncProcess(TenantId tenantId, EdgeId edgeId);
+ void processSyncRequest(ToEdgeSyncRequest request, Consumer responseConsumer);
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java
index 1df44dd589..d6bfff76f6 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java
@@ -19,6 +19,7 @@ import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.service.edge.EdgeContextComponent;
import org.thingsboard.server.service.edge.rpc.fetch.AdminSettingsEdgeEventFetcher;
+import org.thingsboard.server.service.edge.rpc.fetch.AssetProfilesEdgeEventFetcher;
import org.thingsboard.server.service.edge.rpc.fetch.AssetsEdgeEventFetcher;
import org.thingsboard.server.service.edge.rpc.fetch.CustomerEdgeEventFetcher;
import org.thingsboard.server.service.edge.rpc.fetch.CustomerUsersEdgeEventFetcher;
@@ -47,6 +48,7 @@ public class EdgeSyncCursor {
fetchers.add(new RuleChainsEdgeEventFetcher(ctx.getRuleChainService()));
fetchers.add(new AdminSettingsEdgeEventFetcher(ctx.getAdminSettingsService(), ctx.getFreemarkerConfig()));
fetchers.add(new DeviceProfilesEdgeEventFetcher(ctx.getDeviceProfileService()));
+ fetchers.add(new AssetProfilesEdgeEventFetcher(ctx.getAssetProfileService()));
fetchers.add(new TenantAdminUsersEdgeEventFetcher(ctx.getUserService()));
if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) {
fetchers.add(new CustomerEdgeEventFetcher());
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java
index 0a78afa6b7..fb84d7f85a 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java
@@ -42,6 +42,10 @@ public class AssetMsgConstructor {
builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits());
builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits());
}
+ if (asset.getAssetProfileId() != null) {
+ builder.setAssetProfileIdMSB(asset.getAssetProfileId().getId().getMostSignificantBits());
+ builder.setAssetProfileIdLSB(asset.getAssetProfileId().getId().getLeastSignificantBits());
+ }
if (asset.getAdditionalInfo() != null) {
builder.setAdditionalInfo(JacksonUtil.toString(asset.getAdditionalInfo()));
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetProfileMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetProfileMsgConstructor.java
new file mode 100644
index 0000000000..78f7b0327e
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetProfileMsgConstructor.java
@@ -0,0 +1,67 @@
+/**
+ * Copyright © 2016-2022 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.service.edge.rpc.constructor;
+
+import com.google.protobuf.ByteString;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+import org.thingsboard.server.common.data.id.AssetProfileId;
+import org.thingsboard.server.gen.edge.v1.AssetProfileUpdateMsg;
+import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
+import org.thingsboard.server.queue.util.DataDecodingEncodingService;
+import org.thingsboard.server.queue.util.TbCoreComponent;
+
+import java.nio.charset.StandardCharsets;
+
+@Component
+@TbCoreComponent
+public class AssetProfileMsgConstructor {
+
+ @Autowired
+ private DataDecodingEncodingService dataDecodingEncodingService;
+
+ public AssetProfileUpdateMsg constructAssetProfileUpdatedMsg(UpdateMsgType msgType, AssetProfile assetProfile) {
+ AssetProfileUpdateMsg.Builder builder = AssetProfileUpdateMsg.newBuilder()
+ .setMsgType(msgType)
+ .setIdMSB(assetProfile.getId().getId().getMostSignificantBits())
+ .setIdLSB(assetProfile.getId().getId().getLeastSignificantBits())
+ .setName(assetProfile.getName())
+ .setDefault(assetProfile.isDefault());
+ if (assetProfile.getDefaultRuleChainId() != null) {
+ builder.setDefaultRuleChainIdMSB(assetProfile.getDefaultRuleChainId().getId().getMostSignificantBits())
+ .setDefaultRuleChainIdLSB(assetProfile.getDefaultRuleChainId().getId().getLeastSignificantBits());
+ }
+ if (assetProfile.getDefaultQueueName() != null) {
+ builder.setDefaultQueueName(assetProfile.getDefaultQueueName());
+ }
+ if (assetProfile.getDescription() != null) {
+ builder.setDescription(assetProfile.getDescription());
+ }
+ if (assetProfile.getImage() != null) {
+ builder.setImage(ByteString.copyFrom(assetProfile.getImage().getBytes(StandardCharsets.UTF_8)));
+ }
+ return builder.build();
+ }
+
+ public AssetProfileUpdateMsg constructAssetProfileDeleteMsg(AssetProfileId assetProfileId) {
+ return AssetProfileUpdateMsg.newBuilder()
+ .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE)
+ .setIdMSB(assetProfileId.getId().getMostSignificantBits())
+ .setIdLSB(assetProfileId.getId().getLeastSignificantBits()).build();
+ }
+
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/AssetProfilesEdgeEventFetcher.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/AssetProfilesEdgeEventFetcher.java
new file mode 100644
index 0000000000..efade1b235
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/AssetProfilesEdgeEventFetcher.java
@@ -0,0 +1,47 @@
+/**
+ * Copyright © 2016-2022 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.service.edge.rpc.fetch;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.common.data.EdgeUtils;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+import org.thingsboard.server.common.data.edge.Edge;
+import org.thingsboard.server.common.data.edge.EdgeEvent;
+import org.thingsboard.server.common.data.edge.EdgeEventActionType;
+import org.thingsboard.server.common.data.edge.EdgeEventType;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.dao.asset.AssetProfileService;
+
+@AllArgsConstructor
+@Slf4j
+public class AssetProfilesEdgeEventFetcher extends BasePageableEdgeEventFetcher {
+
+ private final AssetProfileService assetProfileService;
+
+ @Override
+ PageData fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
+ return assetProfileService.findAssetProfiles(tenantId, pageLink);
+ }
+
+ @Override
+ EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, AssetProfile assetProfile) {
+ return EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ASSET_PROFILE,
+ EdgeEventActionType.ADDED, assetProfile.getId(), null);
+ }
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetProfileEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetProfileEdgeProcessor.java
new file mode 100644
index 0000000000..ec11764fea
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetProfileEdgeProcessor.java
@@ -0,0 +1,63 @@
+/**
+ * Copyright © 2016-2022 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.service.edge.rpc.processor;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.EdgeUtils;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+import org.thingsboard.server.common.data.edge.EdgeEvent;
+import org.thingsboard.server.common.data.edge.EdgeEventActionType;
+import org.thingsboard.server.common.data.id.AssetProfileId;
+import org.thingsboard.server.gen.edge.v1.AssetProfileUpdateMsg;
+import org.thingsboard.server.gen.edge.v1.DownlinkMsg;
+import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
+import org.thingsboard.server.queue.util.TbCoreComponent;
+
+@Component
+@Slf4j
+@TbCoreComponent
+public class AssetProfileEdgeProcessor extends BaseEdgeProcessor {
+
+ public DownlinkMsg processAssetProfileToEdge(EdgeEvent edgeEvent, UpdateMsgType msgType, EdgeEventActionType action) {
+ AssetProfileId assetProfileId = new AssetProfileId(edgeEvent.getEntityId());
+ DownlinkMsg downlinkMsg = null;
+ switch (action) {
+ case ADDED:
+ case UPDATED:
+ AssetProfile assetProfile = assetProfileService.findAssetProfileById(edgeEvent.getTenantId(), assetProfileId);
+ if (assetProfile != null) {
+ AssetProfileUpdateMsg assetProfileUpdateMsg =
+ assetProfileMsgConstructor.constructAssetProfileUpdatedMsg(msgType, assetProfile);
+ downlinkMsg = DownlinkMsg.newBuilder()
+ .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
+ .addAssetProfileUpdateMsg(assetProfileUpdateMsg)
+ .build();
+ }
+ break;
+ case DELETED:
+ AssetProfileUpdateMsg assetProfileUpdateMsg =
+ assetProfileMsgConstructor.constructAssetProfileDeleteMsg(assetProfileId);
+ downlinkMsg = DownlinkMsg.newBuilder()
+ .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
+ .addAssetProfileUpdateMsg(assetProfileUpdateMsg)
+ .build();
+ break;
+ }
+ return downlinkMsg;
+ }
+
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
index 9d6199bdcd..de71fa32a3 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
@@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.dao.alarm.AlarmService;
+import org.thingsboard.server.dao.asset.AssetProfileService;
import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.customer.CustomerService;
@@ -61,6 +62,7 @@ import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
import org.thingsboard.server.service.edge.rpc.constructor.AdminSettingsMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.AlarmMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.AssetMsgConstructor;
+import org.thingsboard.server.service.edge.rpc.constructor.AssetProfileMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.CustomerMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.DashboardMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.DeviceMsgConstructor;
@@ -74,9 +76,12 @@ import org.thingsboard.server.service.edge.rpc.constructor.RuleChainMsgConstruct
import org.thingsboard.server.service.edge.rpc.constructor.UserMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.WidgetTypeMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.WidgetsBundleMsgConstructor;
+import org.thingsboard.server.service.entitiy.TbNotificationEntityService;
import org.thingsboard.server.service.executors.DbCallbackExecutorService;
+import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.service.state.DeviceStateService;
+import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.util.ArrayList;
import java.util.List;
@@ -88,6 +93,12 @@ public abstract class BaseEdgeProcessor {
protected static final int DEFAULT_PAGE_SIZE = 1000;
+ @Autowired
+ protected TelemetrySubscriptionService tsSubService;
+
+ @Autowired
+ protected TbNotificationEntityService notificationEntityService;
+
@Autowired
protected RuleChainService ruleChainService;
@@ -100,6 +111,9 @@ public abstract class BaseEdgeProcessor {
@Autowired
protected TbDeviceProfileCache deviceProfileCache;
+ @Autowired
+ protected TbAssetProfileCache assetProfileCache;
+
@Autowired
protected DashboardService dashboardService;
@@ -124,6 +138,9 @@ public abstract class BaseEdgeProcessor {
@Autowired
protected DeviceProfileService deviceProfileService;
+ @Autowired
+ protected AssetProfileService assetProfileService;
+
@Autowired
protected RelationService relationService;
@@ -197,6 +214,9 @@ public abstract class BaseEdgeProcessor {
@Autowired
protected DeviceProfileMsgConstructor deviceProfileMsgConstructor;
+ @Autowired
+ protected AssetProfileMsgConstructor assetProfileMsgConstructor;
+
@Autowired
protected WidgetsBundleMsgConstructor widgetsBundleMsgConstructor;
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
index ad6446cb95..171d5518fc 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
@@ -35,6 +35,8 @@ import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.EntityView;
import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.audit.ActionType;
+import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.id.AssetId;
@@ -44,7 +46,6 @@ import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityViewId;
-import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UserId;
@@ -57,6 +58,7 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
import org.thingsboard.server.common.transport.util.JsonUtils;
+import org.thingsboard.server.controller.BaseController;
import org.thingsboard.server.gen.edge.v1.AttributeDeleteMsg;
import org.thingsboard.server.gen.edge.v1.DownlinkMsg;
import org.thingsboard.server.gen.edge.v1.EntityDataProto;
@@ -101,7 +103,7 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
}
if (entityData.hasAttributesUpdatedMsg()) {
metaData.putValue("scope", entityData.getPostAttributeScope());
- result.add(processAttributesUpdate(tenantId, customerId, entityId, entityData.getAttributesUpdatedMsg(), metaData));
+ result.add(processAttributesUpdate(tenantId, entityId, entityData.getAttributesUpdatedMsg(), metaData));
}
if (entityData.hasPostTelemetryMsg()) {
result.add(processPostTelemetry(tenantId, customerId, entityId, entityData.getPostTelemetryMsg(), metaData));
@@ -159,23 +161,26 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
}
private Pair getDefaultQueueNameAndRuleChainId(TenantId tenantId, EntityId entityId) {
+ RuleChainId ruleChainId = null;
+ String queueName = null;
if (EntityType.DEVICE.equals(entityId.getEntityType())) {
DeviceProfile deviceProfile = deviceProfileCache.get(tenantId, new DeviceId(entityId.getId()));
- RuleChainId ruleChainId;
- String queueName;
-
if (deviceProfile == null) {
log.warn("[{}] Device profile is null!", entityId);
- ruleChainId = null;
- queueName = null;
} else {
ruleChainId = deviceProfile.getDefaultRuleChainId();
queueName = deviceProfile.getDefaultQueueName();
}
- return new ImmutablePair<>(queueName, ruleChainId);
- } else {
- return new ImmutablePair<>(null, null);
+ } else if (EntityType.ASSET.equals(entityId.getEntityType())) {
+ AssetProfile assetProfile = assetProfileCache.get(tenantId, new AssetId(entityId.getId()));
+ if (assetProfile == null) {
+ log.warn("[{}] Asset profile is null!", entityId);
+ } else {
+ ruleChainId = assetProfile.getDefaultRuleChainId();
+ queueName = assetProfile.getDefaultQueueName();
+ }
}
+ return new ImmutablePair<>(queueName, ruleChainId);
}
private ListenableFuture processPostTelemetry(TenantId tenantId, CustomerId customerId, EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) {
@@ -221,39 +226,36 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
return futureToSet;
}
- private ListenableFuture processAttributesUpdate(TenantId tenantId, CustomerId customerId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) {
+ private ListenableFuture processAttributesUpdate(TenantId tenantId,
+ EntityId entityId,
+ TransportProtos.PostAttributeMsg msg,
+ TbMsgMetaData metaData) {
SettableFuture futureToSet = SettableFuture.create();
JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
- Set attributes = JsonConverter.convertToAttributes(json);
- ListenableFuture> future = attributesService.save(tenantId, entityId, metaData.getValue("scope"), new ArrayList<>(attributes));
- Futures.addCallback(future, new FutureCallback<>() {
+ List attributes = new ArrayList<>(JsonConverter.convertToAttributes(json));
+ String scope = metaData.getValue("scope");
+ tsSubService.saveAndNotify(tenantId, entityId, scope, attributes, new FutureCallback() {
@Override
- public void onSuccess(@Nullable List keys) {
- var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
- TbMsg tbMsg = TbMsg.newMsg(defaultQueueAndRuleChain.getKey(), DataConstants.ATTRIBUTES_UPDATED, entityId, customerId, metaData, gson.toJson(json), defaultQueueAndRuleChain.getValue(), null);
- tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
- @Override
- public void onSuccess(TbQueueMsgMetadata metadata) {
- futureToSet.set(null);
- }
-
- @Override
- public void onFailure(Throwable t) {
- log.error("Can't process attributes update [{}]", msg, t);
- futureToSet.setException(t);
- }
- });
+ public void onSuccess(@Nullable Void tmp) {
+ logAttributesUpdated(tenantId, entityId, scope, attributes, null);
+ futureToSet.set(null);
}
@Override
public void onFailure(Throwable t) {
log.error("Can't process attributes update [{}]", msg, t);
+ logAttributesUpdated(tenantId, entityId, scope, attributes, t);
futureToSet.setException(t);
}
- }, dbCallbackExecutorService);
+ });
return futureToSet;
}
+ private void logAttributesUpdated(TenantId tenantId, EntityId entityId, String scope, List attributes, Throwable e) {
+ notificationEntityService.logEntityAction(tenantId, entityId, ActionType.ATTRIBUTES_UPDATED, null,
+ BaseController.toException(e), scope, attributes);
+ }
+
private ListenableFuture processAttributeDeleteMsg(TenantId tenantId, EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, String entityType) {
SettableFuture futureToSet = SettableFuture.create();
String scope = attributeDeleteMsg.getScope();
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java
index e66d7d5595..8db5c4f4c7 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java
@@ -33,10 +33,13 @@ import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.EntityView;
+import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.edge.EdgeEventType;
+import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.EdgeId;
@@ -57,6 +60,8 @@ import org.thingsboard.server.common.data.relation.RelationTypeGroup;
import org.thingsboard.server.common.data.relation.RelationsSearchParameters;
import org.thingsboard.server.common.data.widget.WidgetType;
import org.thingsboard.server.common.data.widget.WidgetsBundle;
+import org.thingsboard.server.dao.asset.AssetProfileService;
+import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.device.DeviceProfileService;
import org.thingsboard.server.dao.device.DeviceService;
@@ -64,6 +69,7 @@ import org.thingsboard.server.dao.edge.EdgeEventService;
import org.thingsboard.server.dao.relation.RelationService;
import org.thingsboard.server.dao.widget.WidgetTypeService;
import org.thingsboard.server.dao.widget.WidgetsBundleService;
+import org.thingsboard.server.gen.edge.v1.AssetProfileAssetsRequestMsg;
import org.thingsboard.server.gen.edge.v1.AttributesRequestMsg;
import org.thingsboard.server.gen.edge.v1.DeviceCredentialsRequestMsg;
import org.thingsboard.server.gen.edge.v1.DeviceProfileDevicesRequestMsg;
@@ -104,6 +110,9 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
@Autowired
private DeviceService deviceService;
+ @Autowired
+ private AssetService assetService;
+
@Lazy
@Autowired
private TbEntityViewService entityViewService;
@@ -111,6 +120,9 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
@Autowired
private DeviceProfileService deviceProfileService;
+ @Autowired
+ private AssetProfileService assetProfileService;
+
@Autowired
private WidgetsBundleService widgetsBundleService;
@@ -351,6 +363,44 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
}
+ @Override
+ public ListenableFuture processAssetProfileAssetsRequestMsg(TenantId tenantId, Edge edge, AssetProfileAssetsRequestMsg assetProfileAssetsRequestMsg) {
+ log.trace("[{}] processAssetProfileAssetsRequestMsg [{}][{}]", tenantId, edge.getName(), assetProfileAssetsRequestMsg);
+ if (assetProfileAssetsRequestMsg.getAssetProfileIdMSB() == 0 || assetProfileAssetsRequestMsg.getAssetProfileIdLSB() == 0) {
+ return Futures.immediateFuture(null);
+ }
+ AssetProfileId assetProfileId = new AssetProfileId(new UUID(assetProfileAssetsRequestMsg.getAssetProfileIdMSB(), assetProfileAssetsRequestMsg.getAssetProfileIdLSB()));
+ AssetProfile assetProfileById = assetProfileService.findAssetProfileById(tenantId, assetProfileId);
+ if (assetProfileById == null) {
+ return Futures.immediateFuture(null);
+ }
+ return syncAssets(tenantId, edge, assetProfileById.getName());
+ }
+
+ private ListenableFuture syncAssets(TenantId tenantId, Edge edge, String assetType) {
+ log.trace("[{}] syncAssets [{}][{}]", tenantId, edge.getName(), assetType);
+ List> futures = new ArrayList<>();
+ try {
+ PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
+ PageData pageData;
+ do {
+ pageData = assetService.findAssetsByTenantIdAndEdgeIdAndType(tenantId, edge.getId(), assetType, pageLink);
+ if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
+ log.trace("[{}] [{}] asset(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
+ for (Asset asset : pageData.getData()) {
+ futures.add(saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ASSET, EdgeEventActionType.ADDED, asset.getId(), null));
+ }
+ if (pageData.hasNext()) {
+ pageLink = pageLink.nextPageLink();
+ }
+ }
+ } while (pageData != null && pageData.hasNext());
+ } catch (Exception e) {
+ log.error("Exception during loading edge asset(s) on sync!", e);
+ }
+ return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
+ }
+
@Override
public ListenableFuture processWidgetBundleTypesRequestMsg(TenantId tenantId, Edge edge,
WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg) {
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java
index 2ba1b3c9d3..7140ac52a5 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java
@@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.sync;
import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.gen.edge.v1.AssetProfileAssetsRequestMsg;
import org.thingsboard.server.gen.edge.v1.AttributesRequestMsg;
import org.thingsboard.server.gen.edge.v1.DeviceCredentialsRequestMsg;
import org.thingsboard.server.gen.edge.v1.DeviceProfileDevicesRequestMsg;
@@ -41,6 +42,8 @@ public interface EdgeRequestsService {
ListenableFuture processDeviceProfileDevicesRequestMsg(TenantId tenantId, Edge edge, DeviceProfileDevicesRequestMsg deviceProfileDevicesRequestMsg);
+ ListenableFuture processAssetProfileAssetsRequestMsg(TenantId tenantId, Edge edge, AssetProfileAssetsRequestMsg assetProfileAssetsRequestMsg);
+
ListenableFuture processWidgetBundleTypesRequestMsg(TenantId tenantId, Edge edge, WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg);
ListenableFuture processEntityViewsRequestMsg(TenantId tenantId, Edge edge, EntityViewsRequestMsg entityViewsRequestMsg);
diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/asset/profile/DefaultTbAssetProfileService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/asset/profile/DefaultTbAssetProfileService.java
new file mode 100644
index 0000000000..d053b99a3b
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/entitiy/asset/profile/DefaultTbAssetProfileService.java
@@ -0,0 +1,110 @@
+/**
+ * Copyright © 2016-2022 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.service.entitiy.asset.profile;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+import org.thingsboard.server.common.data.audit.ActionType;
+import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+import org.thingsboard.server.common.data.id.AssetProfileId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
+import org.thingsboard.server.dao.asset.AssetProfileService;
+import org.thingsboard.server.queue.util.TbCoreComponent;
+import org.thingsboard.server.service.entitiy.AbstractTbEntityService;
+
+import static org.thingsboard.server.dao.asset.BaseAssetService.TB_SERVICE_QUEUE;
+
+@Service
+@TbCoreComponent
+@AllArgsConstructor
+@Slf4j
+public class DefaultTbAssetProfileService extends AbstractTbEntityService implements TbAssetProfileService {
+
+ private final AssetProfileService assetProfileService;
+
+ @Override
+ public AssetProfile save(AssetProfile assetProfile, User user) throws Exception {
+ ActionType actionType = assetProfile.getId() == null ? ActionType.ADDED : ActionType.UPDATED;
+ TenantId tenantId = assetProfile.getTenantId();
+ try {
+ if (TB_SERVICE_QUEUE.equals(assetProfile.getName())) {
+ throw new ThingsboardException("Unable to save asset profile with name " + TB_SERVICE_QUEUE, ThingsboardErrorCode.BAD_REQUEST_PARAMS);
+ } else if (assetProfile.getId() != null) {
+ AssetProfile foundAssetProfile = assetProfileService.findAssetProfileById(tenantId, assetProfile.getId());
+ if (foundAssetProfile != null && TB_SERVICE_QUEUE.equals(foundAssetProfile.getName())) {
+ throw new ThingsboardException("Updating asset profile with name " + TB_SERVICE_QUEUE + " is prohibited!", ThingsboardErrorCode.BAD_REQUEST_PARAMS);
+ }
+ }
+ AssetProfile savedAssetProfile = checkNotNull(assetProfileService.saveAssetProfile(assetProfile));
+ autoCommit(user, savedAssetProfile.getId());
+ tbClusterService.broadcastEntityStateChangeEvent(tenantId, savedAssetProfile.getId(),
+ actionType.equals(ActionType.ADDED) ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED);
+
+ notificationEntityService.notifyCreateOrUpdateOrDelete(tenantId, null, savedAssetProfile.getId(),
+ savedAssetProfile, user, actionType, true, null);
+ return savedAssetProfile;
+ } catch (Exception e) {
+ notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.ASSET_PROFILE), assetProfile, actionType, user, e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void delete(AssetProfile assetProfile, User user) {
+ AssetProfileId assetProfileId = assetProfile.getId();
+ TenantId tenantId = assetProfile.getTenantId();
+ try {
+ assetProfileService.deleteAssetProfile(tenantId, assetProfileId);
+
+ tbClusterService.broadcastEntityStateChangeEvent(tenantId, assetProfileId, ComponentLifecycleEvent.DELETED);
+ notificationEntityService.notifyCreateOrUpdateOrDelete(tenantId, null, assetProfileId, assetProfile,
+ user, ActionType.DELETED, true, null, assetProfileId.toString());
+ } catch (Exception e) {
+ notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.ASSET_PROFILE), ActionType.DELETED,
+ user, e, assetProfileId.toString());
+ throw e;
+ }
+ }
+
+ @Override
+ public AssetProfile setDefaultAssetProfile(AssetProfile assetProfile, AssetProfile previousDefaultAssetProfile, User user) throws ThingsboardException {
+ TenantId tenantId = assetProfile.getTenantId();
+ AssetProfileId assetProfileId = assetProfile.getId();
+ try {
+ if (assetProfileService.setDefaultAssetProfile(tenantId, assetProfileId)) {
+ if (previousDefaultAssetProfile != null) {
+ previousDefaultAssetProfile = assetProfileService.findAssetProfileById(tenantId, previousDefaultAssetProfile.getId());
+ notificationEntityService.logEntityAction(tenantId, previousDefaultAssetProfile.getId(), previousDefaultAssetProfile,
+ ActionType.UPDATED, user);
+ }
+ assetProfile = assetProfileService.findAssetProfileById(tenantId, assetProfileId);
+
+ notificationEntityService.logEntityAction(tenantId, assetProfileId, assetProfile, ActionType.UPDATED, user);
+ }
+ return assetProfile;
+ } catch (Exception e) {
+ notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.ASSET_PROFILE), ActionType.UPDATED,
+ user, e, assetProfileId.toString());
+ throw e;
+ }
+ }
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/asset/profile/TbAssetProfileService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/asset/profile/TbAssetProfileService.java
new file mode 100644
index 0000000000..e451b3245a
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/entitiy/asset/profile/TbAssetProfileService.java
@@ -0,0 +1,26 @@
+/**
+ * Copyright © 2016-2022 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.service.entitiy.asset.profile;
+
+import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+import org.thingsboard.server.service.entitiy.SimpleTbEntityService;
+
+public interface TbAssetProfileService extends SimpleTbEntityService {
+
+ AssetProfile setDefaultAssetProfile(AssetProfile assetProfile, AssetProfile previousDefaultAssetProfile, User user) throws ThingsboardException;
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
index ea976b2dec..092726d0c8 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
@@ -15,8 +15,6 @@
*/
package org.thingsboard.server.service.install;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
@@ -25,10 +23,7 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.EntitySubtype;
-import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.Tenant;
-import org.thingsboard.server.common.data.TenantProfile;
-import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
@@ -37,8 +32,8 @@ import org.thingsboard.server.common.data.queue.ProcessingStrategyType;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.data.queue.SubmitStrategy;
import org.thingsboard.server.common.data.queue.SubmitStrategyType;
-import org.thingsboard.server.common.data.rule.RuleNode;
-import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration;
+import org.thingsboard.server.dao.asset.AssetProfileService;
+import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.dashboard.DashboardService;
import org.thingsboard.server.dao.device.DeviceProfileService;
import org.thingsboard.server.dao.device.DeviceService;
@@ -61,11 +56,8 @@ import java.sql.SQLException;
import java.sql.SQLSyntaxErrorException;
import java.sql.SQLWarning;
import java.sql.Statement;
-import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
import static org.thingsboard.server.service.install.DatabaseHelper.ADDITIONAL_INFO;
import static org.thingsboard.server.service.install.DatabaseHelper.ASSIGNED_CUSTOMERS;
@@ -117,9 +109,15 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
@Autowired
private DeviceService deviceService;
+ @Autowired
+ private AssetService assetService;
+
@Autowired
private DeviceProfileService deviceProfileService;
+ @Autowired
+ private AssetProfileService assetProfileService;
+
@Autowired
private ApiUsageStateService apiUsageStateService;
@@ -609,11 +607,63 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
log.error("Failed updating schema!!!", e);
}
break;
+ case "3.4.1":
+ try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
+ log.info("Updating schema ...");
+ runSchemaUpdateScript(conn, "3.4.1");
+ if (isOldSchema(conn, 3004001)) {
+ try {
+ conn.createStatement().execute("ALTER TABLE asset ADD COLUMN asset_profile_id uuid");
+ } catch (Exception e) {
+ }
+
+ schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "3.4.1", "schema_update_before.sql");
+ loadSql(schemaUpdateFile, conn);
+
+ log.info("Creating default asset profiles...");
+ PageLink pageLink = new PageLink(100);
+ PageData pageData;
+ do {
+ pageData = tenantService.findTenants(pageLink);
+ for (Tenant tenant : pageData.getData()) {
+ List assetTypes = assetService.findAssetTypesByTenantId(tenant.getId()).get();
+ try {
+ assetProfileService.createDefaultAssetProfile(tenant.getId());
+ } catch (Exception e) {
+ }
+ for (EntitySubtype assetType : assetTypes) {
+ try {
+ assetProfileService.findOrCreateAssetProfile(tenant.getId(), assetType.getType());
+ } catch (Exception e) {
+ }
+ }
+ }
+ pageLink = pageLink.nextPageLink();
+ } while (pageData.hasNext());
+
+ log.info("Updating asset profiles...");
+ conn.createStatement().execute("call update_asset_profiles()");
+
+ schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "3.4.1", "schema_update_after.sql");
+ loadSql(schemaUpdateFile, conn);
+
+ conn.createStatement().execute("UPDATE tb_schema_settings SET schema_version = 3004002;");
+ }
+ log.info("Schema updated.");
+ } catch (Exception e) {
+ log.error("Failed updating schema!!!", e);
+ }
+ break;
default:
throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion);
}
}
+ private void runSchemaUpdateScript(Connection connection, String version) throws Exception {
+ Path schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", version, SCHEMA_UPDATE_SQL);
+ loadSql(schemaUpdateFile, connection);
+ }
+
private void loadSql(Path sqlFile, Connection conn) throws Exception {
String sql = new String(Files.readAllBytes(sqlFile), Charset.forName("UTF-8"));
Statement st = conn.createStatement();
diff --git a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java
index d02f50b37b..a8cf5374a2 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java
@@ -63,6 +63,7 @@ import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration;
import org.thingsboard.server.dao.DaoUtil;
import org.thingsboard.server.dao.alarm.AlarmDao;
+import org.thingsboard.server.dao.audit.AuditLogDao;
import org.thingsboard.server.dao.entity.EntityService;
import org.thingsboard.server.dao.entityview.EntityViewService;
import org.thingsboard.server.dao.event.EventService;
@@ -138,6 +139,9 @@ public class DefaultDataUpdateService implements DataUpdateService {
@Autowired
private EventService eventService;
+ @Autowired
+ private AuditLogDao auditLogDao;
+
@Override
public void updateData(String fromVersion) throws Exception {
switch (fromVersion) {
@@ -170,12 +174,22 @@ public class DefaultDataUpdateService implements DataUpdateService {
rateLimitsUpdater.updateEntities();
break;
case "3.4.0":
- String skipEventsMigration = System.getenv("TB_SKIP_EVENTS_MIGRATION");
- if (skipEventsMigration == null || skipEventsMigration.equalsIgnoreCase("false")) {
+ boolean skipEventsMigration = getEnv("TB_SKIP_EVENTS_MIGRATION", false);
+ if (!skipEventsMigration) {
log.info("Updating data from version 3.4.0 to 3.4.1 ...");
eventService.migrateEvents();
}
break;
+ case "3.4.1":
+ boolean skipAuditLogsMigration = getEnv("TB_SKIP_AUDIT_LOGS_MIGRATION", false);
+ if (!skipAuditLogsMigration) {
+ log.info("Updating data from version 3.4.1 to 3.4.2 ...");
+ log.info("Starting audit logs migration. Can be skipped with TB_SKIP_AUDIT_LOGS_MIGRATION env variable set to true");
+ auditLogDao.migrateAuditLogs();
+ } else {
+ log.info("Skipping audit logs migration");
+ }
+ break;
default:
throw new RuntimeException("Unable to update data, unsupported fromVersion: " + fromVersion);
}
@@ -645,4 +659,13 @@ public class DefaultDataUpdateService implements DataUpdateService {
return mainQueueConfiguration;
}
+ private boolean getEnv(String name, boolean defaultValue) {
+ String env = System.getenv(name);
+ if (env == null) {
+ return defaultValue;
+ } else {
+ return Boolean.parseBoolean(env);
+ }
+ }
+
}
diff --git a/application/src/main/java/org/thingsboard/server/service/profile/DefaultTbAssetProfileCache.java b/application/src/main/java/org/thingsboard/server/service/profile/DefaultTbAssetProfileCache.java
new file mode 100644
index 0000000000..23d0e1b616
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/profile/DefaultTbAssetProfileCache.java
@@ -0,0 +1,162 @@
+/**
+ * Copyright © 2016-2022 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.service.profile;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+import org.thingsboard.server.common.data.id.AssetId;
+import org.thingsboard.server.common.data.id.AssetProfileId;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.dao.asset.AssetProfileService;
+import org.thingsboard.server.dao.asset.AssetService;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+@Service
+@Slf4j
+public class DefaultTbAssetProfileCache implements TbAssetProfileCache {
+
+ private final Lock assetProfileFetchLock = new ReentrantLock();
+ private final AssetProfileService assetProfileService;
+ private final AssetService assetService;
+
+ private final ConcurrentMap assetProfilesMap = new ConcurrentHashMap<>();
+ private final ConcurrentMap assetsMap = new ConcurrentHashMap<>();
+ private final ConcurrentMap>> profileListeners = new ConcurrentHashMap<>();
+ private final ConcurrentMap>> assetProfileListeners = new ConcurrentHashMap<>();
+
+ public DefaultTbAssetProfileCache(AssetProfileService assetProfileService, AssetService assetService) {
+ this.assetProfileService = assetProfileService;
+ this.assetService = assetService;
+ }
+
+ @Override
+ public AssetProfile get(TenantId tenantId, AssetProfileId assetProfileId) {
+ AssetProfile profile = assetProfilesMap.get(assetProfileId);
+ if (profile == null) {
+ assetProfileFetchLock.lock();
+ try {
+ profile = assetProfilesMap.get(assetProfileId);
+ if (profile == null) {
+ profile = assetProfileService.findAssetProfileById(tenantId, assetProfileId);
+ if (profile != null) {
+ assetProfilesMap.put(assetProfileId, profile);
+ log.debug("[{}] Fetch asset profile into cache: {}", profile.getId(), profile);
+ }
+ }
+ } finally {
+ assetProfileFetchLock.unlock();
+ }
+ }
+ log.trace("[{}] Found asset profile in cache: {}", assetProfileId, profile);
+ return profile;
+ }
+
+ @Override
+ public AssetProfile get(TenantId tenantId, AssetId assetId) {
+ AssetProfileId profileId = assetsMap.get(assetId);
+ if (profileId == null) {
+ Asset asset = assetService.findAssetById(tenantId, assetId);
+ if (asset != null) {
+ profileId = asset.getAssetProfileId();
+ assetsMap.put(assetId, profileId);
+ } else {
+ return null;
+ }
+ }
+ return get(tenantId, profileId);
+ }
+
+ @Override
+ public void evict(TenantId tenantId, AssetProfileId profileId) {
+ AssetProfile oldProfile = assetProfilesMap.remove(profileId);
+ log.debug("[{}] evict asset profile from cache: {}", profileId, oldProfile);
+ AssetProfile newProfile = get(tenantId, profileId);
+ if (newProfile != null) {
+ notifyProfileListeners(newProfile);
+ }
+ }
+
+ @Override
+ public void evict(TenantId tenantId, AssetId assetId) {
+ AssetProfileId old = assetsMap.remove(assetId);
+ if (old != null) {
+ AssetProfile newProfile = get(tenantId, assetId);
+ if (newProfile == null || !old.equals(newProfile.getId())) {
+ notifyAssetListeners(tenantId, assetId, newProfile);
+ }
+ }
+ }
+
+ @Override
+ public void addListener(TenantId tenantId, EntityId listenerId,
+ Consumer profileListener,
+ BiConsumer assetListener) {
+ if (profileListener != null) {
+ profileListeners.computeIfAbsent(tenantId, id -> new ConcurrentHashMap<>()).put(listenerId, profileListener);
+ }
+ if (assetListener != null) {
+ assetProfileListeners.computeIfAbsent(tenantId, id -> new ConcurrentHashMap<>()).put(listenerId, assetListener);
+ }
+ }
+
+ @Override
+ public AssetProfile find(AssetProfileId assetProfileId) {
+ return assetProfileService.findAssetProfileById(TenantId.SYS_TENANT_ID, assetProfileId);
+ }
+
+ @Override
+ public AssetProfile findOrCreateAssetProfile(TenantId tenantId, String profileName) {
+ return assetProfileService.findOrCreateAssetProfile(tenantId, profileName);
+ }
+
+ @Override
+ public void removeListener(TenantId tenantId, EntityId listenerId) {
+ ConcurrentMap> tenantListeners = profileListeners.get(tenantId);
+ if (tenantListeners != null) {
+ tenantListeners.remove(listenerId);
+ }
+ ConcurrentMap> assetListeners = assetProfileListeners.get(tenantId);
+ if (assetListeners != null) {
+ assetListeners.remove(listenerId);
+ }
+ }
+
+ private void notifyProfileListeners(AssetProfile profile) {
+ ConcurrentMap> tenantListeners = profileListeners.get(profile.getTenantId());
+ if (tenantListeners != null) {
+ tenantListeners.forEach((id, listener) -> listener.accept(profile));
+ }
+ }
+
+ private void notifyAssetListeners(TenantId tenantId, AssetId assetId, AssetProfile profile) {
+ if (profile != null) {
+ ConcurrentMap> tenantListeners = assetProfileListeners.get(tenantId);
+ if (tenantListeners != null) {
+ tenantListeners.forEach((id, listener) -> listener.accept(assetId, profile));
+ }
+ }
+ }
+
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/profile/TbAssetProfileCache.java b/application/src/main/java/org/thingsboard/server/service/profile/TbAssetProfileCache.java
new file mode 100644
index 0000000000..6c332e63ff
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/profile/TbAssetProfileCache.java
@@ -0,0 +1,33 @@
+/**
+ * Copyright © 2016-2022 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.service.profile;
+
+import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache;
+import org.thingsboard.server.common.data.asset.AssetProfile;
+import org.thingsboard.server.common.data.id.AssetId;
+import org.thingsboard.server.common.data.id.AssetProfileId;
+import org.thingsboard.server.common.data.id.TenantId;
+
+public interface TbAssetProfileCache extends RuleEngineAssetProfileCache {
+
+ void evict(TenantId tenantId, AssetProfileId id);
+
+ void evict(TenantId tenantId, AssetId id);
+
+ AssetProfile find(AssetProfileId assetProfileId);
+
+ AssetProfile findOrCreateAssetProfile(TenantId tenantId, String assetType);
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
index 062e4128d4..7983a0adcd 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
@@ -35,13 +35,15 @@ import org.thingsboard.server.common.data.HasName;
import org.thingsboard.server.common.data.TbResource;
import org.thingsboard.server.common.data.Tenant;
import org.thingsboard.server.common.data.TenantProfile;
+import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.edge.EdgeEventType;
+import org.thingsboard.server.common.data.id.AssetId;
+import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.EntityId;
-import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
@@ -49,11 +51,12 @@ import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg;
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
+import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
+import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
-import org.thingsboard.server.queue.util.DataDecodingEncodingService;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCResponseProto;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
@@ -68,11 +71,12 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.NotificationsTopicService;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
+import org.thingsboard.server.queue.util.DataDecodingEncodingService;
import org.thingsboard.server.service.gateway_device.GatewayNotificationsService;
import org.thingsboard.server.service.ota.OtaPackageStateService;
+import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
-import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
@@ -110,6 +114,7 @@ public class DefaultTbClusterService implements TbClusterService {
private final NotificationsTopicService notificationsTopicService;
private final DataDecodingEncodingService encodingService;
private final TbDeviceProfileCache deviceProfileCache;
+ private final TbAssetProfileCache assetProfileCache;
private final GatewayNotificationsService gatewayNotificationsService;
@Override
@@ -167,7 +172,7 @@ public class DefaultTbClusterService implements TbClusterService {
@Override
public void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, TbMsg tbMsg, TbQueueCallback callback) {
- if (tenantId.isNullUid()) {
+ if (tenantId == null || tenantId.isNullUid()) {
if (entityId.getEntityType().equals(EntityType.TENANT)) {
tenantId = TenantId.fromUUID(entityId.getId());
} else {
@@ -179,6 +184,10 @@ public class DefaultTbClusterService implements TbClusterService {
tbMsg = transformMsg(tbMsg, deviceProfileCache.get(tenantId, new DeviceId(entityId.getId())));
} else if (entityId.getEntityType().equals(EntityType.DEVICE_PROFILE)) {
tbMsg = transformMsg(tbMsg, deviceProfileCache.get(tenantId, new DeviceProfileId(entityId.getId())));
+ } else if (entityId.getEntityType().equals(EntityType.ASSET)) {
+ tbMsg = transformMsg(tbMsg, assetProfileCache.get(tenantId, new AssetId(entityId.getId())));
+ } else if (entityId.getEntityType().equals(EntityType.ASSET_PROFILE)) {
+ tbMsg = transformMsg(tbMsg, assetProfileCache.get(tenantId, new AssetProfileId(entityId.getId())));
}
}
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, entityId);
@@ -195,16 +204,30 @@ public class DefaultTbClusterService implements TbClusterService {
if (deviceProfile != null) {
RuleChainId targetRuleChainId = deviceProfile.getDefaultRuleChainId();
String targetQueueName = deviceProfile.getDefaultQueueName();
- boolean isRuleChainTransform = targetRuleChainId != null && !targetRuleChainId.equals(tbMsg.getRuleChainId());
- boolean isQueueTransform = targetQueueName != null && !targetQueueName.equals(tbMsg.getQueueName());
-
- if (isRuleChainTransform && isQueueTransform) {
- tbMsg = TbMsg.transformMsg(tbMsg, targetRuleChainId, targetQueueName);
- } else if (isRuleChainTransform) {
- tbMsg = TbMsg.transformMsg(tbMsg, targetRuleChainId);
- } else if (isQueueTransform) {
- tbMsg = TbMsg.transformMsg(tbMsg, targetQueueName);
- }
+ tbMsg = transformMsg(tbMsg, targetRuleChainId, targetQueueName);
+ }
+ return tbMsg;
+ }
+
+ private TbMsg transformMsg(TbMsg tbMsg, AssetProfile assetProfile) {
+ if (assetProfile != null) {
+ RuleChainId targetRuleChainId = assetProfile.getDefaultRuleChainId();
+ String targetQueueName = assetProfile.getDefaultQueueName();
+ tbMsg = transformMsg(tbMsg, targetRuleChainId, targetQueueName);
+ }
+ return tbMsg;
+ }
+
+ private TbMsg transformMsg(TbMsg tbMsg, RuleChainId targetRuleChainId, String targetQueueName) {
+ boolean isRuleChainTransform = targetRuleChainId != null && !targetRuleChainId.equals(tbMsg.getRuleChainId());
+ boolean isQueueTransform = targetQueueName != null && !targetQueueName.equals(tbMsg.getQueueName());
+
+ if (isRuleChainTransform && isQueueTransform) {
+ tbMsg = TbMsg.transformMsg(tbMsg, targetRuleChainId, targetQueueName);
+ } else if (isRuleChainTransform) {
+ tbMsg = TbMsg.transformMsg(tbMsg, targetRuleChainId);
+ } else if (isQueueTransform) {
+ tbMsg = TbMsg.transformMsg(tbMsg, targetQueueName);
}
return tbMsg;
}
@@ -351,12 +374,32 @@ public class DefaultTbClusterService implements TbClusterService {
log.trace("[{}] Processing edge {} event update ", tenantId, edgeId);
EdgeEventUpdateMsg msg = new EdgeEventUpdateMsg(tenantId, edgeId);
byte[] msgBytes = encodingService.encode(msg);
+ ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setEdgeEventUpdateMsg(ByteString.copyFrom(msgBytes)).build();
+ pushEdgeSyncMsgToCore(edgeId, toCoreMsg);
+ }
+
+ @Override
+ public void pushEdgeSyncRequestToCore(ToEdgeSyncRequest toEdgeSyncRequest) {
+ log.trace("[{}] Processing edge sync request {} ", toEdgeSyncRequest.getTenantId(), toEdgeSyncRequest);
+ byte[] msgBytes = encodingService.encode(toEdgeSyncRequest);
+ ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setToEdgeSyncRequestMsg(ByteString.copyFrom(msgBytes)).build();
+ pushEdgeSyncMsgToCore(toEdgeSyncRequest.getEdgeId(), toCoreMsg);
+ }
+
+ @Override
+ public void pushEdgeSyncResponseToCore(FromEdgeSyncResponse fromEdgeSyncResponse) {
+ log.trace("[{}] Processing edge sync response {}", fromEdgeSyncResponse.getTenantId(), fromEdgeSyncResponse);
+ byte[] msgBytes = encodingService.encode(fromEdgeSyncResponse);
+ ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setFromEdgeSyncResponseMsg(ByteString.copyFrom(msgBytes)).build();
+ pushEdgeSyncMsgToCore(fromEdgeSyncResponse.getEdgeId(), toCoreMsg);
+ }
+
+ private void pushEdgeSyncMsgToCore(EdgeId edgeId, ToCoreNotificationMsg toCoreMsg) {
TbQueueProducer> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer();
Set tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE);
for (String serviceId : tbCoreServices) {
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId);
- ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setEdgeEventUpdateMsg(ByteString.copyFrom(msgBytes)).build();
- toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEdgeId().getId(), toCoreMsg), null);
+ toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(edgeId.getId(), toCoreMsg), null);
toCoreNfs.incrementAndGet();
}
}
@@ -369,6 +412,7 @@ public class DefaultTbClusterService implements TbClusterService {
if (entityType.equals(EntityType.TENANT)
|| entityType.equals(EntityType.TENANT_PROFILE)
|| entityType.equals(EntityType.DEVICE_PROFILE)
+ || entityType.equals(EntityType.ASSET_PROFILE)
|| entityType.equals(EntityType.API_USAGE_STATE)
|| (entityType.equals(EntityType.DEVICE) && msg.getEvent() == ComponentLifecycleEvent.UPDATED)
|| entityType.equals(EntityType.ENTITY_VIEW)
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
index a3bae8bc4e..ec58ac1aa9 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
@@ -20,8 +20,6 @@ import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
-import org.springframework.context.event.EventListener;
-import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
@@ -67,6 +65,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
import org.thingsboard.server.service.edge.EdgeNotificationService;
import org.thingsboard.server.service.ota.OtaPackageStateService;
+import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.service.queue.processing.AbstractConsumerService;
import org.thingsboard.server.service.queue.processing.IdMsgPair;
@@ -76,7 +75,6 @@ import org.thingsboard.server.service.state.DeviceStateService;
import org.thingsboard.server.service.subscription.SubscriptionManagerService;
import org.thingsboard.server.service.subscription.TbLocalSubscriptionService;
import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
-import org.thingsboard.server.service.sync.vc.EntitiesVersionControlService;
import org.thingsboard.server.service.sync.vc.GitVersionControlQueueService;
import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
@@ -138,6 +136,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService actorMsg = encodingService.decode(toCoreNotification.getEdgeEventUpdateMsg().toByteArray());
- if (actorMsg.isPresent()) {
- log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
- actorContext.tellWithHighPriority(actorMsg.get());
- }
- callback.onSuccess();
+ } else if (!toCoreNotification.getEdgeEventUpdateMsg().isEmpty()) {
+ forwardToAppActor(id, encodingService.decode(toCoreNotification.getEdgeEventUpdateMsg().toByteArray()), callback);
+ } else if (!toCoreNotification.getToEdgeSyncRequestMsg().isEmpty()) {
+ forwardToAppActor(id, encodingService.decode(toCoreNotification.getToEdgeSyncRequestMsg().toByteArray()), callback);
+ } else if (!toCoreNotification.getFromEdgeSyncResponseMsg().isEmpty()) {
+ forwardToAppActor(id, encodingService.decode(toCoreNotification.getFromEdgeSyncResponseMsg().toByteArray()), callback);
} else if (toCoreNotification.hasQueueUpdateMsg()) {
TransportProtos.QueueUpdateMsg queue = toCoreNotification.getQueueUpdateMsg();
partitionService.updateQueue(queue);
@@ -552,6 +550,14 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService actorMsg, TbCallback callback) {
+ if (actorMsg.isPresent()) {
+ log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
+ actorContext.tell(actorMsg.get());
+ }
+ callback.onSuccess();
+ }
+
private void throwNotHandled(Object msg, TbCallback callback) {
log.warn("Message not handled: {}", msg);
callback.onFailure(new RuntimeException("Message not handled!"));
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
index 3c7552ff9b..d870af318e 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
@@ -51,6 +51,7 @@ import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory;
import org.thingsboard.server.queue.util.TbRuleEngineComponent;
import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
+import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.service.queue.processing.AbstractConsumerService;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision;
@@ -121,10 +122,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
TbRuleEngineDeviceRpcService tbDeviceRpcService,
StatsFactory statsFactory,
TbDeviceProfileCache deviceProfileCache,
+ TbAssetProfileCache assetProfileCache,
TbTenantProfileCache tenantProfileCache,
TbApiUsageStateService apiUsageStateService,
PartitionService partitionService, TbServiceInfoProvider serviceInfoProvider, QueueService queueService) {
- super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, apiUsageStateService, partitionService, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer());
+ super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer());
this.statisticsService = statisticsService;
this.tbRuleEngineQueueFactory = tbRuleEngineQueueFactory;
this.submitStrategyFactory = submitStrategyFactory;
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
index 98fb95ad15..47b7f3f9f9 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
@@ -18,11 +18,11 @@ package org.thingsboard.server.service.queue.processing;
import com.google.protobuf.ByteString;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.event.ApplicationReadyEvent;
-import org.springframework.context.event.EventListener;
-import org.springframework.core.annotation.Order;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.id.AssetId;
+import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
@@ -33,16 +33,17 @@ import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionService;
-import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
-import org.thingsboard.server.queue.util.DataDecodingEncodingService;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
+import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
import org.thingsboard.server.queue.util.AfterStartUp;
+import org.thingsboard.server.queue.util.DataDecodingEncodingService;
import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
+import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
-import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.service.queue.TbPackCallback;
import org.thingsboard.server.service.queue.TbPackProcessingContext;
@@ -70,6 +71,7 @@ public abstract class AbstractConsumerService> nfConsumer) {
+ TbAssetProfileCache assetProfileCache, TbApiUsageStateService apiUsageStateService,
+ PartitionService partitionService, TbQueueConsumer> nfConsumer) {
this.actorContext = actorContext;
this.encodingService = encodingService;
this.tenantProfileCache = tenantProfileCache;
this.deviceProfileCache = deviceProfileCache;
+ this.assetProfileCache = assetProfileCache;
this.apiUsageStateService = apiUsageStateService;
this.partitionService = partitionService;
this.nfConsumer = nfConsumer;
@@ -180,6 +183,10 @@ public abstract class AbstractConsumerService scriptIdToNameMap = new ConcurrentHashMap<>();
- protected Map disabledFunctions = new ConcurrentHashMap<>();
+
+ protected final Map> scriptIdToNameAndHashMap = new ConcurrentHashMap<>();
+ protected final Map disabledFunctions = new ConcurrentHashMap<>();
+
+ @Getter
+ @Value("${js.max_total_args_size:100000}")
+ private long maxTotalArgsSize;
+ @Getter
+ @Value("${js.max_result_size:300000}")
+ private long maxResultSize;
+ @Getter
+ @Value("${js.max_script_body_size:50000}")
+ private long maxScriptBodySize;
protected AbstractJsInvokeService(TbApiUsageStateService apiUsageStateService, TbApiUsageClient apiUsageClient) {
this.apiUsageStateService = apiUsageStateService;
@@ -65,44 +83,63 @@ public abstract class AbstractJsInvokeService implements JsInvokeService {
@Override
public ListenableFuture eval(TenantId tenantId, JsScriptType scriptType, String scriptBody, String... argNames) {
if (apiUsageStateService.getApiUsageState(tenantId).isJsExecEnabled()) {
+ if (scriptBodySizeExceeded(scriptBody)) {
+ return error(format("Script body exceeds maximum allowed size of %s symbols", getMaxScriptBodySize()));
+ }
UUID scriptId = UUID.randomUUID();
- String functionName = "invokeInternal_" + scriptId.toString().replace('-', '_');
+ String scriptHash = hash(tenantId, scriptBody);
+ String functionName = constructFunctionName(scriptId, scriptHash);
String jsScript = generateJsScript(scriptType, functionName, scriptBody, argNames);
- return doEval(scriptId, functionName, jsScript);
+ return doEval(scriptId, scriptHash, functionName, jsScript);
} else {
- return Futures.immediateFailedFuture(new RuntimeException("JS Execution is disabled due to API limits!"));
+ return error("JS Execution is disabled due to API limits!");
}
}
+ protected String constructFunctionName(UUID scriptId, String scriptHash) {
+ return "invokeInternal_" + scriptId.toString().replace('-', '_');
+ }
+
@Override
- public ListenableFuture