tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Improvements to Device Profile Rule Node

pull/3688/head
Andrii Shvaika 6 years ago
parent
d4ae02f9a6
commit
feb9120050
9 changed files with 80 additions and 28 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 6
      application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
  2. 3
      application/src/main/java/org/thingsboard/server/controller/DeviceController.java
  3. 54
      application/src/main/java/org/thingsboard/server/service/profile/DefaultTbDeviceProfileCache.java
  4. 4
      application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java
  5. 2
      application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
  6. 3
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceProfileCache.java
  7. 4
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
  8. 1
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java
  9. 31
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java

6
application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View File

@ -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<DeviceProfile> listener) {
mainCtx.getDeviceProfileCache().addListener(getTenantId(), getSelfId(), listener);
public void addDeviceProfileListeners(Consumer<DeviceProfile> profileListener, BiConsumer<DeviceId, DeviceProfile> deviceListener) {
mainCtx.getDeviceProfileCache().addListener(getTenantId(), getSelfId(), profileListener, deviceListener);
}
@Override

3
application/src/main/java/org/thingsboard/server/controller/DeviceController.java
View File

@ -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(),

54
application/src/main/java/org/thingsboard/server/service/profile/DefaultTbDeviceProfileCache.java
View File

@ -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<DeviceProfileId, DeviceProfile> deviceProfilesMap = new ConcurrentHashMap<>();
private final ConcurrentMap<DeviceId, DeviceProfileId> devicesMap = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, ConcurrentMap<EntityId, Consumer<DeviceProfile>>> listeners = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, ConcurrentMap<EntityId, Consumer<DeviceProfile>>> profileListeners = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, ConcurrentMap<EntityId, BiConsumer<DeviceId, DeviceProfile>>> 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<DeviceProfile> listener) {
listeners.computeIfAbsent(tenantId, id -> new ConcurrentHashMap<>()).put(listenerId, listener);
public void addListener(TenantId tenantId, EntityId listenerId,
Consumer<DeviceProfile> profileListener,
BiConsumer<DeviceId, DeviceProfile> 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<EntityId, Consumer<DeviceProfile>> tenantListeners = listeners.get(tenantId);
ConcurrentMap<EntityId, Consumer<DeviceProfile>> tenantListeners = profileListeners.get(tenantId);
if (tenantListeners != null) {
tenantListeners.remove(listenerId);
}
ConcurrentMap<EntityId, BiConsumer<DeviceId, DeviceProfile>> deviceListeners = deviceProfileListeners.get(tenantId);
if (deviceListeners != null) {
deviceListeners.remove(listenerId);
}
}
private void notifyListeners(DeviceProfile profile) {
ConcurrentMap<EntityId, Consumer<DeviceProfile>> tenantListeners = listeners.get(profile.getTenantId());
private void notifyProfileListeners(DeviceProfile profile) {
ConcurrentMap<EntityId, Consumer<DeviceProfile>> tenantListeners = profileListeners.get(profile.getTenantId());
if (tenantListeners != null) {
tenantListeners.forEach((id, listener) -> listener.accept(profile));
}
}
private void notifyDeviceListeners(DeviceId deviceId, DeviceProfile profile) {
ConcurrentMap<EntityId, BiConsumer<DeviceId, DeviceProfile>> tenantListeners = deviceProfileListeners.get(profile.getTenantId());
if (tenantListeners != null) {
tenantListeners.forEach((id, listener) -> listener.accept(deviceId, profile));
}
}
}

4
application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java
View File

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

2
application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
View File

@ -165,7 +165,7 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
} else if (EntityType.DEVICE_PROFILE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
deviceProfileCache.evict(componentLifecycleMsg.getTenantId(), new DeviceProfileId(componentLifecycleMsg.getEntityId().getId()));
} else if (EntityType.DEVICE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
deviceProfileCache.evict(new DeviceId(componentLifecycleMsg.getEntityId().getId()));
deviceProfileCache.evict(componentLifecycleMsg.getTenantId(), new DeviceId(componentLifecycleMsg.getEntityId().getId()));
} else if (EntityType.API_USAGE_STATE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
apiUsageStateService.onApiUsageStateUpdate(componentLifecycleMsg.getTenantId());
}

3
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceProfileCache.java
View File

@ -21,6 +21,7 @@ import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
/**
@ -32,7 +33,7 @@ public interface RuleEngineDeviceProfileCache {
DeviceProfile get(TenantId tenantId, DeviceId deviceId);
void addListener(TenantId tenantId, EntityId listenerId, Consumer<DeviceProfile> listener);
void addListener(TenantId tenantId, EntityId listenerId, Consumer<DeviceProfile> profileListener, BiConsumer<DeviceId, DeviceProfile> devicelistener);
void removeListener(TenantId tenantId, EntityId listenerId);

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

@ -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<DeviceProfile> listener);
void addDeviceProfileListeners(Consumer<DeviceProfile> listener, BiConsumer<DeviceId, DeviceProfile> deviceListener);
void removeProfileListener();
}

1
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java
View File

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

31
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java
View File

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

Write Preview
Loading…
Cancel
Save