dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improvement to handling of subscription updates

Browse Source
This commit is contained in:
Andrii Shvaika 2020-11-16 10:52:18 +02:00
parent 9f2bae9815
commit a87a3d684e
2 changed files with 8 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java
View File

@ -21,12 +21,6 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.context.event.EventListener; import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.EntityView;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityViewId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.dao.entityview.EntityViewService;
import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent; import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent;
import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
@ -36,7 +30,6 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.queue.TbClusterService;
import org.thingsboard.server.service.telemetry.DefaultTelemetryWebSocketService;
import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate; import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate;
@ -49,7 +42,6 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.stream.Collectors;
@Slf4j @Slf4j
@TbCoreComponent @TbCoreComponent
@ -59,9 +51,6 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
private final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet(); private final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();
private final Map<String, Map<Integer, TbSubscription>> subscriptionsBySessionId = new ConcurrentHashMap<>(); private final Map<String, Map<Integer, TbSubscription>> subscriptionsBySessionId = new ConcurrentHashMap<>();
@Autowired
private EntityViewService entityViewService;
@Autowired @Autowired
private PartitionService partitionService; private PartitionService partitionService;
@ -72,17 +61,17 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
@Lazy @Lazy
private SubscriptionManagerService subscriptionManagerService; private SubscriptionManagerService subscriptionManagerService;
private ExecutorService wsCallBackExecutor; private ExecutorService subscriptionUpdateExecutor;
@PostConstruct @PostConstruct
public void initExecutor() { public void initExecutor() {
wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ws-sub-callback")); subscriptionUpdateExecutor = Executors.newWorkStealingPool(20);
} }
@PreDestroy @PreDestroy
public void shutdownExecutor() { public void shutdownExecutor() {
if (wsCallBackExecutor != null) { if (subscriptionUpdateExecutor != null) {
wsCallBackExecutor.shutdownNow(); subscriptionUpdateExecutor.shutdownNow();
} }
} }
@ -148,7 +137,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
update.getLatestValues().forEach((key, value) -> attrSub.getKeyStates().put(key, value)); update.getLatestValues().forEach((key, value) -> attrSub.getKeyStates().put(key, value));
break; break;
} }
subscription.getUpdateConsumer().accept(sessionId, update); subscriptionUpdateExecutor.submit(() -> subscription.getUpdateConsumer().accept(sessionId, update));
} }
callback.onSuccess(); callback.onSuccess();
} }
@ -158,7 +147,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
TbSubscription subscription = subscriptionsBySessionId TbSubscription subscription = subscriptionsBySessionId
.getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId()); .getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId());
if (subscription != null && subscription.getType() == TbSubscriptionType.ALARMS) { if (subscription != null && subscription.getType() == TbSubscriptionType.ALARMS) {
subscription.getUpdateConsumer().accept(sessionId, update); subscriptionUpdateExecutor.submit(() -> subscription.getUpdateConsumer().accept(sessionId, update));
} }
callback.onSuccess(); callback.onSuccess();
} }

4
ui-ngx/src/assets/locale/locale.constant-en_US.json
View File

@ -941,8 +941,8 @@
"provision-strategy": "Provision strategy", "provision-strategy": "Provision strategy",
"provision-strategy-required": "Provision strategy is required.", "provision-strategy-required": "Provision strategy is required.",
"provision-strategy-disabled": "Disabled", "provision-strategy-disabled": "Disabled",
"provision-strategy-created-new": "Allow create new devices", "provision-strategy-created-new": "Allow to create new devices",
"provision-strategy-check-pre-provisioned": "Check pre provisioned devices", "provision-strategy-check-pre-provisioned": "Check for pre-provisioned devices",
"provision-device-key": "Provision device key", "provision-device-key": "Provision device key",
"provision-device-key-required": "Provision device key is required.", "provision-device-key-required": "Provision device key is required.",
"provision-device-secret": "Provision device secret", "provision-device-secret": "Provision device secret",