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 e7a54385a7..ca401596f6 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 @@ -38,6 +38,7 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; @@ -73,6 +74,7 @@ import org.thingsboard.server.service.script.RuleNodeJsScriptEngine; import java.util.Collections; import java.util.Set; +import java.util.function.BiConsumer; import java.util.function.Consumer; /** @@ -479,8 +481,8 @@ class DefaultTbContext implements TbContext { } @Override - public void addProfileListener(Consumer listener) { - mainCtx.getDeviceProfileCache().addListener(getTenantId(), getSelfId(), listener); + public void addDeviceProfileListeners(Consumer profileListener, BiConsumer deviceListener) { + mainCtx.getDeviceProfileCache().addListener(getTenantId(), getSelfId(), profileListener, deviceListener); } @Override diff --git a/application/src/main/java/org/thingsboard/server/controller/DeviceController.java b/application/src/main/java/org/thingsboard/server/controller/DeviceController.java index c8fde63c4d..6bb92648cf 100644 --- a/application/src/main/java/org/thingsboard/server/controller/DeviceController.java +++ b/application/src/main/java/org/thingsboard/server/controller/DeviceController.java @@ -51,6 +51,7 @@ import org.thingsboard.server.common.data.id.DeviceProfileId; 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.plugin.ComponentLifecycleEvent; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; @@ -119,6 +120,8 @@ public class DeviceController extends BaseController { tbClusterService.pushMsgToCore(new DeviceNameOrTypeUpdateMsg(savedDevice.getTenantId(), savedDevice.getId(), savedDevice.getName(), savedDevice.getType()), null); + tbClusterService.onEntityStateChange(device.getTenantId(), device.getId(), + device.getId() == null ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED); logEntityAction(savedDevice.getId(), savedDevice, savedDevice.getCustomerId(), diff --git a/application/src/main/java/org/thingsboard/server/service/profile/DefaultTbDeviceProfileCache.java b/application/src/main/java/org/thingsboard/server/service/profile/DefaultTbDeviceProfileCache.java index 6784405adb..20dcb47d1a 100644 --- a/application/src/main/java/org/thingsboard/server/service/profile/DefaultTbDeviceProfileCache.java +++ b/application/src/main/java/org/thingsboard/server/service/profile/DefaultTbDeviceProfileCache.java @@ -26,11 +26,11 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.dao.device.DeviceProfileService; import org.thingsboard.server.dao.device.DeviceService; -import java.util.Map; 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 @@ -43,7 +43,8 @@ public class DefaultTbDeviceProfileCache implements TbDeviceProfileCache { private final ConcurrentMap deviceProfilesMap = new ConcurrentHashMap<>(); private final ConcurrentMap devicesMap = new ConcurrentHashMap<>(); - private final ConcurrentMap>> listeners = new ConcurrentHashMap<>(); + private final ConcurrentMap>> profileListeners = new ConcurrentHashMap<>(); + private final ConcurrentMap>> deviceProfileListeners = new ConcurrentHashMap<>(); public DefaultTbDeviceProfileCache(DeviceProfileService deviceProfileService, DeviceService deviceService) { this.deviceProfileService = deviceProfileService; @@ -87,33 +88,37 @@ public class DefaultTbDeviceProfileCache implements TbDeviceProfileCache { return get(tenantId, profileId); } - @Override - public void put(DeviceProfile profile) { - if (profile.getId() != null) { - deviceProfilesMap.put(profile.getId(), profile); - log.debug("[{}] pushed device profile to cache: {}", profile.getId(), profile); - notifyListeners(profile); - } - } - @Override public void evict(TenantId tenantId, DeviceProfileId profileId) { DeviceProfile oldProfile = deviceProfilesMap.remove(profileId); log.debug("[{}] evict device profile from cache: {}", profileId, oldProfile); DeviceProfile newProfile = get(tenantId, profileId); if (newProfile != null) { - notifyListeners(newProfile); + notifyProfileListeners(newProfile); } } @Override - public void evict(DeviceId deviceId) { - devicesMap.remove(deviceId); + public void evict(TenantId tenantId, DeviceId deviceId) { + DeviceProfileId old = devicesMap.remove(deviceId); + if (old != null) { + DeviceProfile newProfile = get(tenantId, deviceId); + if (newProfile == null || !old.equals(newProfile.getId())) { + notifyDeviceListeners(deviceId, newProfile); + } + } } @Override - public void addListener(TenantId tenantId, EntityId listenerId, Consumer listener) { - listeners.computeIfAbsent(tenantId, id -> new ConcurrentHashMap<>()).put(listenerId, listener); + public void addListener(TenantId tenantId, EntityId listenerId, + Consumer profileListener, + BiConsumer deviceListener) { + if (profileListener != null) { + profileListeners.computeIfAbsent(tenantId, id -> new ConcurrentHashMap<>()).put(listenerId, profileListener); + } + if (deviceListener != null) { + deviceProfileListeners.computeIfAbsent(tenantId, id -> new ConcurrentHashMap<>()).put(listenerId, deviceListener); + } } @Override @@ -128,17 +133,28 @@ public class DefaultTbDeviceProfileCache implements TbDeviceProfileCache { @Override public void removeListener(TenantId tenantId, EntityId listenerId) { - ConcurrentMap> tenantListeners = listeners.get(tenantId); + ConcurrentMap> tenantListeners = profileListeners.get(tenantId); if (tenantListeners != null) { tenantListeners.remove(listenerId); } + ConcurrentMap> deviceListeners = deviceProfileListeners.get(tenantId); + if (deviceListeners != null) { + deviceListeners.remove(listenerId); + } } - private void notifyListeners(DeviceProfile profile) { - ConcurrentMap> tenantListeners = listeners.get(profile.getTenantId()); + private void notifyProfileListeners(DeviceProfile profile) { + ConcurrentMap> tenantListeners = profileListeners.get(profile.getTenantId()); if (tenantListeners != null) { tenantListeners.forEach((id, listener) -> listener.accept(profile)); } } + private void notifyDeviceListeners(DeviceId deviceId, DeviceProfile profile) { + ConcurrentMap> tenantListeners = deviceProfileListeners.get(profile.getTenantId()); + if (tenantListeners != null) { + tenantListeners.forEach((id, listener) -> listener.accept(deviceId, profile)); + } + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java b/application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java index e65f297d53..10e28bd299 100644 --- a/application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java +++ b/application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java @@ -23,11 +23,9 @@ import org.thingsboard.server.common.data.id.TenantId; public interface TbDeviceProfileCache extends RuleEngineDeviceProfileCache { - void put(DeviceProfile profile); - void evict(TenantId tenantId, DeviceProfileId id); - void evict(DeviceId id); + void evict(TenantId tenantId, DeviceId id); DeviceProfile find(DeviceProfileId deviceProfileId); 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 9a4d488e0f..80bb6acd1d 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 @@ -165,7 +165,7 @@ public abstract class AbstractConsumerService listener); + void addListener(TenantId tenantId, EntityId listenerId, Consumer profileListener, BiConsumer devicelistener); void removeListener(TenantId tenantId, EntityId listenerId); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 3d45fcf445..24af4f0d22 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; @@ -47,6 +48,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; import java.util.Set; +import java.util.function.BiConsumer; import java.util.function.Consumer; /** @@ -225,7 +227,7 @@ public interface TbContext { void clearRuleNodeStates(); - void addProfileListener(Consumer listener); + void addDeviceProfileListeners(Consumer listener, BiConsumer deviceListener); void removeProfileListener(); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java index df46a4bf88..6c771f0c86 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java @@ -103,6 +103,7 @@ class AlarmState { } AlarmEvalResult evalResult = evalFunction.apply(clearState, data); if (AlarmEvalResult.TRUE.equals(evalResult)) { + clearState.clear(); stateUpdate |= clearState.checkUpdate(); for (AlarmRuleState state : createRulesSortedBySeverityDesc) { state.clear(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java index b4bbf08323..0766265368 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java @@ -15,6 +15,8 @@ */ package org.thingsboard.rule.engine.profile; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.extern.slf4j.Slf4j; import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; import org.thingsboard.rule.engine.api.RuleNode; @@ -59,6 +61,7 @@ import java.util.concurrent.TimeUnit; public class TbDeviceProfileNode implements TbNode { private static final String PERIODIC_MSG_TYPE = "TbDeviceProfilePeriodicMsg"; private static final String PROFILE_UPDATE_MSG_TYPE = "TbDeviceProfileUpdateMsg"; + private static final String DEVICE_UPDATE_MSG_TYPE = "TbDeviceUpdateMsg"; private TbDeviceProfileNodeConfiguration config; private RuleEngineDeviceProfileCache cache; @@ -71,7 +74,7 @@ public class TbDeviceProfileNode implements TbNode { this.cache = ctx.getDeviceProfileCache(); this.ctx = ctx; scheduleAlarmHarvesting(ctx); - ctx.addProfileListener(this::onProfileUpdate); + ctx.addDeviceProfileListeners(this::onProfileUpdate, this::onDeviceUpdate); if (config.isFetchAlarmRulesStateOnStart()) { log.info("[{}] Fetching alarm rule state", ctx.getSelfId()); int fetchCount = 0; @@ -108,6 +111,15 @@ public class TbDeviceProfileNode implements TbNode { harvestAlarms(ctx, System.currentTimeMillis()); } else if (msg.getType().equals(PROFILE_UPDATE_MSG_TYPE)) { updateProfile(ctx, new DeviceProfileId(UUID.fromString(msg.getData()))); + } else if (msg.getType().equals(DEVICE_UPDATE_MSG_TYPE)) { + JsonNode data = JacksonUtil.toJsonNode(msg.getData()); + DeviceId deviceId = new DeviceId(UUID.fromString(data.get("deviceId").asText())); + if (data.has("profileId")) { + invalidateDeviceProfileCache(deviceId, new DeviceProfileId(UUID.fromString(data.get("deviceProfileId").asText()))); + } else { + deviceStates.remove(deviceId); + } + } else { if (EntityType.DEVICE.equals(originatorType)) { DeviceId deviceId = new DeviceId(msg.getOriginator().getId()); @@ -182,6 +194,15 @@ public class TbDeviceProfileNode implements TbNode { ctx.tellSelf(TbMsg.newMsg(PROFILE_UPDATE_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, profile.getId().getId().toString()), 0L); } + private void onDeviceUpdate(DeviceId deviceId, DeviceProfile deviceProfile) { + ObjectNode msgData = JacksonUtil.newObjectNode(); + msgData.put("deviceId", deviceId.getId().toString()); + if (deviceProfile != null) { + msgData.put("deviceProfileId", deviceProfile.getId().getId().toString()); + } + ctx.tellSelf(TbMsg.newMsg(DEVICE_UPDATE_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, JacksonUtil.toString(msgData)), 0L); + } + protected void invalidateDeviceProfileCache(DeviceId deviceId, String deviceJson) { DeviceState deviceState = deviceStates.get(deviceId); if (deviceState != null) { @@ -193,4 +214,12 @@ public class TbDeviceProfileNode implements TbNode { } } + protected void invalidateDeviceProfileCache(DeviceId deviceId, DeviceProfileId deviceProfileId) { + DeviceState deviceState = deviceStates.get(deviceId); + if (deviceState != null) { + if (!deviceState.getProfileId().equals(deviceProfileId)) { + deviceStates.remove(deviceId); + } + } + } }