From 411c9dabdafa687362e66e7b3358405f1431fe4c Mon Sep 17 00:00:00 2001 From: VoBa Date: Mon, 15 Feb 2021 12:24:30 +0200 Subject: [PATCH 01/36] Added usage statistics configuration to yml file (#4097) * Remove device from cache in case null value cached in the distributed redis * Handle case when device was removed from db but message in the queue exists * Code review chagnes * Added usage statistics configuration to yml file --- application/src/main/resources/thingsboard.yml | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 0cc7c08669..4e9693448a 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/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 From 6c1074a8b04767c652940aba3b3298f075727b89 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Mon, 15 Feb 2021 14:25:40 +0200 Subject: [PATCH 02/36] Fix for race condition in the partition change events --- .../actors/service/DefaultActorService.java | 9 ++-- .../DefaultTbApiUsageStateService.java | 5 +- .../queue/DefaultTbCoreConsumerService.java | 10 ++-- .../DefaultTbRuleEngineConsumerService.java | 10 ++-- .../processing/AbstractConsumerService.java | 3 +- .../state/DefaultDeviceStateService.java | 5 +- .../DefaultSubscriptionManagerService.java | 5 +- .../DefaultTbLocalSubscriptionService.java | 49 +++++++++++------ .../AbstractSubscriptionService.java | 6 +-- .../discovery/ClusterTopologyChangeEvent.java | 4 +- .../queue/discovery/HashPartitionService.java | 4 +- .../queue/discovery/PartitionChangeEvent.java | 4 +- .../queue/discovery/TbApplicationEvent.java | 37 +++++++++++++ .../discovery/TbApplicationEventListener.java | 52 +++++++++++++++++++ 14 files changed, 158 insertions(+), 45 deletions(-) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java diff --git a/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java b/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java index 6b7b5ff3d1..05363dfd59 100644 --- a/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java +++ b/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 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 diff --git a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java index d4a5d42320..d0a3984660 100644 --- a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java +++ b/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 implements TbApiUsageStateService { public static final String HOURLY = "Hourly"; public static final FutureCallback VOID_CALLBACK = new FutureCallback() { @@ -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()); 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 df0e7f86b5..af9239d298 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 @@ -151,12 +151,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService tpi.newByTopic(usageStatsConsumer.getTopic())) 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 106ca40c94..390798a3e2 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 @@ -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()); } } 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 31d5cf47c3..02378eb557 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 @@ -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 implements ApplicationListener { +public abstract class AbstractConsumerService extends TbApplicationEventListener { protected volatile ExecutorService consumersExecutor; protected volatile ExecutorService notificationsConsumerExecutor; diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index 3087c68f88..be31a0df45 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/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 implements DeviceStateService { public static final String ACTIVITY_STATE = "active"; public static final String LAST_CONNECT_TIME = "lastConnectTime"; @@ -294,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()); } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index bafbb45ba8..844db92de8 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/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 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 removedPartitions = new HashSet<>(currentPartitions); removedPartitions.removeAll(partitionChangeEvent.getPartitions()); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index ee00b28562..0220a94964 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/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 partitionChangeListener = new TbApplicationEventListener<>() { + @Override + protected void onTbApplicationEvent(PartitionChangeEvent event) { + if (ServiceType.TB_CORE.equals(event.getServiceType())) { + currentPartitions.clear(); + currentPartitions.addAll(event.getPartitions()); + } + } + }; + + private TbApplicationEventListener 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. diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java index 8827a71f70..168e1271fd 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java +++ b/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 { +public abstract class AbstractSubscriptionService extends TbApplicationEventListener{ protected final Set 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()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java index 0602dee3ae..1e5b90b5fe 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java +++ b/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 serviceQueueKeys; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java index d8164f4be8..2da438417a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java +++ b/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 otherServices) { + public synchronized void recalculatePartitions(ServiceInfo currentService, List otherServices) { logServiceInfo(currentService); otherServices.forEach(this::logServiceInfo); Map> 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> oldPartitions = myPartitions; TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java index e4edabbe19..2edcd2ceca 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java +++ b/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; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java new file mode 100644 index 0000000000..face2d36d6 --- /dev/null +++ b/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(); + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java new file mode 100644 index 0000000000..9158d8f0c8 --- /dev/null +++ b/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 implements ApplicationListener { + + 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); + + +} From a15e991d23d9bd421fdfa673f170418d5fe2d5fd Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Tue, 16 Feb 2021 12:52:25 +0200 Subject: [PATCH 03/36] Added support of BigDecimal to the JsonConverter --- .../transport/adaptor/JsonConverter.java | 41 ++++++++++---- .../src/test/java/JsonConverterTest.java | 53 +++++++++++++++++++ .../engine/metadata/CalculateDeltaNode.java | 7 ++- 3 files changed, 90 insertions(+), 11 deletions(-) create mode 100644 common/transport/transport-api/src/test/java/JsonConverterTest.java diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java index 362be1ed00..4bb1b96b0a 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java @@ -51,6 +51,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateBasicMqttCre import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; +import java.math.BigDecimal; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; @@ -245,12 +246,25 @@ public class JsonConverter { } private static void parseNumericValue(List result, Entry valueEntry, JsonPrimitive value) { - if (value.getAsString().contains(".")) { - result.add(new DoubleDataEntry(valueEntry.getKey(), value.getAsDouble())); + String valueAsString = value.getAsString(); + String key = valueEntry.getKey(); + if (valueAsString.contains("e") || valueAsString.contains("E")) { + var bd = new BigDecimal(valueAsString); + if (bd.stripTrailingZeros().scale() <= 0) { + try { + result.add(new LongDataEntry(key, bd.longValueExact())); + } catch (ArithmeticException e) { + result.add(new DoubleDataEntry(key, bd.doubleValue())); + } + } else { + result.add(new DoubleDataEntry(key, bd.doubleValue())); + } + } else if (valueAsString.contains(".")) { + result.add(new DoubleDataEntry(key, value.getAsDouble())); } else { try { long longValue = Long.parseLong(value.getAsString()); - result.add(new LongDataEntry(valueEntry.getKey(), longValue)); + result.add(new LongDataEntry(key, longValue)); } catch (NumberFormatException e) { throw new JsonSyntaxException("Big integer values are not supported!"); } @@ -285,7 +299,8 @@ public class JsonConverter { return result; } - public static JsonObject getJsonObjectForGateway(String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg) { + public static JsonObject getJsonObjectForGateway(String deviceName, TransportProtos.GetAttributeResponseMsg + responseMsg) { JsonObject result = new JsonObject(); result.addProperty("id", responseMsg.getRequestId()); result.addProperty(DEVICE_PROPERTY, deviceName); @@ -298,7 +313,8 @@ public class JsonConverter { return result; } - public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg notificationMsg) { + public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg + notificationMsg) { JsonObject result = new JsonObject(); result.addProperty(DEVICE_PROPERTY, deviceName); result.add("data", toJson(notificationMsg)); @@ -446,7 +462,8 @@ public class JsonConverter { return result; } - public static JsonElement toGatewayJson(String deviceName, TransportProtos.ProvisionDeviceResponseMsg responseRequest) { + public static JsonElement toGatewayJson(String deviceName, TransportProtos.ProvisionDeviceResponseMsg + responseRequest) { JsonObject result = new JsonObject(); result.addProperty(DEVICE_PROPERTY, deviceName); result.add("data", JsonConverter.toJson(responseRequest)); @@ -496,15 +513,18 @@ public class JsonConverter { return result; } - public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException { + public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs) throws + JsonSyntaxException { return convertToTelemetry(jsonElement, systemTs, false); } - public static Map> convertToSortedTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException { + public static Map> convertToSortedTelemetry(JsonElement jsonElement, long systemTs) throws + JsonSyntaxException { return convertToTelemetry(jsonElement, systemTs, true); } - public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws JsonSyntaxException { + public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws + JsonSyntaxException { Map> result = sorted ? new TreeMap<>() : new HashMap<>(); convertToTelemetry(jsonElement, systemTs, result, null); return result; @@ -574,7 +594,8 @@ public class JsonConverter { .build(); } - private static TransportProtos.ProvisionDeviceCredentialsMsg buildProvisionDeviceCredentialsMsg(String provisionKey, String provisionSecret) { + private static TransportProtos.ProvisionDeviceCredentialsMsg buildProvisionDeviceCredentialsMsg(String + provisionKey, String provisionSecret) { return TransportProtos.ProvisionDeviceCredentialsMsg.newBuilder() .setProvisionDeviceKey(provisionKey) .setProvisionDeviceSecret(provisionSecret) diff --git a/common/transport/transport-api/src/test/java/JsonConverterTest.java b/common/transport/transport-api/src/test/java/JsonConverterTest.java new file mode 100644 index 0000000000..cedbef50c9 --- /dev/null +++ b/common/transport/transport-api/src/test/java/JsonConverterTest.java @@ -0,0 +1,53 @@ +/** + * 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. + */ + +import com.google.gson.JsonParser; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.junit.MockitoJUnitRunner; +import org.thingsboard.server.common.transport.adaptor.JsonConverter; + +@RunWith(MockitoJUnitRunner.class) +public class JsonConverterTest { + + private static final JsonParser JSON_PARSER = new JsonParser(); + + @Test + public void testParseBigDecimalAsLong() { + var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1E+1}"), 0L); + Assert.assertEquals(10L, result.get(0L).get(0).getLongValue().get().longValue()); + } + + @Test + public void testParseBigDecimalAsDouble() { + var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 101E-1}"), 0L); + Assert.assertEquals(10.1, result.get(0L).get(0).getDoubleValue().get(), 0.0); + } + + @Test + public void testParseAsDouble() { + var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1.1}"), 0L); + Assert.assertEquals(1.1, result.get(0L).get(0).getDoubleValue().get(), 0.0); + } + + @Test + public void testParseAsLong() { + var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 11}"), 0L); + Assert.assertEquals(11L, result.get(0L).get(0).getLongValue().get().longValue()); + } + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java index 1d5ae744c3..55fc7e18ab 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java @@ -93,12 +93,17 @@ public class CalculateDeltaNode implements TbNode { return; } + if (config.getRound() != null) { delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); } ObjectNode result = (ObjectNode) json; - result.put(config.getOutputValueKey(), delta); + if (delta.stripTrailingZeros().scale() > 0) { + result.put(config.getOutputValueKey(), delta.doubleValue()); + } else { + result.put(config.getOutputValueKey(), delta.longValueExact()); + } if (config.isAddPeriodBetweenMsgs()) { long period = previousData != null ? currentTs - previousData.ts : 0; From d5b28222a37d87703531da0c7134060e96c9dd2f Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Mon, 15 Feb 2021 15:16:11 +0200 Subject: [PATCH 04/36] fixed unlimited error messages in TbMsgGeneratorNode --- .../org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java index 69482a7b89..6047370193 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java @@ -104,9 +104,10 @@ public class TbMsgGeneratorNode implements TbNode { } }, t -> { - if (initialized) { + if (initialized && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) { ctx.tellFailure(msg, t); scheduleTickMsg(ctx); + currentMsgCount++; } }); } From ada0af2a4109b3f6a32eef9d821e4a6795ce57fc Mon Sep 17 00:00:00 2001 From: Vladyslav_Prykhodko Date: Thu, 11 Feb 2021 13:38:18 +0200 Subject: [PATCH 05/36] UI: Added in device profile alarm rule condition to dynamic mode checkbox inherit from owner --- .../boolean-filter-predicate.component.html | 4 ++-- .../filter-predicate-list.component.html | 4 ++-- .../filter-predicate-value.component.html | 9 +++++++ .../filter-predicate-value.component.ts | 24 ++++++++++++++++--- .../components/filter/filter-predicate.scss | 4 ++++ .../numeric-filter-predicate.component.html | 4 ++-- .../string-filter-predicate.component.html | 4 ++-- .../app/shared/models/query/query.models.ts | 1 + 8 files changed, 43 insertions(+), 11 deletions(-) diff --git a/ui-ngx/src/app/modules/home/components/filter/boolean-filter-predicate.component.html b/ui-ngx/src/app/modules/home/components/filter/boolean-filter-predicate.component.html index 6194581625..253f7e2138 100644 --- a/ui-ngx/src/app/modules/home/components/filter/boolean-filter-predicate.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/boolean-filter-predicate.component.html @@ -16,7 +16,7 @@ -->
- + @@ -25,7 +25,7 @@ diff --git a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-list.component.html b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-list.component.html index 41c04ea238..c6aee52976 100644 --- a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-list.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-list.component.html @@ -26,12 +26,12 @@
-
+
- +
diff --git a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html index adda4bea1a..8365e2ec8c 100644 --- a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html @@ -61,6 +61,15 @@
filter.dynamic-source-type
+
+ +
Inherit from owner
+
diff --git a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.ts b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.ts index b5406c9e9b..437a5993e7 100644 --- a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.ts +++ b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.ts @@ -44,6 +44,9 @@ import { }) export class FilterPredicateValueComponent implements ControlValueAccessor, OnInit { + private readonly inheritFromSources: DynamicValueSourceType[] = [DynamicValueSourceType.CURRENT_CUSTOMER, + DynamicValueSourceType.CURRENT_DEVICE]; + @Input() disabled: boolean; @Input() @@ -72,6 +75,8 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn dynamicMode = false; + inheritMode = false; + allow = true; private propagateChange = null; @@ -105,7 +110,8 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn dynamicValue: this.fb.group( { sourceType: [null], - sourceAttribute: [null] + sourceAttribute: [null], + inherit: [false] } ) }); @@ -114,6 +120,7 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn if (!sourceType) { this.filterPredicateValueFormGroup.get('dynamicValue').get('sourceAttribute').patchValue(null, {emitEvent: false}); } + this.updateInheritValue(sourceType); } ); this.filterPredicateValueFormGroup.valueChanges.subscribe(() => { @@ -139,10 +146,13 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn writeValue(predicateValue: FilterPredicateValue): void { this.filterPredicateValueFormGroup.get('defaultValue').patchValue(predicateValue.defaultValue, {emitEvent: false}); - this.filterPredicateValueFormGroup.get('dynamicValue').get('sourceType').patchValue(predicateValue.dynamicValue ? + this.filterPredicateValueFormGroup.get('dynamicValue.sourceType').patchValue(predicateValue.dynamicValue ? predicateValue.dynamicValue.sourceType : null, {emitEvent: false}); - this.filterPredicateValueFormGroup.get('dynamicValue').get('sourceAttribute').patchValue(predicateValue.dynamicValue ? + this.filterPredicateValueFormGroup.get('dynamicValue.inherit').patchValue(predicateValue.dynamicValue ? + predicateValue.dynamicValue.inherit : false, {emitEvent: false}); + this.filterPredicateValueFormGroup.get('dynamicValue.sourceAttribute').patchValue(predicateValue.dynamicValue ? predicateValue.dynamicValue.sourceAttribute : null, {emitEvent: false}); + this.updateInheritValue(predicateValue?.dynamicValue?.sourceType); } private updateModel() { @@ -158,4 +168,12 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn this.propagateChange(predicateValue); } + private updateInheritValue(sourceType: DynamicValueSourceType) { + if (this.inheritFromSources.includes(sourceType)) { + this.inheritMode = true; + } else { + this.filterPredicateValueFormGroup.get('dynamicValue.inherit').patchValue(false, {emitEvent: false}); + this.inheritMode = false; + } + } } diff --git a/ui-ngx/src/app/modules/home/components/filter/filter-predicate.scss b/ui-ngx/src/app/modules/home/components/filter/filter-predicate.scss index e88454b8b3..6230a3da0f 100644 --- a/ui-ngx/src/app/modules/home/components/filter/filter-predicate.scss +++ b/ui-ngx/src/app/modules/home/components/filter/filter-predicate.scss @@ -17,6 +17,10 @@ .tb-hint { padding-bottom: 0; } + + .invisible{ + visibility: hidden; + } } :host ::ng-deep { diff --git a/ui-ngx/src/app/modules/home/components/filter/numeric-filter-predicate.component.html b/ui-ngx/src/app/modules/home/components/filter/numeric-filter-predicate.component.html index ffbdaa01c1..678df9c3f9 100644 --- a/ui-ngx/src/app/modules/home/components/filter/numeric-filter-predicate.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/numeric-filter-predicate.component.html @@ -16,7 +16,7 @@ -->
- + @@ -25,7 +25,7 @@ diff --git a/ui-ngx/src/app/modules/home/components/filter/string-filter-predicate.component.html b/ui-ngx/src/app/modules/home/components/filter/string-filter-predicate.component.html index f1913149e5..53f5dee077 100644 --- a/ui-ngx/src/app/modules/home/components/filter/string-filter-predicate.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/string-filter-predicate.component.html @@ -16,7 +16,7 @@ -->
-
+
@@ -29,7 +29,7 @@
diff --git a/ui-ngx/src/app/shared/models/query/query.models.ts b/ui-ngx/src/app/shared/models/query/query.models.ts index 86d2ddd89b..dcc89852b7 100644 --- a/ui-ngx/src/app/shared/models/query/query.models.ts +++ b/ui-ngx/src/app/shared/models/query/query.models.ts @@ -285,6 +285,7 @@ export const dynamicValueSourceTypeTranslationMap = new Map { sourceType: DynamicValueSourceType; sourceAttribute: string; + inherit?: boolean; } export interface FilterPredicateValue { From 5e990da0ba785cf7716d3acb2e61495127852e88 Mon Sep 17 00:00:00 2001 From: Vladyslav_Prykhodko Date: Thu, 11 Feb 2021 16:16:06 +0200 Subject: [PATCH 06/36] UI: Refactoring alarm-rule inherit --- .../filter-predicate-value.component.html | 19 +++++++++---------- .../filter-predicate-value.component.ts | 15 ++++++++------- .../components/filter/filter-predicate.scss | 4 ---- .../assets/locale/locale.constant-en_US.json | 4 +++- 4 files changed, 20 insertions(+), 22 deletions(-) diff --git a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html index 8365e2ec8c..dd1b92cf7f 100644 --- a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html @@ -47,7 +47,7 @@
-
+
@@ -61,15 +61,6 @@
filter.dynamic-source-type
-
- -
Inherit from owner
-
@@ -77,6 +68,14 @@
filter.source-attribute
+
+ + {{ 'filter.inherit-owner' | translate}} + +
filter.source-attribute-not-set
+
diff --git a/ui-ngx/src/app/modules/home/components/filter/string-filter-predicate.component.ts b/ui-ngx/src/app/modules/home/components/filter/string-filter-predicate.component.ts index b52f36a572..11b4dad275 100644 --- a/ui-ngx/src/app/modules/home/components/filter/string-filter-predicate.component.ts +++ b/ui-ngx/src/app/modules/home/components/filter/string-filter-predicate.component.ts @@ -42,6 +42,8 @@ export class StringFilterPredicateComponent implements ControlValueAccessor, OnI @Input() allowUserDynamicSource = true; + @Input() onlyUserDynamicSource = false; + valueTypeEnum = EntityKeyValueType; stringFilterPredicateFormGroup: FormGroup; diff --git a/ui-ngx/src/app/shared/models/query/query.models.ts b/ui-ngx/src/app/shared/models/query/query.models.ts index dcc89852b7..b8dc2564bc 100644 --- a/ui-ngx/src/app/shared/models/query/query.models.ts +++ b/ui-ngx/src/app/shared/models/query/query.models.ts @@ -35,14 +35,16 @@ export enum EntityKeyType { SERVER_ATTRIBUTE = 'SERVER_ATTRIBUTE', TIME_SERIES = 'TIME_SERIES', ENTITY_FIELD = 'ENTITY_FIELD', - ALARM_FIELD = 'ALARM_FIELD' + ALARM_FIELD = 'ALARM_FIELD', + CONSTANT = 'CONSTANT' } export const entityKeyTypeTranslationMap = new Map( [ [EntityKeyType.ATTRIBUTE, 'filter.key-type.attribute'], [EntityKeyType.TIME_SERIES, 'filter.key-type.timeseries'], - [EntityKeyType.ENTITY_FIELD, 'filter.key-type.entity-field'] + [EntityKeyType.ENTITY_FIELD, 'filter.key-type.entity-field'], + [EntityKeyType.CONSTANT, 'filter.key-type.constant'] ] ); @@ -344,12 +346,14 @@ export interface KeyFilterPredicateInfo { export interface KeyFilter { key: EntityKey; valueType: EntityKeyValueType; + value: string | number | boolean; predicate: KeyFilterPredicate; } export interface KeyFilterInfo { key: EntityKey; valueType: EntityKeyValueType; + value: string | number | boolean; predicates: Array; } @@ -466,6 +470,7 @@ export function keyFilterInfosToKeyFilters(keyFilterInfos: Array) const keyFilter: KeyFilter = { key, valueType: keyFilterInfo.valueType, + value: keyFilterInfo.value, predicate: keyFilterPredicateInfoToKeyFilterPredicate(predicate) }; keyFilters.push(keyFilter); @@ -486,6 +491,7 @@ export function keyFiltersToKeyFilterInfos(keyFilters: Array): Array< keyFilterInfo = { key, valueType: keyFilter.valueType, + value: keyFilter.value, predicates: [] }; keyFilterInfoMap[infoKey] = keyFilterInfo; @@ -508,6 +514,7 @@ export function filterInfoToKeyFilters(filter: FilterInfo): Array { const keyFilter: KeyFilter = { key, valueType: keyFilterInfo.valueType, + value: keyFilterInfo.value, predicate: keyFilterPredicateInfoToKeyFilterPredicate(predicate) }; keyFilters.push(keyFilter); diff --git a/ui-ngx/src/assets/locale/locale.constant-en_US.json b/ui-ngx/src/assets/locale/locale.constant-en_US.json index dff58d17d0..216ae2c1b7 100644 --- a/ui-ngx/src/assets/locale/locale.constant-en_US.json +++ b/ui-ngx/src/assets/locale/locale.constant-en_US.json @@ -1564,7 +1564,8 @@ "key-type": "Key type", "attribute": "Attribute", "timeseries": "Timeseries", - "entity-field": "Entity field" + "entity-field": "Entity field", + "constant": "Constant" }, "value-type": { "value-type": "Value type", From 594a1290db0e08d2e1f368557ec8dda4f061687e Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Fri, 26 Feb 2021 09:24:10 +0200 Subject: [PATCH 35/36] Async timeout connector customizer. Refactoring --- .../service/queue/TbCoreConsumerStats.java | 39 +++++++------------ .../transport/http/HttpTransportContext.java | 14 +++++++ 2 files changed, 29 insertions(+), 24 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java index d342b4b565..bca5a90fa8 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java @@ -55,31 +55,22 @@ public class TbCoreConsumerStats { public TbCoreConsumerStats(StatsFactory statsFactory) { String statsKey = StatsType.CORE.getName(); - this.totalCounter = statsFactory.createStatsCounter(statsKey, TOTAL_MSGS); - this.sessionEventCounter = statsFactory.createStatsCounter(statsKey, SESSION_EVENTS); - this.getAttributesCounter = statsFactory.createStatsCounter(statsKey, GET_ATTRIBUTE); - this.subscribeToAttributesCounter = statsFactory.createStatsCounter(statsKey, ATTRIBUTE_SUBSCRIBES); - this.subscribeToRPCCounter = statsFactory.createStatsCounter(statsKey, RPC_SUBSCRIBES); - this.toDeviceRPCCallResponseCounter = statsFactory.createStatsCounter(statsKey, TO_DEVICE_RPC_CALL_RESPONSES); - this.subscriptionInfoCounter = statsFactory.createStatsCounter(statsKey, SUBSCRIPTION_INFO); - this.claimDeviceCounter = statsFactory.createStatsCounter(statsKey, DEVICE_CLAIMS); - this.deviceStateCounter = statsFactory.createStatsCounter(statsKey, DEVICE_STATES); - this.subscriptionMsgCounter = statsFactory.createStatsCounter(statsKey, SUBSCRIPTION_MSGS); - this.toCoreNotificationsCounter = statsFactory.createStatsCounter(statsKey, TO_CORE_NOTIFICATIONS); - - - counters.add(totalCounter); - counters.add(sessionEventCounter); - counters.add(getAttributesCounter); - counters.add(subscribeToAttributesCounter); - counters.add(subscribeToRPCCounter); - counters.add(toDeviceRPCCallResponseCounter); - counters.add(subscriptionInfoCounter); - counters.add(claimDeviceCounter); + this.totalCounter = register(statsFactory.createStatsCounter(statsKey, TOTAL_MSGS)); + this.sessionEventCounter = register(statsFactory.createStatsCounter(statsKey, SESSION_EVENTS)); + this.getAttributesCounter = register(statsFactory.createStatsCounter(statsKey, GET_ATTRIBUTE)); + this.subscribeToAttributesCounter = register(statsFactory.createStatsCounter(statsKey, ATTRIBUTE_SUBSCRIBES)); + this.subscribeToRPCCounter = register(statsFactory.createStatsCounter(statsKey, RPC_SUBSCRIBES)); + this.toDeviceRPCCallResponseCounter = register(statsFactory.createStatsCounter(statsKey, TO_DEVICE_RPC_CALL_RESPONSES)); + this.subscriptionInfoCounter = register(statsFactory.createStatsCounter(statsKey, SUBSCRIPTION_INFO)); + this.claimDeviceCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_CLAIMS)); + this.deviceStateCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_STATES)); + this.subscriptionMsgCounter = register(statsFactory.createStatsCounter(statsKey, SUBSCRIPTION_MSGS)); + this.toCoreNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NOTIFICATIONS)); + } - counters.add(deviceStateCounter); - counters.add(subscriptionMsgCounter); - counters.add(toCoreNotificationsCounter); + private StatsCounter register(StatsCounter counter){ + counters.add(counter); + return counter; } public void log(TransportProtos.TransportToDeviceActorMsg msg) { diff --git a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/HttpTransportContext.java b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/HttpTransportContext.java index aad01c7c2f..fea07daffc 100644 --- a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/HttpTransportContext.java +++ b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/HttpTransportContext.java @@ -17,9 +17,13 @@ package org.thingsboard.server.transport.http; import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.apache.coyote.ProtocolHandler; +import org.apache.coyote.http11.Http11NioProtocol; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.web.embedded.tomcat.TomcatConnectorCustomizer; +import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; import org.thingsboard.server.common.transport.TransportContext; @@ -37,4 +41,14 @@ public class HttpTransportContext extends TransportContext { @Value("${transport.http.request_timeout}") private long defaultTimeout; + @Bean + public TomcatConnectorCustomizer tomcatAsyncTimeoutConnectorCustomizer() { + return connector -> { + ProtocolHandler handler = connector.getProtocolHandler(); + if (handler instanceof Http11NioProtocol) { + log.trace("Setting async timeout {}", defaultTimeout); + connector.setAsyncTimeout(defaultTimeout); + } + }; + } } From eec8bb0202052275ac4591d216adfa3da9adcd7c Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Fri, 26 Feb 2021 18:22:30 +0200 Subject: [PATCH 36/36] Introduced new configuration option --- application/src/main/resources/thingsboard.yml | 1 + .../server/transport/http/HttpTransportContext.java | 8 ++++++-- transport/http/src/main/resources/tb-http-transport.yml | 1 + 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 5983a606e1..7097acd4dd 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -545,6 +545,7 @@ transport: http: enabled: "${HTTP_ENABLED:true}" request_timeout: "${HTTP_REQUEST_TIMEOUT:60000}" + max_request_timeout: "${HTTP_MAX_REQUEST_TIMEOUT:300000}" # Local MQTT transport parameters mqtt: # Enable/disable mqtt transport protocol. diff --git a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/HttpTransportContext.java b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/HttpTransportContext.java index fea07daffc..13f38bfc80 100644 --- a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/HttpTransportContext.java +++ b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/HttpTransportContext.java @@ -41,13 +41,17 @@ public class HttpTransportContext extends TransportContext { @Value("${transport.http.request_timeout}") private long defaultTimeout; + @Getter + @Value("${transport.http.max_request_timeout}") + private long maxRequestTimeout; + @Bean public TomcatConnectorCustomizer tomcatAsyncTimeoutConnectorCustomizer() { return connector -> { ProtocolHandler handler = connector.getProtocolHandler(); if (handler instanceof Http11NioProtocol) { - log.trace("Setting async timeout {}", defaultTimeout); - connector.setAsyncTimeout(defaultTimeout); + log.trace("Setting async max request timeout {}", maxRequestTimeout); + connector.setAsyncTimeout(maxRequestTimeout); } }; } diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index 7c98fa46dd..585bf50483 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -39,6 +39,7 @@ zk: transport: http: request_timeout: "${HTTP_REQUEST_TIMEOUT:60000}" + max_request_timeout: "${HTTP_MAX_REQUEST_TIMEOUT:300000}" sessions: inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}" report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"