Browse Source

Merge remote-tracking branch 'upstream/master' into develop/3.3-edge

pull/3811/head
Volodymyr Babak 6 years ago
parent
commit
c46c1b08e7
  1. 3
      application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
  2. 2
      application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActor.java
  3. 9
      application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java
  4. 4
      application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java
  5. 9
      application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java
  6. 4
      application/src/main/java/org/thingsboard/server/controller/BaseController.java
  7. 5
      application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java
  8. 10
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
  9. 10
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
  10. 3
      application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
  11. 14
      application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java
  12. 5
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
  13. 2
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java
  14. 49
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java
  15. 6
      application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java
  16. 9
      application/src/main/resources/thingsboard.yml
  17. 20
      common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java
  18. 26
      common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java
  19. 1
      common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
  20. 5
      common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
  21. 40
      common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java
  22. 4
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java
  23. 4
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java
  24. 4
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java
  25. 37
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java
  26. 52
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java
  27. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
  28. 75
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java
  29. 24
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java
  30. 3
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java

3
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java

@ -36,6 +36,7 @@ import org.thingsboard.server.common.data.rule.RuleChainType;
import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.plugin.RuleNodeUpdatedMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
@ -132,7 +133,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
} else {
log.trace("[{}][{}] Updating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
existing.setSelf(ruleNode);
existing.getSelfActor().tellWithHighPriority(new ComponentLifecycleMsg(tenantId, existing.getSelf().getId(), ComponentLifecycleEvent.UPDATED));
existing.getSelfActor().tellWithHighPriority(new RuleNodeUpdatedMsg(tenantId, existing.getSelf().getId()));
}
}

2
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActor.java

@ -26,7 +26,6 @@ import org.thingsboard.server.actors.service.ContextBasedCreator;
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.rule.RuleChain;
import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
@ -54,6 +53,7 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa
protected boolean doProcess(TbActorMsg msg) {
switch (msg.getMsgType()) {
case COMPONENT_LIFE_CYCLE_MSG:
case RULE_NODE_UPDATED_MSG:
onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
break;
case RULE_CHAIN_TO_RULE_MSG:

9
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java

@ -20,14 +20,13 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.TbActorCtx;
import org.thingsboard.server.actors.TbActorRef;
import org.thingsboard.server.actors.TbRuleNodeUpdateException;
import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
import org.thingsboard.server.common.data.ApiUsageRecordKey;
import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.common.data.tenant.profile.TenantProfileConfiguration;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.RuleNodeException;
@ -78,7 +77,11 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
if (tbNode != null) {
tbNode.destroy();
}
start(context);
try {
start(context);
} catch (Exception e) {
throw new TbRuleNodeUpdateException("Failed to update rule node", e);
}
}
}

4
application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java

@ -20,6 +20,7 @@ import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.TbActor;
import org.thingsboard.server.actors.TbActorCtx;
import org.thingsboard.server.actors.TbActorException;
import org.thingsboard.server.actors.TbRuleNodeUpdateException;
import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
import org.thingsboard.server.actors.stats.StatsPersistMsg;
import org.thingsboard.server.common.data.id.EntityId;
@ -123,6 +124,9 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
} catch (Exception e) {
logAndPersist("onLifecycleMsg", e, true);
logLifecycleEvent(msg.getEvent(), e);
if (e instanceof TbRuleNodeUpdateException) {
throw (TbRuleNodeUpdateException) e;
}
}
}

9
application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java

@ -34,6 +34,7 @@ import org.thingsboard.server.actors.app.AppInitMsg;
import org.thingsboard.server.actors.stats.StatsActor;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
@ -43,7 +44,7 @@ import java.util.concurrent.ScheduledExecutorService;
@Service
@Slf4j
public class DefaultActorService implements ActorService {
public class DefaultActorService extends TbApplicationEventListener<PartitionChangeEvent> implements ActorService {
public static final String APP_DISPATCHER_NAME = "app-dispatcher";
public static final String TENANT_DISPATCHER_NAME = "tenant-dispatcher";
@ -120,10 +121,10 @@ public class DefaultActorService implements ActorService {
appActor.tellWithHighPriority(new AppInitMsg());
}
@EventListener(PartitionChangeEvent.class)
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
@Override
protected void onTbApplicationEvent(PartitionChangeEvent event) {
log.info("Received partition change event.");
this.appActor.tellWithHighPriority(new PartitionChangeMsg(partitionChangeEvent.getServiceQueueKey(), partitionChangeEvent.getPartitions()));
this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceQueueKey(), event.getPartitions()));
}
@PreDestroy

4
application/src/main/java/org/thingsboard/server/controller/BaseController.java

@ -852,7 +852,7 @@ public abstract class BaseController {
String scope = extractParameter(String.class, 0, additionalInfo);
@SuppressWarnings("unchecked")
List<AttributeKvEntry> attributes = extractParameter(List.class, 1, additionalInfo);
metaData.putValue("scope", scope);
metaData.putValue(DataConstants.SCOPE, scope);
if (attributes != null) {
for (AttributeKvEntry attr : attributes) {
addKvEntry(entityNode, attr);
@ -862,7 +862,7 @@ public abstract class BaseController {
String scope = extractParameter(String.class, 0, additionalInfo);
@SuppressWarnings("unchecked")
List<String> keys = extractParameter(List.class, 1, additionalInfo);
metaData.putValue("scope", scope);
metaData.putValue(DataConstants.SCOPE, scope);
ArrayNode attrsArrayNode = entityNode.putArray("attributes");
if (keys != null) {
keys.forEach(attrsArrayNode::add);

5
application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java

@ -54,6 +54,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.UsageStatsKVProto;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import org.thingsboard.server.queue.scheduler.SchedulerComponent;
import org.thingsboard.server.service.queue.TbClusterService;
import org.thingsboard.server.service.telemetry.InternalTelemetryService;
@ -78,7 +79,7 @@ import java.util.stream.Collectors;
@Slf4j
@Service
public class DefaultTbApiUsageStateService implements TbApiUsageStateService {
public class DefaultTbApiUsageStateService extends TbApplicationEventListener<PartitionChangeEvent> implements TbApiUsageStateService {
public static final String HOURLY = "Hourly";
public static final FutureCallback<Integer> VOID_CALLBACK = new FutureCallback<Integer>() {
@ -188,7 +189,7 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService {
}
@Override
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (partitionChangeEvent.getServiceType().equals(ServiceType.TB_CORE)) {
myTenantStates.entrySet().removeIf(entry -> !partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition());
otherTenantStates.entrySet().removeIf(entry -> partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition());

10
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java

@ -156,12 +156,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
}
@Override
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (partitionChangeEvent.getServiceType().equals(getServiceType())) {
log.info("Subscribing to partitions: {}", partitionChangeEvent.getPartitions());
this.mainConsumer.subscribe(partitionChangeEvent.getPartitions());
protected void onTbApplicationEvent(PartitionChangeEvent event) {
if (event.getServiceType().equals(getServiceType())) {
log.info("Subscribing to partitions: {}", event.getPartitions());
this.mainConsumer.subscribe(event.getPartitions());
this.usageStatsConsumer.subscribe(
partitionChangeEvent
event
.getPartitions()
.stream()
.map(tpi -> tpi.newByTopic(usageStatsConsumer.getTopic()))

10
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java

@ -140,11 +140,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
}
@Override
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (partitionChangeEvent.getServiceType().equals(getServiceType())) {
ServiceQueue serviceQueue = partitionChangeEvent.getServiceQueueKey().getServiceQueue();
log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), partitionChangeEvent.getPartitions());
consumers.get(serviceQueue.getQueue()).subscribe(partitionChangeEvent.getPartitions());
protected void onTbApplicationEvent(PartitionChangeEvent event) {
if (event.getServiceType().equals(getServiceType())) {
ServiceQueue serviceQueue = event.getServiceQueueKey().getServiceQueue();
log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), event.getPartitions());
consumers.get(serviceQueue.getQueue()).subscribe(event.getPartitions());
}
}

3
application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

@ -36,6 +36,7 @@ import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
import org.thingsboard.server.common.transport.util.DataDecodingEncodingService;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
@ -56,7 +57,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
@Slf4j
public abstract class AbstractConsumerService<N extends com.google.protobuf.GeneratedMessageV3> implements ApplicationListener<PartitionChangeEvent> {
public abstract class AbstractConsumerService<N extends com.google.protobuf.GeneratedMessageV3> extends TbApplicationEventListener<PartitionChangeEvent> {
protected volatile ExecutorService consumersExecutor;
protected volatile ExecutorService notificationsConsumerExecutor;

14
application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java

@ -56,6 +56,7 @@ import org.thingsboard.server.dao.util.mapping.JacksonUtil;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.queue.TbClusterService;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
@ -90,7 +91,7 @@ import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE;
@Service
@TbCoreComponent
@Slf4j
public class DefaultDeviceStateService implements DeviceStateService {
public class DefaultDeviceStateService extends TbApplicationEventListener<PartitionChangeEvent> implements DeviceStateService {
public static final String ACTIVITY_STATE = "active";
public static final String LAST_CONNECT_TIME = "lastConnectTime";
@ -206,7 +207,6 @@ public class DefaultDeviceStateService implements DeviceStateService {
if (!state.isActive()) {
state.setActive(true);
save(deviceId, ACTIVITY_STATE, state.isActive());
stateData.getMetaData().putValue("scope", SERVER_SCOPE);
pushRuleEngineMessage(stateData, ACTIVITY_EVENT);
}
}
@ -295,7 +295,7 @@ public class DefaultDeviceStateService implements DeviceStateService {
}
@Override
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
deduplicationExecutor.submit(partitionChangeEvent.getPartitions());
}
@ -447,7 +447,7 @@ public class DefaultDeviceStateService implements DeviceStateService {
}
private <T extends KvEntry> Function<List<T>, DeviceStateData> extractDeviceStateData(Device device) {
return new Function<List<T>, DeviceStateData>() {
return new Function<>() {
@Nullable
@Override
public DeviceStateData apply(@Nullable List<T> data) {
@ -503,7 +503,11 @@ public class DefaultDeviceStateService implements DeviceStateService {
} else {
data = JacksonUtil.toString(state);
}
TbMsg tbMsg = TbMsg.newMsg(msgType, stateData.getDeviceId(), stateData.getMetaData().copy(), TbMsgDataType.JSON, data);
TbMsgMetaData md = stateData.getMetaData().copy();
if(!persistToTelemetry){
md.putValue(DataConstants.SCOPE, SERVER_SCOPE);
}
TbMsg tbMsg = TbMsg.newMsg(msgType, stateData.getDeviceId(), md, TbMsgDataType.JSON, data);
clusterService.pushMsgToRuleEngine(stateData.getTenantId(), stateData.getDeviceId(), tbMsg, null);
} catch (Exception e) {
log.warn("[{}] Failed to push inactivity alarm: {}", stateData.getDeviceId(), state, e);

5
application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java

@ -48,6 +48,7 @@ import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
import org.thingsboard.server.queue.util.TbCoreComponent;
@ -76,7 +77,7 @@ import java.util.function.Predicate;
@Slf4j
@TbCoreComponent
@Service
public class DefaultSubscriptionManagerService implements SubscriptionManagerService {
public class DefaultSubscriptionManagerService extends TbApplicationEventListener<PartitionChangeEvent> implements SubscriptionManagerService {
@Autowired
private AttributesService attrService;
@ -178,7 +179,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
}
@Override
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
Set<TopicPartitionInfo> removedPartitions = new HashSet<>(currentPartitions);
removedPartitions.removeAll(partitionChangeEvent.getPartitions());

2
application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java

@ -476,7 +476,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
public void cancelAllSessionSubscriptions(String sessionId) {
Map<Integer, TbAbstractDataSubCtx> sessionSubs = subscriptionsBySessionId.remove(sessionId);
if (sessionSubs != null) {
sessionSubs.values().stream().filter(sub -> sub instanceof TbEntityDataSubCtx).map(sub -> (TbEntityDataSubCtx) sub).forEach(this::cleanupAndCancel);
sessionSubs.values().forEach(this::cleanupAndCancel);
}
}

49
application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java

@ -28,6 +28,7 @@ import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.queue.TbClusterService;
import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
@ -62,6 +63,34 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
private SubscriptionManagerService subscriptionManagerService;
private ExecutorService subscriptionUpdateExecutor;
private TbApplicationEventListener<PartitionChangeEvent> partitionChangeListener = new TbApplicationEventListener<>() {
@Override
protected void onTbApplicationEvent(PartitionChangeEvent event) {
if (ServiceType.TB_CORE.equals(event.getServiceType())) {
currentPartitions.clear();
currentPartitions.addAll(event.getPartitions());
}
}
};
private TbApplicationEventListener<ClusterTopologyChangeEvent> clusterTopologyChangeListener = new TbApplicationEventListener<>() {
@Override
protected void onTbApplicationEvent(ClusterTopologyChangeEvent event) {
if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) {
/*
* If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again.
* Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart.
* Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically
* It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService.
* Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache
* Since number of subscriptions is usually much less then number of devices that are pushing data.
*/
subscriptionsBySessionId.values().forEach(map -> map.values()
.forEach(sub -> pushSubscriptionToManagerService(sub, true)));
}
}
};
@PostConstruct
public void initExecutor() {
@ -77,28 +106,14 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
@Override
@EventListener(PartitionChangeEvent.class)
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
currentPartitions.clear();
currentPartitions.addAll(partitionChangeEvent.getPartitions());
}
public void onApplicationEvent(PartitionChangeEvent event) {
partitionChangeListener.onApplicationEvent(event);
}
@Override
@EventListener(ClusterTopologyChangeEvent.class)
public void onApplicationEvent(ClusterTopologyChangeEvent event) {
if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) {
/*
* If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again.
* Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart.
* Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically
* It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService.
* Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache
* Since number of subscriptions is usually much less then number of devices that are pushing data.
*/
subscriptionsBySessionId.values().forEach(map -> map.values()
.forEach(sub -> pushSubscriptionToManagerService(sub, true)));
}
clusterTopologyChangeListener.onApplicationEvent(event);
}
//TODO 3.1: replace null callbacks with callbacks from websocket service.

6
application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java

@ -41,6 +41,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import org.thingsboard.server.service.queue.TbClusterService;
import org.thingsboard.server.service.subscription.SubscriptionManagerService;
import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
@ -61,7 +62,7 @@ import java.util.function.Consumer;
* Created by ashvayka on 27.03.18.
*/
@Slf4j
public abstract class AbstractSubscriptionService implements ApplicationListener<PartitionChangeEvent> {
public abstract class AbstractSubscriptionService extends TbApplicationEventListener<PartitionChangeEvent>{
protected final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();
@ -97,8 +98,7 @@ public abstract class AbstractSubscriptionService implements ApplicationListener
}
@Override
@EventListener(PartitionChangeEvent.class)
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
currentPartitions.clear();
currentPartitions.addAll(partitionChangeEvent.getPartitions());

9
application/src/main/resources/thingsboard.yml

@ -118,6 +118,15 @@ security:
githubMapper:
emailUrl: "${SECURITY_OAUTH2_GITHUB_MAPPER_EMAIL_URL_KEY:https://api.github.com/user/emails}"
# Usage statistics parameters
usage:
stats:
report:
enabled: "${USAGE_STATS_REPORT_ENABLED:true}"
interval: "${USAGE_STATS_REPORT_INTERVAL:10}"
check:
cycle: "${USAGE_STATS_CHECK_CYCLE:60000}"
# Dashboard parameters
dashboard:
# Maximum allowed datapoints fetched by widgets

20
common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java

@ -17,6 +17,7 @@ package org.thingsboard.server.actors;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.TbActorStopReason;
@ -73,7 +74,7 @@ public final class TbActorMailbox implements TbActorCtx {
if (strategy.isStop() || (settings.getMaxActorInitAttempts() > 0 && attemptIdx > settings.getMaxActorInitAttempts())) {
log.info("[{}] Failed to init actor, attempt {}, going to stop attempts.", selfId, attempt, t);
stopReason = TbActorStopReason.INIT_FAILED;
system.stop(selfId);
destroy();
} else if (strategy.getRetryDelay() > 0) {
log.info("[{}] Failed to init actor, attempt {}, going to retry in attempts in {}ms", selfId, attempt, strategy.getRetryDelay());
log.debug("[{}] Error", selfId, t);
@ -95,7 +96,19 @@ public final class TbActorMailbox implements TbActorCtx {
}
tryProcessQueue(true);
} else {
msg.onTbActorStopped(stopReason);
if (highPriority && msg.getMsgType().equals(MsgType.RULE_NODE_UPDATED_MSG)) {
synchronized (this) {
if (stopReason == TbActorStopReason.INIT_FAILED) {
destroyInProgress.set(false);
stopReason = null;
initActor();
} else {
msg.onTbActorStopped(stopReason);
}
}
} else {
msg.onTbActorStopped(stopReason);
}
}
}
@ -126,6 +139,9 @@ public final class TbActorMailbox implements TbActorCtx {
try {
log.debug("[{}] Going to process message: {}", selfId, msg);
actor.process(msg);
} catch (TbRuleNodeUpdateException updateException){
stopReason = TbActorStopReason.INIT_FAILED;
destroy();
} catch (Throwable t) {
log.debug("[{}] Failed to process message: {}", selfId, msg, t);
ProcessFailureStrategy strategy = actor.onProcessFailure(t);

26
common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java

@ -0,0 +1,26 @@
/**
* Copyright © 2016-2021 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.actors;
public class TbRuleNodeUpdateException extends RuntimeException {
private static final long serialVersionUID = 8209771144711980882L;
public TbRuleNodeUpdateException(String message, Throwable cause) {
super(message, cause);
}
}

1
common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java

@ -24,6 +24,7 @@ public class DataConstants {
public static final String CUSTOMER = "CUSTOMER";
public static final String DEVICE = "DEVICE";
public static final String SCOPE = "scope";
public static final String CLIENT_SCOPE = "CLIENT_SCOPE";
public static final String SERVER_SCOPE = "SERVER_SCOPE";
public static final String SHARED_SCOPE = "SHARED_SCOPE";

5
common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java

@ -39,6 +39,11 @@ public enum MsgType {
*/
COMPONENT_LIFE_CYCLE_MSG,
/**
* Special message to indicate rule node update request
*/
RULE_NODE_UPDATED_MSG,
/**
* Misc messages consumed from the Queue and forwarded to Rule Engine Actor.
*

40
common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java

@ -0,0 +1,40 @@
/**
* Copyright © 2016-2021 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.common.msg.plugin;
import lombok.ToString;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.MsgType;
import java.util.Optional;
/**
* @author Andrew Shvayka
*/
@ToString
public class RuleNodeUpdatedMsg extends ComponentLifecycleMsg {
public RuleNodeUpdatedMsg(TenantId tenantId, EntityId entityId) {
super(tenantId, entityId, ComponentLifecycleEvent.UPDATED);
}
@Override
public MsgType getMsgType() {
return MsgType.RULE_NODE_UPDATED_MSG;
}
}

4
common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java

@ -22,7 +22,9 @@ import org.thingsboard.server.common.msg.queue.ServiceQueueKey;
import java.util.Set;
public class ClusterTopologyChangeEvent extends ApplicationEvent {
public class ClusterTopologyChangeEvent extends TbApplicationEvent {
private static final long serialVersionUID = -2441739930040282254L;
@Getter
private final Set<ServiceQueueKey> serviceQueueKeys;

4
common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

@ -126,7 +126,7 @@ public class HashPartitionService implements PartitionService {
}
@Override
public void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
public synchronized void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
logServiceInfo(currentService);
otherServices.forEach(this::logServiceInfo);
Map<ServiceQueueKey, List<ServiceInfo>> queueServicesMap = new HashMap<>();
@ -134,7 +134,7 @@ public class HashPartitionService implements PartitionService {
for (ServiceInfo other : otherServices) {
addNode(queueServicesMap, other);
}
queueServicesMap.values().forEach(list -> list.sort((a, b) -> a.getServiceId().compareTo(b.getServiceId())));
queueServicesMap.values().forEach(list -> list.sort(Comparator.comparing(ServiceInfo::getServiceId)));
ConcurrentMap<ServiceQueueKey, List<Integer>> oldPartitions = myPartitions;
TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService);

4
common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java

@ -24,7 +24,9 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import java.util.Set;
public class PartitionChangeEvent extends ApplicationEvent {
public class PartitionChangeEvent extends TbApplicationEvent {
private static final long serialVersionUID = -8731788167026510559L;
@Getter
private final ServiceQueueKey serviceQueueKey;

37
common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java

@ -0,0 +1,37 @@
/**
* Copyright © 2016-2021 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.queue.discovery;
import lombok.Getter;
import org.springframework.context.ApplicationEvent;
import java.util.concurrent.atomic.AtomicInteger;
public class TbApplicationEvent extends ApplicationEvent {
private static final long serialVersionUID = 3884264064887765146L;
private static final AtomicInteger sequence = new AtomicInteger();
@Getter
private final int sequenceNumber;
public TbApplicationEvent(Object source) {
super(source);
sequenceNumber = sequence.incrementAndGet();
}
}

52
common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java

@ -0,0 +1,52 @@
/**
* Copyright © 2016-2021 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.queue.discovery;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationListener;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public abstract class TbApplicationEventListener<T extends TbApplicationEvent> implements ApplicationListener<T> {
private int lastProcessedSequenceNumber = Integer.MIN_VALUE;
private final Lock seqNumberLock = new ReentrantLock();
@Override
public void onApplicationEvent(T event) {
boolean validUpdate = false;
seqNumberLock.lock();
try {
if (event.getSequenceNumber() > lastProcessedSequenceNumber) {
validUpdate = true;
lastProcessedSequenceNumber = event.getSequenceNumber();
}
} finally {
seqNumberLock.unlock();
}
if (validUpdate) {
onTbApplicationEvent(event);
} else {
log.info("Application event ignored due to invalid sequence number ({} > {}). Event: {}", lastProcessedSequenceNumber, event.getSequenceNumber(), event);
}
}
protected abstract void onTbApplicationEvent(T event);
}

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java

@ -76,7 +76,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode {
if (!msg.getMetaData().getData().isEmpty()) {
long now = System.currentTimeMillis();
String scope = msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) ?
DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue("scope");
DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue(DataConstants.SCOPE);
ListenableFuture<List<EntityView>> entityViewsFuture =
ctx.getEntityViewService().findEntityViewsByTenantIdAndEntityIdAsync(ctx.getTenantId(), msg.getOriginator());

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

@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.dao.timeseries.TimeseriesService;
import org.thingsboard.server.dao.util.mapping.JacksonUtil;
@ -72,41 +73,45 @@ public class CalculateDeltaNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
JsonNode json = JacksonUtil.toJsonNode(msg.getData());
String inputKey = config.getInputValueKey();
if (json.has(inputKey)) {
DonAsynchron.withCallback(getLastValue(msg.getOriginator()),
previousData -> {
double currentValue = json.get(inputKey).asDouble();
long currentTs = TbMsgTimeseriesNode.getTs(msg);
if (useCache) {
cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue));
}
BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0);
if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) {
ctx.tellNext(msg, TbRelationTypes.FAILURE);
return;
}
if (config.getRound() != null) {
delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP);
}
ObjectNode result = (ObjectNode) json;
result.put(config.getOutputValueKey(), delta);
if (config.isAddPeriodBetweenMsgs()) {
long period = previousData != null ? currentTs - previousData.ts : 0;
result.put(config.getPeriodValueKey(), period);
}
ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result)));
},
t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
} else if (config.isTellFailureIfInputValueKeyIsAbsent()) {
ctx.tellNext(msg, TbRelationTypes.FAILURE);
if (msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) {
JsonNode json = JacksonUtil.toJsonNode(msg.getData());
String inputKey = config.getInputValueKey();
if (json.has(inputKey)) {
DonAsynchron.withCallback(getLastValue(msg.getOriginator()),
previousData -> {
double currentValue = json.get(inputKey).asDouble();
long currentTs = TbMsgTimeseriesNode.getTs(msg);
if (useCache) {
cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue));
}
BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0);
if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) {
ctx.tellNext(msg, TbRelationTypes.FAILURE);
return;
}
if (config.getRound() != null) {
delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP);
}
ObjectNode result = (ObjectNode) json;
result.put(config.getOutputValueKey(), delta);
if (config.isAddPeriodBetweenMsgs()) {
long period = previousData != null ? currentTs - previousData.ts : 0;
result.put(config.getPeriodValueKey(), period);
}
ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result)));
},
t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
} else if (config.isTellFailureIfInputValueKeyIsAbsent()) {
ctx.tellNext(msg, TbRelationTypes.FAILURE);
} else {
ctx.tellSuccess(msg);
}
} else {
ctx.tellSuccess(msg);
}

24
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java

@ -138,6 +138,8 @@ class DeviceState {
stateChanged = processTelemetry(ctx, msg);
} else if (msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name())) {
stateChanged = processAttributesUpdateRequest(ctx, msg);
} else if (msg.getType().equals(DataConstants.ACTIVITY_EVENT) || msg.getType().equals(DataConstants.INACTIVITY_EVENT)) {
stateChanged = processDeviceActivityEvent(ctx, msg);
} else if (msg.getType().equals(DataConstants.ATTRIBUTES_UPDATED)) {
stateChanged = processAttributesUpdateNotification(ctx, msg);
} else if (msg.getType().equals(DataConstants.ATTRIBUTES_DELETED)) {
@ -158,6 +160,15 @@ class DeviceState {
}
}
private boolean processDeviceActivityEvent(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
String scope = msg.getMetaData().getValue(DataConstants.SCOPE);
if (StringUtils.isEmpty(scope)) {
return processTelemetry(ctx, msg);
} else {
return processAttributes(ctx, msg, scope);
}
}
private boolean processAlarmClearNotification(TbContext ctx, TbMsg msg) {
boolean stateChanged = false;
Alarm alarmNf = JacksonUtil.fromString(msg.getData(), Alarm.class);
@ -181,19 +192,18 @@ class DeviceState {
}
private boolean processAttributesUpdateNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
String scope = msg.getMetaData().getValue("scope");
String scope = msg.getMetaData().getValue(DataConstants.SCOPE);
if (StringUtils.isEmpty(scope)) {
scope = DataConstants.CLIENT_SCOPE;
}
return processAttributesUpdate(ctx, msg, attributes, scope);
return processAttributes(ctx, msg, scope);
}
private boolean processAttributesDeleteNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
boolean stateChanged = false;
List<String> keys = new ArrayList<>();
new JsonParser().parse(msg.getData()).getAsJsonObject().get("attributes").getAsJsonArray().forEach(e -> keys.add(e.getAsString()));
String scope = msg.getMetaData().getValue("scope");
String scope = msg.getMetaData().getValue(DataConstants.SCOPE);
if (StringUtils.isEmpty(scope)) {
scope = DataConstants.CLIENT_SCOPE;
}
@ -211,12 +221,12 @@ class DeviceState {
}
protected boolean processAttributesUpdateRequest(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
return processAttributesUpdate(ctx, msg, attributes, DataConstants.CLIENT_SCOPE);
return processAttributes(ctx, msg, DataConstants.CLIENT_SCOPE);
}
private boolean processAttributesUpdate(TbContext ctx, TbMsg msg, Set<AttributeKvEntry> attributes, String scope) throws ExecutionException, InterruptedException {
private boolean processAttributes(TbContext ctx, TbMsg msg, String scope) throws ExecutionException, InterruptedException {
boolean stateChanged = false;
Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
if (!attributes.isEmpty()) {
SnapshotUpdate update = merge(latestValues, attributes, scope);
for (DeviceProfileAlarm alarm : deviceProfile.getAlarmSettings()) {

3
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java

@ -134,7 +134,8 @@ public class TbDeviceProfileNode implements TbNode {
if (deviceState != null) {
deviceState.process(ctx, msg);
} else {
ctx.tellFailure(msg, new IllegalStateException("Device profile for device [" + deviceId + "] not found!"));
log.info("Device was not found! Most probably device [" + deviceId + "] has been removed from the database. Acknowledging msg.");
ctx.ack(msg);
}
}
} else {

Loading…
Cancel
Save