diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 6b5dfbf6c3..11d556c819 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -224,7 +224,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } void process(ActorContext context, TransportToDeviceActorMsgWrapper wrapper) { - boolean reportDeviceActivity = false; + boolean reportDeviceActivity = true; TransportToDeviceActorMsg msg = wrapper.getMsg(); TbMsgCallback callback = wrapper.getCallback(); if (msg.hasSessionEvent()) { @@ -263,7 +263,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { callback.onSuccess(); } - //TODO 2.5 move this as a notification to the queue; private void reportLogicalDeviceActivity() { systemContext.getDeviceStateService().onDeviceActivity(deviceId); } diff --git a/application/src/main/java/org/thingsboard/server/actors/service/ActorService.java b/application/src/main/java/org/thingsboard/server/actors/service/ActorService.java index 890128e645..8adc6e8ac2 100644 --- a/application/src/main/java/org/thingsboard/server/actors/service/ActorService.java +++ b/application/src/main/java/org/thingsboard/server/actors/service/ActorService.java @@ -19,6 +19,7 @@ import org.thingsboard.server.common.data.id.DeviceId; 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.TbActorMsg; import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; import org.thingsboard.server.common.transport.SessionMsgProcessor; @@ -26,7 +27,7 @@ public interface ActorService extends SessionMsgProcessor { void onEntityStateChange(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state); - void onMsg(SendToClusterMsg msg); + void onMsg(TbActorMsg msg); void onCredentialsUpdate(TenantId tenantId, DeviceId deviceId); 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 fc9eb537c1..cb681182f6 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 @@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.id.DeviceId; 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.TbActorMsg; import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; import org.thingsboard.server.common.msg.cluster.ToAllNodesMsg; @@ -108,7 +109,7 @@ public class DefaultActorService implements ActorService { } @Override - public void onMsg(SendToClusterMsg msg) { + public void onMsg(TbActorMsg msg) { appActor.tell(msg, ActorRef.noSender()); } 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 07b7704d06..27d5453add 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 @@ -24,6 +24,7 @@ import org.springframework.context.event.EventListener; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.queue.common.TbProtoQueueMsg; @@ -34,6 +35,10 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; import org.thingsboard.server.queue.provider.TbCoreQueueProvider; import org.thingsboard.server.service.state.DeviceStateService; +import org.thingsboard.server.service.subscription.LocalSubscriptionService; +import org.thingsboard.server.service.subscription.SubscriptionManagerService; +import org.thingsboard.server.service.subscription.TbSubscriptionUtils; +import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate; import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; import javax.annotation.PostConstruct; @@ -62,15 +67,19 @@ public class DefaultTbCoreConsumerService implements TbCoreConsumerService { private final ActorSystemContext actorContext; private final DeviceStateService stateService; + private final LocalSubscriptionService localSubscriptionService; + private final SubscriptionManagerService subscriptionManagerService; private final TbQueueConsumer> consumer; private final TbCoreConsumerStats stats = new TbCoreConsumerStats(); private volatile ExecutorService mainConsumerExecutor; private volatile boolean stopped = false; - public DefaultTbCoreConsumerService(TbCoreQueueProvider tbCoreQueueProvider, ActorSystemContext actorContext, DeviceStateService stateService) { + public DefaultTbCoreConsumerService(TbCoreQueueProvider tbCoreQueueProvider, ActorSystemContext actorContext, DeviceStateService stateService, LocalSubscriptionService localSubscriptionService, SubscriptionManagerService subscriptionManagerService) { this.consumer = tbCoreQueueProvider.getToCoreMsgConsumer(); this.actorContext = actorContext; this.stateService = stateService; + this.localSubscriptionService = localSubscriptionService; + this.subscriptionManagerService = subscriptionManagerService; } @PostConstruct @@ -108,8 +117,15 @@ public class DefaultTbCoreConsumerService implements TbCoreConsumerService { } else if (toCoreMsg.hasDeviceStateServiceMsg()) { log.trace("[{}] Forwarding message to state service {}", id, toCoreMsg.getDeviceStateServiceMsg()); forwardToStateService(toCoreMsg.getDeviceStateServiceMsg(), callback); + } else if (toCoreMsg.hasToSubscriptionMgrMsg()) { + log.trace("[{}] Forwarding message to subscription manager service {}", id, toCoreMsg.getToSubscriptionMgrMsg()); + forwardToSubMgrService(toCoreMsg.getToSubscriptionMgrMsg(), callback); + } else if (toCoreMsg.hasToLocalSubscriptionServiceMsg()) { + log.trace("[{}] Forwarding message to local subscription service {}", id, toCoreMsg.getToLocalSubscriptionServiceMsg()); + forwardToLocalSubMgrService(toCoreMsg.getToLocalSubscriptionServiceMsg(), callback); } } catch (Throwable e) { + log.warn("[{}] Failed to process message: {}", id, msg, e); callback.onFailure(e); } }); @@ -126,23 +142,10 @@ public class DefaultTbCoreConsumerService implements TbCoreConsumerService { } } } + log.info("Tb Core Consumer stopped."); }); } - private void forwardToStateService(TransportProtos.DeviceStateServiceMsgProto deviceStateServiceMsg, TbMsgCallback callback) { - if (statsEnabled) { - stats.log(deviceStateServiceMsg); - } - stateService.onQueueMsg(deviceStateServiceMsg, callback); - } - - private void forwardToDeviceActor(TransportToDeviceActorMsg toDeviceActorMsg, TbMsgCallback callback) { - if (statsEnabled) { - stats.log(toDeviceActorMsg); - } - actorContext.getAppActor().tell(new TransportToDeviceActorMsgWrapper(toDeviceActorMsg, callback), ActorRef.noSender()); - } - @Scheduled(fixedDelayString = "${queue.core.stats.print_interval_ms}") public void printStats() { if (statsEnabled) { @@ -161,4 +164,55 @@ public class DefaultTbCoreConsumerService implements TbCoreConsumerService { } } + private void forwardToLocalSubMgrService(TransportProtos.LocalSubscriptionServiceMsgProto msg, TbMsgCallback callback) { + if (msg.hasSubUpdate()) { + localSubscriptionService.onSubscriptionUpdate(msg.getSubUpdate().getSessionId(), TbSubscriptionUtils.fromProto(msg.getSubUpdate()), callback); + } else { + throwNotHandled(msg, callback); + } + } + + private void forwardToSubMgrService(TransportProtos.SubscriptionMgrMsgProto msg, TbMsgCallback callback) { + if (msg.hasAttributeSub()) { + subscriptionManagerService.addSubscription(TbSubscriptionUtils.fromProto(msg.getAttributeSub()), callback); + } else if (msg.hasTelemetrySub()) { + subscriptionManagerService.addSubscription(TbSubscriptionUtils.fromProto(msg.getTelemetrySub()), callback); + } else if (msg.hasSubClose()) { + TransportProtos.TbSubscriptionCloseProto closeProto = msg.getSubClose(); + subscriptionManagerService.cancelSubscription(closeProto.getSessionId(), closeProto.getSubscriptionId(), callback); + } else if (msg.hasTsUpdate()) { + TransportProtos.TbTimeSeriesUpdateProto proto = msg.getTsUpdate(); + subscriptionManagerService.onTimeseriesDataUpdate( + new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())), + TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()), + TbSubscriptionUtils.toTsKvEntityList(proto.getDataList()), callback); + } else if (msg.hasAttrUpdate()) { + TransportProtos.TbAttributeUpdateProto proto = msg.getAttrUpdate(); + subscriptionManagerService.onAttributesUpdate( + new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())), + TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()), + proto.getScope(), TbSubscriptionUtils.toAttributeKvList(proto.getDataList()), callback); + } else { + throwNotHandled(msg, callback); + } + } + + private void forwardToStateService(TransportProtos.DeviceStateServiceMsgProto deviceStateServiceMsg, TbMsgCallback callback) { + if (statsEnabled) { + stats.log(deviceStateServiceMsg); + } + stateService.onQueueMsg(deviceStateServiceMsg, callback); + } + + private void forwardToDeviceActor(TransportToDeviceActorMsg toDeviceActorMsg, TbMsgCallback callback) { + if (statsEnabled) { + stats.log(toDeviceActorMsg); + } + actorContext.getAppActor().tell(new TransportToDeviceActorMsgWrapper(toDeviceActorMsg, callback), ActorRef.noSender()); + } + + private void throwNotHandled(Object msg, TbMsgCallback callback) { + log.warn("Message not handled: {}", msg); + callback.onFailure(new RuntimeException("Message not handled!")); + } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgCallback.java b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgCallback.java index 774377407a..85fec1d2ab 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgCallback.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgCallback.java @@ -17,6 +17,19 @@ package org.thingsboard.server.service.queue; public interface TbMsgCallback { + TbMsgCallback EMPTY = new TbMsgCallback() { + + @Override + public void onSuccess() { + + } + + @Override + public void onFailure(Throwable t) { + + } + }; + void onSuccess(); void onFailure(Throwable t); 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 5ee26bc294..51f544712e 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 @@ -5,7 +5,7 @@ * 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 + * 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, @@ -92,7 +92,6 @@ import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE; */ @Service @Slf4j -//TODO: refactor to use page links as cursor and not fetch all public class DefaultDeviceStateService implements DeviceStateService { private static final ObjectMapper json = new ObjectMapper(); @@ -150,7 +149,6 @@ public class DefaultDeviceStateService implements DeviceStateService { private volatile boolean clusterUpdatePending = false; private ListeningScheduledExecutorService queueExecutor; - private ConcurrentMap> tenantDevices = new ConcurrentHashMap<>(); private ConcurrentMap> partitionedDevices = new ConcurrentHashMap<>(); private ConcurrentMap deviceStates = new ConcurrentHashMap<>(); private ConcurrentMap deviceLastReportedActivity = new ConcurrentHashMap<>(); @@ -378,7 +376,6 @@ public class DefaultDeviceStateService implements DeviceStateService { deviceStates.put(state.getDeviceId(), state); } - //TODO 2.5: review this method private void updateState() { long ts = System.currentTimeMillis(); Set deviceIds = new HashSet<>(deviceStates.keySet()); @@ -439,13 +436,9 @@ public class DefaultDeviceStateService implements DeviceStateService { deviceStates.remove(deviceId); deviceLastReportedActivity.remove(deviceId); deviceLastSavedActivity.remove(deviceId); - Set deviceIds = tenantDevices.get(tenantId); - if (deviceIds != null) { - deviceIds.remove(deviceId); - if (deviceIds.isEmpty()) { - tenantDevices.remove(tenantId); - } - } + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId); + Set deviceIdSet = partitionedDevices.get(tpi); + deviceIdSet.remove(deviceId); } private ListenableFuture fetchDeviceState(Device device) { diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultLocalSubscriptionService.java new file mode 100644 index 0000000000..f9d60b37a5 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultLocalSubscriptionService.java @@ -0,0 +1,230 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Lazy; +import org.springframework.context.event.EventListener; +import org.springframework.stereotype.Service; +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.queue.TbQueueProducer; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent; +import org.thingsboard.server.queue.discovery.PartitionChangeEvent; +import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.ServiceType; +import org.thingsboard.server.queue.discovery.TopicPartitionInfo; +import org.thingsboard.server.queue.provider.TbCoreQueueProvider; +import org.thingsboard.server.service.queue.TbMsgCallback; +import org.thingsboard.server.service.telemetry.TelemetryWebSocketService; +import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import java.util.Collections; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.stream.Collectors; + +@Slf4j +@Service +public class DefaultLocalSubscriptionService implements LocalSubscriptionService { + + private final Set currentPartitions = ConcurrentHashMap.newKeySet(); + private final Map> subscriptionsBySessionId = new ConcurrentHashMap<>(); + + @Autowired + private TelemetryWebSocketService wsService; + + @Autowired + private EntityViewService entityViewService; + + @Autowired + private PartitionService partitionService; + + @Autowired + private TbCoreQueueProvider coreQueueProvider; + + @Autowired + @Lazy + private SubscriptionManagerService subscriptionManagerService; + + private ExecutorService wsCallBackExecutor; + private TbQueueProducer> toCoreProducer; + + @PostConstruct + public void initExecutor() { + wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ws-sub-callback")); + toCoreProducer = coreQueueProvider.getTbCoreMsgProducer(); + } + + @PreDestroy + public void shutdownExecutor() { + if (wsCallBackExecutor != null) { + wsCallBackExecutor.shutdownNow(); + } + } + + @Override + @EventListener(PartitionChangeEvent.class) + public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceKey().getServiceType())) { + currentPartitions.clear(); + currentPartitions.addAll(partitionChangeEvent.getPartitions()); + } + } + + @Override + @EventListener(ClusterTopologyChangeEvent.class) + public void onApplicationEvent(ClusterTopologyChangeEvent event) { + if (event.getServiceKeys().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, false))); + } + } + + //TODO 3.1: replace null callbacks with callbacks from websocket service. + @Override + public void addSubscription(TbSubscription subscription) { + EntityId entityId = subscription.getEntityId(); + // Telemetry subscription on Entity Views are handled differently, because we need to allow only certain keys and time ranges; + if (entityId.getEntityType().equals(EntityType.ENTITY_VIEW) && TbSubscriptionType.TIMESERIES.equals(subscription.getType())) { + subscription = resolveEntityViewSubscription((TbTimeseriesSubscription) subscription); + } + pushSubscriptionToManagerService(subscription, true); + registerSubscription(subscription); + } + + private void pushSubscriptionToManagerService(TbSubscription subscription, boolean pushToLocalService) { + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId()); + if (currentPartitions.contains(tpi)) { + // Subscription is managed on the same server; + if (pushToLocalService) { + subscriptionManagerService.addSubscription(subscription, TbMsgCallback.EMPTY); + } + } else { + // Push to the queue; + TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toNewSubscriptionProto(subscription); + toCoreProducer.send(tpi, new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg), null); + } + } + + @Override + public void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbMsgCallback callback) { + TbSubscription subscription = subscriptionsBySessionId + .getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId()); + if (subscription != null) { + switch (subscription.getType()) { + case TIMESERIES: + TbTimeseriesSubscription tsSub = (TbTimeseriesSubscription) subscription; + update.getLatestValues().forEach((key, value) -> tsSub.getKeyStates().put(key, value)); + break; + case ATTRIBUTES: + TbAttributeSubscription attrSub = (TbAttributeSubscription) subscription; + update.getLatestValues().forEach((key, value) -> attrSub.getKeyStates().put(key, value)); + break; + } + wsService.sendWsMsg(sessionId, update); + } + callback.onSuccess(); + } + + @Override + public void cancelSubscription(String sessionId, int subscriptionId) { + log.debug("[{}][{}] Going to remove subscription.", sessionId, subscriptionId); + Map sessionSubscriptions = subscriptionsBySessionId.get(sessionId); + if (sessionSubscriptions != null) { + TbSubscription subscription = sessionSubscriptions.remove(subscriptionId); + if (subscription != null) { + if (sessionSubscriptions.isEmpty()) { + subscriptionsBySessionId.remove(sessionId); + } + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId()); + if (currentPartitions.contains(tpi)) { + // Subscription is managed on the same server; + subscriptionManagerService.cancelSubscription(sessionId, subscriptionId, TbMsgCallback.EMPTY); + } else { + // Push to the queue; + TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toCloseSubscriptionProto(subscription); + toCoreProducer.send(tpi, new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg), null); + } + } else { + log.debug("[{}][{}] Subscription not found!", sessionId, subscriptionId); + } + } else { + log.debug("[{}] No session subscriptions found!", sessionId); + } + } + + @Override + public void cancelAllSessionSubscriptions(String sessionId) { + Map subscriptions = subscriptionsBySessionId.get(sessionId); + if (subscriptions != null) { + Set toRemove = new HashSet<>(subscriptions.keySet()); + toRemove.forEach(id -> cancelSubscription(sessionId, id)); + } + } + + private TbSubscription resolveEntityViewSubscription(TbTimeseriesSubscription subscription) { + EntityView entityView = entityViewService.findEntityViewById(TenantId.SYS_TENANT_ID, new EntityViewId(subscription.getEntityId().getId())); + + Map keyStates; + if (subscription.isAllKeys()) { + keyStates = entityView.getKeys().getTimeseries().stream().collect(Collectors.toMap(k -> k, k -> 0L)); + } else { + keyStates = subscription.getKeyStates().entrySet() + .stream().filter(entry -> entityView.getKeys().getTimeseries().contains(entry.getKey())) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + } + + return TbTimeseriesSubscription.builder() + .serviceId(subscription.getServiceId()) + .sessionId(subscription.getSessionId()) + .subscriptionId(subscription.getSubscriptionId()) + .tenantId(subscription.getTenantId()) + .entityId(entityView.getEntityId()) + .startTime(entityView.getStartTimeMs()) + .endTime(entityView.getEndTimeMs()) + .allKeys(false) + .keyStates(keyStates).build(); + } + + private void registerSubscription(TbSubscription subscription) { + Map sessionSubscriptions = subscriptionsBySessionId.computeIfAbsent(subscription.getSessionId(), k -> new ConcurrentHashMap<>()); + sessionSubscriptions.put(subscription.getSubscriptionId(), subscription); + } + +} 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 new file mode 100644 index 0000000000..267ea14ac1 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -0,0 +1,377 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; +import org.thingsboard.common.util.DonAsynchron; +import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; +import org.thingsboard.server.actors.service.ActorService; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.Aggregation; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.ReadTsKvQuery; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; +import org.thingsboard.server.dao.attributes.AttributesService; +import org.thingsboard.server.dao.timeseries.TimeseriesService; +import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionUpdateProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionUpdateValueListProto; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +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.ServiceType; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.discovery.TopicPartitionInfo; +import org.thingsboard.server.queue.provider.TbCoreQueueProvider; +import org.thingsboard.server.service.queue.TbMsgCallback; +import org.thingsboard.server.service.state.DefaultDeviceStateService; +import org.thingsboard.server.service.state.DeviceStateService; +import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.function.Function; +import java.util.function.Predicate; + +@Slf4j +@Service +public class DefaultSubscriptionManagerService implements SubscriptionManagerService { + + @Autowired + private AttributesService attrService; + + @Autowired + private TimeseriesService tsService; + + @Autowired + private PartitionService partitionService; + + @Autowired + private TbServiceInfoProvider serviceInfoProvider; + + @Autowired + private TbCoreQueueProvider coreQueueProvider; + + @Autowired + private LocalSubscriptionService localSubscriptionService; + + @Autowired + private DeviceStateService deviceStateService; + + @Autowired + private ActorService actorService; + + private final Map> subscriptionsByEntityId = new ConcurrentHashMap<>(); + private final Map> subscriptionsByWsSessionId = new ConcurrentHashMap<>(); + private final ConcurrentMap> partitionedSubscriptions = new ConcurrentHashMap<>(); + private final Set currentPartitions = ConcurrentHashMap.newKeySet(); + + private ExecutorService tsCallBackExecutor; + private String serviceId; + private TbQueueProducer> toCoreProducer; + + @PostConstruct + public void initExecutor() { + tsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ts-sub-callback")); + serviceId = serviceInfoProvider.getServiceId(); + toCoreProducer = coreQueueProvider.getTbCoreMsgProducer(); + } + + @PreDestroy + public void shutdownExecutor() { + if (tsCallBackExecutor != null) { + tsCallBackExecutor.shutdownNow(); + } + } + + @Override + public void addSubscription(TbSubscription subscription, TbMsgCallback callback) { + log.trace("[{}][{}][{}] Registering remote subscription for entity [{}]", + subscription.getServiceId(), subscription.getSessionId(), subscription.getSubscriptionId(), subscription.getEntityId()); + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId()); + if (currentPartitions.contains(tpi)) { + partitionedSubscriptions.computeIfAbsent(tpi, k -> ConcurrentHashMap.newKeySet()).add(subscription); + callback.onSuccess(); + } else { + log.warn("[{}][{}] Entity belongs to external partition. Probably rebalancing is in progress. Topic: {}" + , subscription.getTenantId(), subscription.getEntityId(), tpi.getFullTopicName()); + callback.onFailure(new RuntimeException("Entity belongs to external partition " + tpi.getFullTopicName() + "!")); + } + boolean newSubscription = subscriptionsByEntityId + .computeIfAbsent(subscription.getEntityId(), k -> ConcurrentHashMap.newKeySet()).add(subscription); + subscriptionsByWsSessionId.computeIfAbsent(subscription.getSessionId(), k -> new ConcurrentHashMap<>()).put(subscription.getSubscriptionId(), subscription); + if (newSubscription) { + switch (subscription.getType()) { + case TIMESERIES: + handleNewTelemetrySubscription((TbTimeseriesSubscription) subscription); + break; + case ATTRIBUTES: + handleNewAttributeSubscription((TbAttributeSubscription) subscription); + break; + } + } + } + + @Override + public void cancelSubscription(String sessionId, int subscriptionId, TbMsgCallback callback) { + log.debug("[{}][{}] Going to remove subscription.", sessionId, subscriptionId); + Map sessionSubscriptions = subscriptionsByWsSessionId.get(sessionId); + if (sessionSubscriptions != null) { + TbSubscription subscription = sessionSubscriptions.remove(subscriptionId); + if (subscription != null) { + removeSubscriptionFromEntityMap(subscription); + removeSubscriptionFromPartitionMap(subscription); + if (sessionSubscriptions.isEmpty()) { + subscriptionsByWsSessionId.remove(sessionId); + } + } else { + log.debug("[{}][{}] Subscription not found!", sessionId, subscriptionId); + } + } else { + log.debug("[{}] No session subscriptions found!", sessionId); + } + callback.onSuccess(); + } + + @Override + public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + Set removedPartitions = new HashSet<>(currentPartitions); + removedPartitions.removeAll(partitionChangeEvent.getPartitions()); + + currentPartitions.clear(); + currentPartitions.addAll(partitionChangeEvent.getPartitions()); + + // We no longer manage current partition of devices; + removedPartitions.forEach(partition -> { + Set subs = partitionedSubscriptions.remove(partition); + if (subs != null) { + subs.forEach(this::removeSubscriptionFromEntityMap); + } + }); + } + + @Override + public void onTimeseriesDataUpdate(TenantId tenantId, EntityId entityId, List ts, TbMsgCallback callback) { + onLocalSubUpdate(entityId, + s -> { + if (TbSubscriptionType.TIMESERIES.equals(s.getType())) { + return (TbTimeseriesSubscription) s; + } else { + return null; + } + }, s -> true, s -> { + List subscriptionUpdate = null; + for (TsKvEntry kv : ts) { + if (isInTimeRange(s, kv.getTs()) && (s.isAllKeys() || s.getKeyStates().containsKey((kv.getKey())))) { + if (subscriptionUpdate == null) { + subscriptionUpdate = new ArrayList<>(); + } + subscriptionUpdate.add(kv); + } + } + return subscriptionUpdate; + }); + callback.onSuccess(); + } + + @Override + public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, TbMsgCallback callback) { + onLocalSubUpdate(entityId, + s -> { + if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) { + return (TbAttributeSubscription) s; + } else { + return null; + } + }, + s -> (StringUtils.isEmpty(s.getScope()) || scope.equals(s.getScope().name())), + s -> { + List subscriptionUpdate = null; + for (AttributeKvEntry kv : attributes) { + if (s.isAllKeys() || s.getKeyStates().containsKey(kv.getKey())) { + if (subscriptionUpdate == null) { + subscriptionUpdate = new ArrayList<>(); + } + subscriptionUpdate.add(new BasicTsKvEntry(kv.getLastUpdateTs(), kv)); + } + } + return subscriptionUpdate; + }); + if (entityId.getEntityType() == EntityType.DEVICE) { + if (TbAttributeSubscriptionScope.SERVER_SCOPE.name().equalsIgnoreCase(scope)) { + for (AttributeKvEntry attribute : attributes) { + if (attribute.getKey().equals(DefaultDeviceStateService.INACTIVITY_TIMEOUT)) { + deviceStateService.onDeviceInactivityTimeoutUpdate(new DeviceId(entityId.getId()), attribute.getLongValue().orElse(0L)); + } + } + } else if (TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope)) { + DeviceAttributesEventNotificationMsg notificationMsg = DeviceAttributesEventNotificationMsg.onUpdate(tenantId, + new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, new ArrayList<>(attributes)); + actorService.onMsg(notificationMsg); + } + } + callback.onSuccess(); + } + + private void onLocalSubUpdate(EntityId entityId, + Function castFunction, + Predicate filterFunction, + Function> processFunction) { + Set entitySubscriptions = subscriptionsByEntityId.get(entityId); + if (entitySubscriptions != null) { + entitySubscriptions.stream().map(castFunction).filter(Objects::nonNull).filter(filterFunction).forEach(s -> { + List subscriptionUpdate = processFunction.apply(s); + if (subscriptionUpdate != null && !subscriptionUpdate.isEmpty()) { + if (serviceId.equals(s.getServiceId())) { + SubscriptionUpdate update = new SubscriptionUpdate(s.getSubscriptionId(), subscriptionUpdate); + localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbMsgCallback.EMPTY); + } else { + TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); + toCoreProducer.send(tpi, toProto(s, subscriptionUpdate), null); + } + } + }); + } else { + log.debug("[{}] No device subscriptions to process!", entityId); + } + } + + private boolean isInTimeRange(TbTimeseriesSubscription subscription, long kvTime) { + return (subscription.getStartTime() == 0 || subscription.getStartTime() <= kvTime) + && (subscription.getEndTime() == 0 || subscription.getEndTime() >= kvTime); + } + + private void removeSubscriptionFromEntityMap(TbSubscription sub) { + Set entitySubSet = subscriptionsByEntityId.get(sub.getEntityId()); + if (entitySubSet != null) { + entitySubSet.remove(sub); + if (entitySubSet.isEmpty()) { + subscriptionsByEntityId.remove(sub.getEntityId()); + } + } + } + + private void removeSubscriptionFromPartitionMap(TbSubscription sub) { + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, sub.getTenantId(), sub.getEntityId()); + Set subs = partitionedSubscriptions.get(tpi); + if (subs != null) { + subs.remove(sub); + } + } + + private void handleNewAttributeSubscription(TbAttributeSubscription subscription) { + log.trace("[{}][{}][{}] Processing remote attribute subscription for entity [{}]", + serviceId, subscription.getSessionId(), subscription.getSubscriptionId(), subscription.getEntityId()); + + final Map keyStates = subscription.getKeyStates(); + DonAsynchron.withCallback(attrService.find(subscription.getTenantId(), subscription.getEntityId(), DataConstants.CLIENT_SCOPE, keyStates.keySet()), values -> { + List missedUpdates = new ArrayList<>(); + values.forEach(latestEntry -> { + if (latestEntry.getLastUpdateTs() > keyStates.get(latestEntry.getKey())) { + missedUpdates.add(new BasicTsKvEntry(latestEntry.getLastUpdateTs(), latestEntry)); + } + }); + if (!missedUpdates.isEmpty()) { + TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); + toCoreProducer.send(tpi, toProto(subscription, missedUpdates), null); + } + }, + e -> log.error("Failed to fetch missed updates.", e), tsCallBackExecutor); + } + + private void handleNewTelemetrySubscription(TbTimeseriesSubscription subscription) { + log.trace("[{}][{}][{}] Processing remote telemetry subscription for entity [{}]", + serviceId, subscription.getSessionId(), subscription.getSubscriptionId(), subscription.getEntityId()); + + long curTs = System.currentTimeMillis(); + List queries = new ArrayList<>(); + subscription.getKeyStates().forEach((key, value) -> { + if (curTs > value) { + long startTs = subscription.getStartTime() > 0 ? Math.max(subscription.getStartTime(), value + 1L) : (value + 1L); + long endTs = subscription.getEndTime() > 0 ? Math.min(subscription.getEndTime(), curTs) : curTs; + queries.add(new BaseReadTsKvQuery(key, startTs, endTs, 0, 1000, Aggregation.NONE)); + } + }); + if (!queries.isEmpty()) { + DonAsynchron.withCallback(tsService.findAll(subscription.getTenantId(), subscription.getEntityId(), queries), + missedUpdates -> { + if (missedUpdates != null && !missedUpdates.isEmpty()) { + TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); + toCoreProducer.send(tpi, toProto(subscription, missedUpdates), null); + } + }, + e -> log.error("Failed to fetch missed updates.", e), + tsCallBackExecutor); + } + } + + private TbProtoQueueMsg toProto(TbSubscription subscription, List updates) { + TbSubscriptionUpdateProto.Builder builder = TbSubscriptionUpdateProto.newBuilder(); + + builder.setSessionId(subscription.getSessionId()); + builder.setSubscriptionId(subscription.getSubscriptionId()); + + Map> data = new TreeMap<>(); + for (TsKvEntry tsEntry : updates) { + List values = data.computeIfAbsent(tsEntry.getKey(), k -> new ArrayList<>()); + Object[] value = new Object[2]; + value[0] = tsEntry.getTs(); + value[1] = tsEntry.getValueAsString(); + values.add(value); + } + + data.forEach((key, value) -> { + TbSubscriptionUpdateValueListProto.Builder dataBuilder = TbSubscriptionUpdateValueListProto.newBuilder(); + dataBuilder.setKey(key); + value.forEach(v -> { + Object[] array = (Object[]) v; + dataBuilder.addTs((long) array[0]); + dataBuilder.addValue((String) array[1]); + }); + builder.addData(dataBuilder.build()); + }); + + ToCoreMsg toCoreMsg = ToCoreMsg.newBuilder().setToLocalSubscriptionServiceMsg( + LocalSubscriptionServiceMsgProto.newBuilder().setSubUpdate(builder.build()).build()) + .build(); + return new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/LocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/LocalSubscriptionService.java new file mode 100644 index 0000000000..a80fb0db92 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/LocalSubscriptionService.java @@ -0,0 +1,36 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent; +import org.thingsboard.server.queue.discovery.PartitionChangeEvent; +import org.thingsboard.server.service.queue.TbMsgCallback; +import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate; + +public interface LocalSubscriptionService { + + void addSubscription(TbSubscription subscription); + + void cancelSubscription(String sessionId, int subscriptionId); + + void cancelAllSessionSubscriptions(String sessionId); + + void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbMsgCallback callback); + + void onApplicationEvent(PartitionChangeEvent event); + + void onApplicationEvent(ClusterTopologyChangeEvent event); +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java new file mode 100644 index 0000000000..bb236141f8 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +import org.springframework.context.ApplicationListener; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.queue.discovery.PartitionChangeEvent; +import org.thingsboard.server.service.queue.TbMsgCallback; + +import java.util.List; + +public interface SubscriptionManagerService extends ApplicationListener { + + void addSubscription(TbSubscription subscription, TbMsgCallback callback); + + void cancelSubscription(String sessionId, int subscriptionId, TbMsgCallback callback); + + void onTimeseriesDataUpdate(TenantId tenantId, EntityId entityId, List ts, TbMsgCallback callback); + + void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, TbMsgCallback callback); +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscription.java new file mode 100644 index 0000000000..83a86efefc --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscription.java @@ -0,0 +1,50 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +import lombok.Builder; +import lombok.Data; +import lombok.Getter; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.Map; + +public class TbAttributeSubscription extends TbSubscription { + + @Getter private final boolean allKeys; + @Getter private final Map keyStates; + @Getter private final TbAttributeSubscriptionScope scope; + + @Builder + public TbAttributeSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, + boolean allKeys, Map keyStates, TbAttributeSubscriptionScope scope) { + super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.ATTRIBUTES); + this.allKeys = allKeys; + this.keyStates = keyStates; + this.scope = scope; + } + + @Override + public boolean equals(Object o) { + return super.equals(o); + } + + @Override + public int hashCode() { + return super.hashCode(); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscriptionScope.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscriptionScope.java new file mode 100644 index 0000000000..d23cc3581d --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscriptionScope.java @@ -0,0 +1,22 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +public enum TbAttributeSubscriptionScope { + + CLIENT_SCOPE, SHARED_SCOPE, SERVER_SCOPE + +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscription.java new file mode 100644 index 0000000000..22b37ff690 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscription.java @@ -0,0 +1,52 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +import lombok.AllArgsConstructor; +import lombok.Data; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.Objects; + +@Data +@AllArgsConstructor +public abstract class TbSubscription { + + private final String serviceId; + private final String sessionId; + private final int subscriptionId; + private final TenantId tenantId; + private final EntityId entityId; + private final TbSubscriptionType type; + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + TbSubscription that = (TbSubscription) o; + return subscriptionId == that.subscriptionId && + sessionId.equals(that.sessionId) && + tenantId.equals(that.tenantId) && + entityId.equals(that.entityId) && + type == that.type; + } + + @Override + public int hashCode() { + return Objects.hash(sessionId, subscriptionId, tenantId, entityId, type); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionType.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionType.java new file mode 100644 index 0000000000..d7a4715460 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionType.java @@ -0,0 +1,20 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +public enum TbSubscriptionType { + TIMESERIES, ATTRIBUTES +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java new file mode 100644 index 0000000000..d0b4d713f9 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java @@ -0,0 +1,247 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.BooleanDataEntry; +import org.thingsboard.server.common.data.kv.DataType; +import org.thingsboard.server.common.data.kv.DoubleDataEntry; +import org.thingsboard.server.common.data.kv.JsonDataEntry; +import org.thingsboard.server.common.data.kv.KvEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; +import org.thingsboard.server.common.data.kv.StringDataEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto; +import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType; +import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeSubscriptionProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeUpdateProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionCloseProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionKetStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionUpdateProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesSubscriptionProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdateProto; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; +import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode; +import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.UUID; + +public class TbSubscriptionUtils { + + public static ToCoreMsg toNewSubscriptionProto(TbSubscription subscription) { + SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder(); + TbSubscriptionProto subscriptionProto = TbSubscriptionProto.newBuilder() + .setServiceId(subscription.getServiceId()) + .setSessionId(subscription.getSessionId()) + .setSubscriptionId(subscription.getSubscriptionId()) + .setTenantIdMSB(subscription.getTenantId().getId().getMostSignificantBits()) + .setTenantIdLSB(subscription.getTenantId().getId().getLeastSignificantBits()) + .setEntityType(subscription.getEntityId().getEntityType().name()) + .setEntityIdMSB(subscription.getEntityId().getId().getMostSignificantBits()) + .setEntityIdLSB(subscription.getEntityId().getId().getLeastSignificantBits()).build(); + + switch (subscription.getType()) { + case TIMESERIES: + TbTimeseriesSubscription tSub = (TbTimeseriesSubscription) subscription; + TbTimeSeriesSubscriptionProto.Builder tSubProto = TbTimeSeriesSubscriptionProto.newBuilder() + .setSub(subscriptionProto) + .setAllKeys(tSub.isAllKeys()); + tSub.getKeyStates().forEach((key, value) -> tSubProto.addKeyStates( + TbSubscriptionKetStateProto.newBuilder().setKey(key).setTs(value).build())); + tSubProto.setStartTime(tSub.getStartTime()); + tSubProto.setEndTime(tSub.getEndTime()); + msgBuilder.setTelemetrySub(tSubProto.build()); + break; + case ATTRIBUTES: + TbAttributeSubscription aSub = (TbAttributeSubscription) subscription; + TbAttributeSubscriptionProto.Builder aSubProto = TbAttributeSubscriptionProto.newBuilder() + .setSub(subscriptionProto) + .setAllKeys(aSub.isAllKeys()) + .setScope(aSub.getScope().name()); + aSub.getKeyStates().forEach((key, value) -> aSubProto.addKeyStates( + TbSubscriptionKetStateProto.newBuilder().setKey(key).setTs(value).build())); + msgBuilder.setAttributeSub(aSubProto.build()); + break; + } + return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build(); + } + + public static ToCoreMsg toCloseSubscriptionProto(TbSubscription subscription) { + SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder(); + TbSubscriptionCloseProto closeProto = TbSubscriptionCloseProto.newBuilder() + .setSessionId(subscription.getSessionId()) + .setSubscriptionId(subscription.getSubscriptionId()).build(); + msgBuilder.setSubClose(closeProto); + return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build(); + } + + public static TbSubscription fromProto(TbAttributeSubscriptionProto attributeSub) { + TbSubscriptionProto subProto = attributeSub.getSub(); + TbAttributeSubscription.TbAttributeSubscriptionBuilder builder = TbAttributeSubscription.builder() + .serviceId(subProto.getServiceId()) + .sessionId(subProto.getSessionId()) + .subscriptionId(subProto.getSubscriptionId()) + .entityId(EntityIdFactory.getByTypeAndUuid(subProto.getEntityType(), new UUID(subProto.getEntityIdMSB(), subProto.getEntityIdLSB()))) + .tenantId(new TenantId(new UUID(subProto.getTenantIdMSB(), subProto.getTenantIdLSB()))); + + builder.scope(TbAttributeSubscriptionScope.valueOf(attributeSub.getScope())); + builder.allKeys(attributeSub.getAllKeys()); + Map keyStates = new HashMap<>(); + attributeSub.getKeyStatesList().forEach(ksProto -> keyStates.put(ksProto.getKey(), ksProto.getTs())); + builder.keyStates(keyStates); + return builder.build(); + } + + public static TbSubscription fromProto(TbTimeSeriesSubscriptionProto telemetrySub) { + TbSubscriptionProto subProto = telemetrySub.getSub(); + TbTimeseriesSubscription.TbTimeseriesSubscriptionBuilder builder = TbTimeseriesSubscription.builder() + .serviceId(subProto.getServiceId()) + .sessionId(subProto.getSessionId()) + .subscriptionId(subProto.getSubscriptionId()) + .entityId(EntityIdFactory.getByTypeAndUuid(subProto.getEntityType(), new UUID(subProto.getEntityIdMSB(), subProto.getEntityIdLSB()))) + .tenantId(new TenantId(new UUID(subProto.getTenantIdMSB(), subProto.getTenantIdLSB()))); + + builder.allKeys(telemetrySub.getAllKeys()); + Map keyStates = new HashMap<>(); + telemetrySub.getKeyStatesList().forEach(ksProto -> keyStates.put(ksProto.getKey(), ksProto.getTs())); + builder.startTime(telemetrySub.getStartTime()); + builder.endTime(telemetrySub.getEndTime()); + builder.keyStates(keyStates); + return builder.build(); + } + + public static SubscriptionUpdate fromProto(TbSubscriptionUpdateProto proto) { + if (proto.getErrorCode() > 0) { + return new SubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg()); + } else { + Map> data = new TreeMap<>(); + proto.getDataList().forEach(v -> { + List values = data.computeIfAbsent(v.getKey(), k -> new ArrayList<>()); + for (int i = 0; i < v.getTsCount(); i++) { + Object[] value = new Object[2]; + value[0] = v.getTs(i); + value[1] = v.getValue(i); + values.add(value); + } + }); + return new SubscriptionUpdate(proto.getSubscriptionId(), data); + } + } + + public static ToCoreMsg toTimeseriesUpdateProto(TenantId tenantId, EntityId entityId, List ts) { + TbTimeSeriesUpdateProto.Builder builder = TbTimeSeriesUpdateProto.newBuilder(); + builder.setEntityType(entityId.getEntityType().name()); + builder.setEntityIdMSB(entityId.getId().getMostSignificantBits()); + builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits()); + builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); + builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); + ts.forEach(v -> builder.addData(toKeyValueProto(v.getTs(), v).build())); + SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder(); + msgBuilder.setTsUpdate(builder); + return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build(); + } + + public static ToCoreMsg toAttributesUpdateProto(TenantId tenantId, EntityId entityId, String scope, List attributes) { + TbAttributeUpdateProto.Builder builder = TbAttributeUpdateProto.newBuilder(); + builder.setEntityType(entityId.getEntityType().name()); + builder.setEntityIdMSB(entityId.getId().getMostSignificantBits()); + builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits()); + builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); + builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); + builder.setScope(scope); + attributes.forEach(v -> builder.addData(toKeyValueProto(v.getLastUpdateTs(), v).build())); + + SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder(); + msgBuilder.setAttrUpdate(builder); + return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build(); + } + + private static TsKvProto.Builder toKeyValueProto(long ts, KvEntry attr) { + KeyValueProto.Builder dataBuilder = KeyValueProto.newBuilder(); + dataBuilder.setKey(attr.getKey()); + dataBuilder.setType(KeyValueType.forNumber(attr.getDataType().ordinal())); + switch (attr.getDataType()) { + case BOOLEAN: + attr.getBooleanValue().ifPresent(dataBuilder::setBoolV); + break; + case LONG: + attr.getLongValue().ifPresent(dataBuilder::setLongV); + break; + case DOUBLE: + attr.getDoubleValue().ifPresent(dataBuilder::setDoubleV); + break; + case JSON: + attr.getJsonValue().ifPresent(dataBuilder::setJsonV); + break; + case STRING: + attr.getStrValue().ifPresent(dataBuilder::setStringV); + break; + } + return TsKvProto.newBuilder().setTs(ts).setKv(dataBuilder); + } + + public static EntityId toEntityId(String entityType, long entityIdMSB, long entityIdLSB) { + return EntityIdFactory.getByTypeAndUuid(entityType, new UUID(entityIdMSB, entityIdLSB)); + } + + public static List toTsKvEntityList(List dataList) { + List result = new ArrayList<>(dataList.size()); + dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), getKvEntry(proto.getKv())))); + return result; + } + + public static List toAttributeKvList(List dataList) { + List result = new ArrayList<>(dataList.size()); + dataList.forEach(proto -> result.add(new BaseAttributeKvEntry(getKvEntry(proto.getKv()), proto.getTs()))); + return result; + } + + private static KvEntry getKvEntry(KeyValueProto proto) { + KvEntry entry = null; + DataType type = DataType.values()[proto.getType().getNumber()]; + switch (type) { + case BOOLEAN: + entry = new BooleanDataEntry(proto.getKey(), proto.getBoolV()); + break; + case LONG: + entry = new LongDataEntry(proto.getKey(), proto.getLongV()); + break; + case DOUBLE: + entry = new DoubleDataEntry(proto.getKey(), proto.getDoubleV()); + break; + case STRING: + entry = new StringDataEntry(proto.getKey(), proto.getStringV()); + break; + case JSON: + entry = new JsonDataEntry(proto.getKey(), proto.getJsonV()); + break; + } + return entry; + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java new file mode 100644 index 0000000000..0be63f7b65 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java @@ -0,0 +1,51 @@ +/** + * Copyright © 2016-2020 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.service.subscription; + +import lombok.Builder; +import lombok.Getter; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.Map; + +public class TbTimeseriesSubscription extends TbSubscription { + + @Getter private final boolean allKeys; + @Getter private final Map keyStates; + @Getter private final long startTime; + @Getter private final long endTime; + + @Builder + public TbTimeseriesSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, + boolean allKeys, Map keyStates, long startTime, long endTime) { + super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.TIMESERIES); + this.allKeys = allKeys; + this.keyStates = keyStates; + this.startTime = startTime; + this.endTime = endTime; + } + + @Override + public boolean equals(Object o) { + return super.equals(o); + } + + @Override + public int hashCode() { + return super.hashCode(); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 04952a3bd1..d94eb3f2b2 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -18,71 +18,44 @@ package org.thingsboard.server.service.telemetry; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.protobuf.InvalidProtocolBufferException; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.annotation.Lazy; +import org.springframework.context.event.EventListener; import org.springframework.stereotype.Service; -import org.springframework.util.StringUtils; -import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.ThingsBoardThreadFactory; -import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; -import org.thingsboard.server.actors.service.ActorService; -import org.thingsboard.server.common.data.DataConstants; -import org.thingsboard.server.common.data.EntityType; -import org.thingsboard.server.common.data.EntityView; -import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.EntityIdFactory; -import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.Aggregation; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; -import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; -import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; -import org.thingsboard.server.common.data.kv.DataType; import org.thingsboard.server.common.data.kv.DoubleDataEntry; -import org.thingsboard.server.common.data.kv.JsonDataEntry; -import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; -import org.thingsboard.server.common.data.kv.ReadTsKvQuery; import org.thingsboard.server.common.data.kv.StringDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; -import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; -import org.thingsboard.server.common.msg.cluster.ServerAddress; import org.thingsboard.server.dao.attributes.AttributesService; -import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.timeseries.TimeseriesService; -import org.thingsboard.server.gen.cluster.ClusterAPIProtos; -import org.thingsboard.server.service.state.DefaultDeviceStateService; -import org.thingsboard.server.service.state.DeviceStateService; -import org.thingsboard.server.service.telemetry.sub.Subscription; -import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode; -import org.thingsboard.server.service.telemetry.sub.SubscriptionState; -import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate; +import org.thingsboard.server.gen.transport.TransportProtos; +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.ServiceType; +import org.thingsboard.server.queue.discovery.TopicPartitionInfo; +import org.thingsboard.server.queue.provider.TbCoreQueueProvider; +import org.thingsboard.server.service.queue.TbMsgCallback; +import org.thingsboard.server.service.subscription.SubscriptionManagerService; +import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import javax.annotation.Nullable; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; -import java.util.ArrayList; import java.util.Collections; -import java.util.HashSet; -import java.util.Iterator; import java.util.List; -import java.util.Map; -import java.util.Optional; import java.util.Set; -import java.util.TreeMap; -import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.function.Consumer; -import java.util.function.Function; -import java.util.function.Predicate; -import java.util.stream.Collectors; /** * Created by ashvayka on 27.03.18. @@ -91,8 +64,7 @@ import java.util.stream.Collectors; @Slf4j public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptionService { - @Autowired - private TelemetryWebSocketService wsService; + private final Set currentPartitions = ConcurrentHashMap.newKeySet(); @Autowired private AttributesService attrService; @@ -101,23 +73,24 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio private TimeseriesService tsService; @Autowired - private EntityViewService entityViewService; + private TbCoreQueueProvider coreQueueProvider; @Autowired - @Lazy - private DeviceStateService stateService; + private PartitionService partitionService; @Autowired - @Lazy - private ActorService actorService; - + private SubscriptionManagerService subscriptionManagerService; + + private TbQueueProducer> toCoreProducer; + private ExecutorService tsCallBackExecutor; private ExecutorService wsCallBackExecutor; @PostConstruct public void initExecutor() { - tsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ts-sub-callback")); - wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ws-sub-callback")); + tsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ts-service-ts-callback")); + wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ts-service-ws-callback")); + toCoreProducer = coreQueueProvider.getTbCoreMsgProducer(); } @PreDestroy @@ -130,65 +103,12 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio } } - private final Map> subscriptionsByEntityId = new ConcurrentHashMap<>(); - private final Map> subscriptionsByWsSessionId = new ConcurrentHashMap<>(); - - @Override - public void addLocalWsSubscription(String sessionId, EntityId entityId, SubscriptionState sub) { - long startTime = 0L; - long endTime = 0L; - if (entityId.getEntityType().equals(EntityType.ENTITY_VIEW) && TelemetryFeature.TIMESERIES.equals(sub.getType())) { - EntityView entityView = entityViewService.findEntityViewById(TenantId.SYS_TENANT_ID, new EntityViewId(entityId.getId())); - entityId = entityView.getEntityId(); - startTime = entityView.getStartTimeMs(); - endTime = entityView.getEndTimeMs(); - sub = getUpdatedSubscriptionState(entityId, sub, entityView); - } - //TODO 2.5 - Optional server = Optional.empty();//routingService.resolveById(entityId); - Subscription subscription; - if (server.isPresent()) { - ServerAddress address = server.get(); - log.trace("[{}] Forwarding subscription [{}] for [{}] entity [{}] to [{}]", sessionId, sub.getSubscriptionId(), entityId.getEntityType().name(), entityId, address); - subscription = new Subscription(sub, true, address, startTime, endTime); - tellNewSubscription(address, sessionId, subscription); - } else { - log.trace("[{}] Registering local subscription [{}] for [{}] entity [{}]", sessionId, sub.getSubscriptionId(), entityId.getEntityType().name(), entityId); - subscription = new Subscription(sub, true, null, startTime, endTime); - } - registerSubscription(sessionId, entityId, subscription); - } - - private SubscriptionState getUpdatedSubscriptionState(EntityId entityId, SubscriptionState sub, EntityView entityView) { - Map keyStates; - if (sub.isAllKeys()) { - keyStates = entityView.getKeys().getTimeseries().stream().collect(Collectors.toMap(k -> k, k -> 0L)); - } else { - keyStates = sub.getKeyStates().entrySet() - .stream().filter(entry -> entityView.getKeys().getTimeseries().contains(entry.getKey())) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); - } - return new SubscriptionState(sub.getWsSessionId(), sub.getSubscriptionId(), sub.getTenantId(), entityId, sub.getType(), false, keyStates, sub.getScope()); - } - @Override - public void cleanupLocalWsSessionSubscriptions(TelemetryWebSocketSessionRef sessionRef, String sessionId) { - cleanupLocalWsSessionSubscriptions(sessionId); - } - - @Override - public void removeSubscription(String sessionId, int subscriptionId) { - log.debug("[{}][{}] Going to remove subscription.", sessionId, subscriptionId); - Map sessionSubscriptions = subscriptionsByWsSessionId.get(sessionId); - if (sessionSubscriptions != null) { - Subscription subscription = sessionSubscriptions.remove(subscriptionId); - if (subscription != null) { - processSubscriptionRemoval(sessionId, sessionSubscriptions, subscription); - } else { - log.debug("[{}][{}] Subscription not found!", sessionId, subscriptionId); - } - } else { - log.debug("[{}] No session subscriptions found!", sessionId); + @EventListener(PartitionChangeEvent.class) + public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceKey().getServiceType())) { + currentPartitions.clear(); + currentPartitions.addAll(partitionChangeEvent.getPartitions()); } } @@ -201,14 +121,14 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio public void saveAndNotify(TenantId tenantId, EntityId entityId, List ts, long ttl, FutureCallback callback) { ListenableFuture> saveFuture = tsService.save(tenantId, entityId, ts, ttl); addMainCallback(saveFuture, callback); - addWsCallback(saveFuture, success -> onTimeseriesUpdate(entityId, ts)); + addWsCallback(saveFuture, success -> onTimeSeriesUpdate(tenantId, entityId, ts)); } @Override public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List attributes, FutureCallback callback) { ListenableFuture> saveFuture = attrService.save(tenantId, entityId, scope, attributes); addMainCallback(saveFuture, callback); - addWsCallback(saveFuture, success -> onAttributesUpdate(entityId, scope, attributes)); + addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes)); } @Override @@ -235,355 +155,23 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio , System.currentTimeMillis())), callback); } - @Override - public void onSharedAttributesUpdate(TenantId tenantId, DeviceId deviceId, Set attributes) { - DeviceAttributesEventNotificationMsg notificationMsg = DeviceAttributesEventNotificationMsg.onUpdate(tenantId, - deviceId, DataConstants.SHARED_SCOPE, new ArrayList<>(attributes)); - actorService.onMsg(new SendToClusterMsg(deviceId, notificationMsg)); - } - - @Override - public void onNewRemoteSubscription(ServerAddress serverAddress, byte[] data) { - ClusterAPIProtos.SubscriptionProto proto; - try { - proto = ClusterAPIProtos.SubscriptionProto.parseFrom(data); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - Map statesMap = proto.getKeyStatesList().stream().collect( - Collectors.toMap(ClusterAPIProtos.SubscriptionKetStateProto::getKey, ClusterAPIProtos.SubscriptionKetStateProto::getTs)); - Subscription subscription = new Subscription( - new SubscriptionState(proto.getSessionId(), proto.getSubscriptionId(), - new TenantId(UUID.fromString(proto.getTenantId())), - EntityIdFactory.getByTypeAndId(proto.getEntityType(), proto.getEntityId()), - TelemetryFeature.valueOf(proto.getType()), proto.getAllKeys(), statesMap, proto.getScope()), - false, new ServerAddress(serverAddress.getHost(), serverAddress.getPort(), serverAddress.getServerType())); - - addRemoteWsSubscription(serverAddress, proto.getSessionId(), subscription); - } - - @Override - public void onRemoteSubscriptionUpdate(ServerAddress serverAddress, byte[] data) { - ClusterAPIProtos.SubscriptionUpdateProto proto; - try { - proto = ClusterAPIProtos.SubscriptionUpdateProto.parseFrom(data); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - SubscriptionUpdate update = convert(proto); - String sessionId = proto.getSessionId(); - log.trace("[{}] Processing remote subscription onUpdate [{}]", sessionId, update); - Optional subOpt = getSubscription(sessionId, update.getSubscriptionId()); - if (subOpt.isPresent()) { - updateSubscriptionState(sessionId, subOpt.get(), update); - wsService.sendWsMsg(sessionId, update); - } - } - - @Override - public void onRemoteSubscriptionClose(ServerAddress serverAddress, byte[] data) { - ClusterAPIProtos.SubscriptionCloseProto proto; - try { - proto = ClusterAPIProtos.SubscriptionCloseProto.parseFrom(data); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - removeSubscription(proto.getSessionId(), proto.getSubscriptionId()); - } - - @Override - public void onRemoteSessionClose(ServerAddress serverAddress, byte[] data) { - ClusterAPIProtos.SessionCloseProto proto; - try { - proto = ClusterAPIProtos.SessionCloseProto.parseFrom(data); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - cleanupRemoteWsSessionSubscriptions(proto.getSessionId()); - } - - @Override - public void onRemoteAttributesUpdate(ServerAddress serverAddress, byte[] data) { - ClusterAPIProtos.AttributeUpdateProto proto; - try { - proto = ClusterAPIProtos.AttributeUpdateProto.parseFrom(data); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - onAttributesUpdate(EntityIdFactory.getByTypeAndId(proto.getEntityType(), proto.getEntityId()), proto.getScope(), - proto.getDataList().stream().map(this::toAttribute).collect(Collectors.toList())); - } - - @Override - public void onRemoteTsUpdate(ServerAddress serverAddress, byte[] data) { - ClusterAPIProtos.TimeseriesUpdateProto proto; - try { - proto = ClusterAPIProtos.TimeseriesUpdateProto.parseFrom(data); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - onTimeseriesUpdate(EntityIdFactory.getByTypeAndId(proto.getEntityType(), proto.getEntityId()), - proto.getDataList().stream().map(this::toTimeseries).collect(Collectors.toList())); - } - - @Override - public void onClusterUpdate() { - log.trace("Processing cluster onUpdate msg!"); - Iterator>> deviceIterator = subscriptionsByEntityId.entrySet().iterator(); - while (deviceIterator.hasNext()) { - Map.Entry> e = deviceIterator.next(); - Set subscriptions = e.getValue(); - //TODO 2.5 - Optional newAddressOptional = Optional.empty();// routingService.resolveById(e.getKey()); - if (newAddressOptional.isPresent()) { - newAddressOptional.ifPresent(serverAddress -> checkSubscriptionsNewAddress(serverAddress, subscriptions)); - } else { - checkSubscriptionsPrevAddress(subscriptions); - } - if (subscriptions.size() == 0) { - log.trace("[{}] No more subscriptions for this device on current server.", e.getKey()); - deviceIterator.remove(); - } - } - } - - private void checkSubscriptionsNewAddress(ServerAddress newAddress, Set subscriptions) { - Iterator subscriptionIterator = subscriptions.iterator(); - while (subscriptionIterator.hasNext()) { - Subscription s = subscriptionIterator.next(); - if (s.isLocal()) { - if (!newAddress.equals(s.getServer())) { - log.trace("[{}] Local subscription is now handled on new server [{}]", s.getWsSessionId(), newAddress); - s.setServer(newAddress); - tellNewSubscription(newAddress, s.getWsSessionId(), s); - } - } else { - log.trace("[{}] Remote subscription is now handled on new server address: [{}]", s.getWsSessionId(), newAddress); - subscriptionIterator.remove(); - //TODO: onUpdate state of subscription by WsSessionId and other maps. - } - } - } - - private void checkSubscriptionsPrevAddress(Set subscriptions) { - for (Subscription s : subscriptions) { - if (s.isLocal() && s.getServer() != null) { - log.trace("[{}] Local subscription is no longer handled on remote server address [{}]", s.getWsSessionId(), s.getServer()); - s.setServer(null); - } else { - log.trace("[{}] Remote subscription is on up to date server address.", s.getWsSessionId()); - } - } - } - - private void addRemoteWsSubscription(ServerAddress address, String sessionId, Subscription subscription) { - EntityId entityId = subscription.getEntityId(); - log.trace("[{}] Registering remote subscription [{}] for entity [{}] to [{}]", sessionId, subscription.getSubscriptionId(), entityId, address); - registerSubscription(sessionId, entityId, subscription); - if (subscription.getType() == TelemetryFeature.ATTRIBUTES) { - final Map keyStates = subscription.getKeyStates(); - DonAsynchron.withCallback(attrService.find(subscription.getSub().getTenantId(), entityId, DataConstants.CLIENT_SCOPE, keyStates.keySet()), values -> { - List missedUpdates = new ArrayList<>(); - values.forEach(latestEntry -> { - if (latestEntry.getLastUpdateTs() > keyStates.get(latestEntry.getKey())) { - missedUpdates.add(new BasicTsKvEntry(latestEntry.getLastUpdateTs(), latestEntry)); - } - }); - if (!missedUpdates.isEmpty()) { - tellRemoteSubUpdate(address, sessionId, new SubscriptionUpdate(subscription.getSubscriptionId(), missedUpdates)); - } - }, - e -> log.error("Failed to fetch missed updates.", e), tsCallBackExecutor); - } else if (subscription.getType() == TelemetryFeature.TIMESERIES) { - long curTs = System.currentTimeMillis(); - List queries = new ArrayList<>(); - subscription.getKeyStates().entrySet().forEach(e -> { - if (curTs > e.getValue()) { - queries.add(new BaseReadTsKvQuery(e.getKey(), e.getValue() + 1L, curTs, 0, 1000, Aggregation.NONE)); - } else { - log.debug("[{}] Invalid subscription [{}], entityId [{}] curTs [{}]", sessionId, subscription, entityId, curTs); - } - }); - if (!queries.isEmpty()) { - DonAsynchron.withCallback(tsService.findAll(subscription.getSub().getTenantId(), entityId, queries), - missedUpdates -> { - if (missedUpdates != null && !missedUpdates.isEmpty()) { - tellRemoteSubUpdate(address, sessionId, new SubscriptionUpdate(subscription.getSubscriptionId(), missedUpdates)); - } - }, - e -> log.error("Failed to fetch missed updates.", e), - tsCallBackExecutor); - } - } - } - - private void onAttributesUpdate(EntityId entityId, String scope, List attributes) { - //TODO 2.5 - Optional serverAddress = Optional.empty();//routingService.resolveById(entityId); - if (!serverAddress.isPresent()) { - onLocalAttributesUpdate(entityId, scope, attributes); - if (entityId.getEntityType() == EntityType.DEVICE && DataConstants.SERVER_SCOPE.equalsIgnoreCase(scope)) { - for (AttributeKvEntry attribute : attributes) { - if (attribute.getKey().equals(DefaultDeviceStateService.INACTIVITY_TIMEOUT)) { - stateService.onDeviceInactivityTimeoutUpdate(new DeviceId(entityId.getId()), attribute.getLongValue().orElse(0L)); - } - } - } + private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes) { + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); + if (currentPartitions.contains(tpi)) { + subscriptionManagerService.onAttributesUpdate(tenantId, entityId, scope, attributes, TbMsgCallback.EMPTY); } else { - tellRemoteAttributesUpdate(serverAddress.get(), entityId, scope, attributes); + TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toAttributesUpdateProto(tenantId, entityId, scope, attributes); + toCoreProducer.send(tpi, new TbProtoQueueMsg<>(entityId.getId(), toCoreMsg), null); } } - private void onTimeseriesUpdate(EntityId entityId, List ts) { - //TODO 2.5 - Optional serverAddress = Optional.empty();//routingService.resolveById(entityId); - if (!serverAddress.isPresent()) { - onLocalTimeseriesUpdate(entityId, ts); + private void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts) { + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); + if (currentPartitions.contains(tpi)) { + subscriptionManagerService.onTimeseriesDataUpdate(tenantId, entityId, ts, TbMsgCallback.EMPTY); } else { - tellRemoteTimeseriesUpdate(serverAddress.get(), entityId, ts); - } - } - - private void onLocalAttributesUpdate(EntityId entityId, String scope, List attributes) { - onLocalSubUpdate(entityId, s -> TelemetryFeature.ATTRIBUTES == s.getType() && (StringUtils.isEmpty(s.getScope()) || scope.equals(s.getScope())), s -> { - List subscriptionUpdate = null; - for (AttributeKvEntry kv : attributes) { - if (s.isAllKeys() || s.getKeyStates().containsKey(kv.getKey())) { - if (subscriptionUpdate == null) { - subscriptionUpdate = new ArrayList<>(); - } - subscriptionUpdate.add(new BasicTsKvEntry(kv.getLastUpdateTs(), kv)); - } - } - return subscriptionUpdate; - }); - } - - private void onLocalTimeseriesUpdate(EntityId entityId, List ts) { - onLocalSubUpdate(entityId, s -> TelemetryFeature.TIMESERIES == s.getType(), s -> { - List subscriptionUpdate = null; - for (TsKvEntry kv : ts) { - if (isInTimeRange(s, kv.getTs()) && (s.isAllKeys() || s.getKeyStates().containsKey((kv.getKey())))) { - if (subscriptionUpdate == null) { - subscriptionUpdate = new ArrayList<>(); - } - subscriptionUpdate.add(kv); - } - } - return subscriptionUpdate; - }); - } - - private boolean isInTimeRange(Subscription subscription, long kvTime) { - return (subscription.getStartTime() == 0 || subscription.getStartTime() <= kvTime) - && (subscription.getEndTime() == 0 || subscription.getEndTime() >= kvTime); - } - - private void onLocalSubUpdate(EntityId entityId, Predicate filter, Function> f) { - Set deviceSubscriptions = subscriptionsByEntityId.get(entityId); - if (deviceSubscriptions != null) { - deviceSubscriptions.stream().filter(filter).forEach(s -> { - String sessionId = s.getWsSessionId(); - List subscriptionUpdate = f.apply(s); - if (subscriptionUpdate != null && !subscriptionUpdate.isEmpty()) { - SubscriptionUpdate update = new SubscriptionUpdate(s.getSubscriptionId(), subscriptionUpdate); - if (s.isLocal()) { - updateSubscriptionState(sessionId, s, update); - wsService.sendWsMsg(sessionId, update); - } else { - tellRemoteSubUpdate(s.getServer(), sessionId, update); - } - } - }); - } else { - log.debug("[{}] No device subscriptions to process!", entityId); - } - } - - private void updateSubscriptionState(String sessionId, Subscription subState, SubscriptionUpdate update) { - log.trace("[{}] updating subscription state {} using onUpdate {}", sessionId, subState, update); - update.getLatestValues().entrySet().forEach(e -> subState.setKeyState(e.getKey(), e.getValue())); - } - - private void registerSubscription(String sessionId, EntityId entityId, Subscription subscription) { - Set deviceSubscriptions = subscriptionsByEntityId.computeIfAbsent(entityId, k -> ConcurrentHashMap.newKeySet()); - deviceSubscriptions.add(subscription); - Map sessionSubscriptions = subscriptionsByWsSessionId.computeIfAbsent(sessionId, k -> new ConcurrentHashMap<>()); - sessionSubscriptions.put(subscription.getSubscriptionId(), subscription); - } - - private void cleanupLocalWsSessionSubscriptions(String sessionId) { - cleanupWsSessionSubscriptions(sessionId, true); - } - - private void cleanupRemoteWsSessionSubscriptions(String sessionId) { - cleanupWsSessionSubscriptions(sessionId, false); - } - - private void cleanupWsSessionSubscriptions(String sessionId, boolean localSession) { - log.debug("[{}] Removing all subscriptions for particular session.", sessionId); - Map sessionSubscriptions = subscriptionsByWsSessionId.get(sessionId); - if (sessionSubscriptions != null) { - int sessionSubscriptionSize = sessionSubscriptions.size(); - - for (Subscription subscription : sessionSubscriptions.values()) { - EntityId entityId = subscription.getEntityId(); - Set deviceSubscriptions = subscriptionsByEntityId.get(entityId); - deviceSubscriptions.remove(subscription); - if (deviceSubscriptions.isEmpty()) { - subscriptionsByEntityId.remove(entityId); - } - } - subscriptionsByWsSessionId.remove(sessionId); - log.debug("[{}] Removed {} subscriptions for particular session.", sessionId, sessionSubscriptionSize); - - if (localSession) { - notifyWsSubscriptionClosed(sessionId, sessionSubscriptions); - } - } else { - log.debug("[{}] No subscriptions found!", sessionId); - } - } - - private void notifyWsSubscriptionClosed(String sessionId, Map sessionSubscriptions) { - Set affectedServers = new HashSet<>(); - for (Subscription subscription : sessionSubscriptions.values()) { - if (subscription.getServer() != null) { - affectedServers.add(subscription.getServer()); - } - } - for (ServerAddress address : affectedServers) { - log.debug("[{}] Going to onSubscriptionUpdate [{}] server about session close event", sessionId, address); - tellRemoteSessionClose(address, sessionId); - } - } - - private void processSubscriptionRemoval(String sessionId, Map sessionSubscriptions, Subscription subscription) { - EntityId entityId = subscription.getEntityId(); - if (subscription.isLocal() && subscription.getServer() != null) { - tellRemoteSubClose(subscription.getServer(), sessionId, subscription.getSubscriptionId()); - } - if (sessionSubscriptions.isEmpty()) { - log.debug("[{}] Removed last subscription for particular session.", sessionId); - subscriptionsByWsSessionId.remove(sessionId); - } else { - log.debug("[{}] Removed session subscription.", sessionId); - } - Set deviceSubscriptions = subscriptionsByEntityId.get(entityId); - if (deviceSubscriptions != null) { - boolean result = deviceSubscriptions.remove(subscription); - if (result) { - if (deviceSubscriptions.size() == 0) { - log.debug("[{}] Removed last subscription for particular device.", sessionId); - subscriptionsByEntityId.remove(entityId); - } else { - log.debug("[{}] Removed device subscription.", sessionId); - } - } else { - log.debug("[{}] Subscription not found!", sessionId); - } - } else { - log.debug("[{}] No device subscriptions found!", sessionId); + TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toTimeseriesUpdateProto(tenantId, entityId, ts); + toCoreProducer.send(tpi, new TbProtoQueueMsg<>(entityId.getId(), toCoreMsg), null); } } @@ -613,167 +201,4 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio } }, wsCallBackExecutor); } - - private void tellNewSubscription(ServerAddress address, String sessionId, Subscription sub) { - ClusterAPIProtos.SubscriptionProto.Builder builder = ClusterAPIProtos.SubscriptionProto.newBuilder(); - builder.setSessionId(sessionId); - builder.setSubscriptionId(sub.getSubscriptionId()); - builder.setTenantId(sub.getSub().getTenantId().getId().toString()); - builder.setEntityType(sub.getEntityId().getEntityType().name()); - builder.setEntityId(sub.getEntityId().getId().toString()); - builder.setType(sub.getType().name()); - builder.setAllKeys(sub.isAllKeys()); - if (sub.getScope() != null) { - builder.setScope(sub.getScope()); - } - sub.getKeyStates().entrySet().forEach(e -> builder.addKeyStates( - ClusterAPIProtos.SubscriptionKetStateProto.newBuilder().setKey(e.getKey()).setTs(e.getValue()).build())); - //TODO 2.5 -// rpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TELEMETRY_SUBSCRIPTION_CREATE_MESSAGE, builder.build().toByteArray()); - } - - private void tellRemoteSubUpdate(ServerAddress address, String sessionId, SubscriptionUpdate update) { - ClusterAPIProtos.SubscriptionUpdateProto.Builder builder = ClusterAPIProtos.SubscriptionUpdateProto.newBuilder(); - builder.setSessionId(sessionId); - builder.setSubscriptionId(update.getSubscriptionId()); - builder.setErrorCode(update.getErrorCode()); - if (update.getErrorMsg() != null) { - builder.setErrorMsg(update.getErrorMsg()); - } - update.getData().entrySet().forEach( - e -> { - ClusterAPIProtos.SubscriptionUpdateValueListProto.Builder dataBuilder = ClusterAPIProtos.SubscriptionUpdateValueListProto.newBuilder(); - - dataBuilder.setKey(e.getKey()); - e.getValue().forEach(v -> { - Object[] array = (Object[]) v; - dataBuilder.addTs((long) array[0]); - dataBuilder.addValue((String) array[1]); - }); - - builder.addData(dataBuilder.build()); - } - ); - //TODO 2.5 -// rpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TELEMETRY_SUBSCRIPTION_UPDATE_MESSAGE, builder.build().toByteArray()); - } - - private void tellRemoteAttributesUpdate(ServerAddress address, EntityId entityId, String scope, List attributes) { - ClusterAPIProtos.AttributeUpdateProto.Builder builder = ClusterAPIProtos.AttributeUpdateProto.newBuilder(); - builder.setEntityId(entityId.getId().toString()); - builder.setEntityType(entityId.getEntityType().name()); - builder.setScope(scope); - attributes.forEach(v -> builder.addData(toKeyValueProto(v.getLastUpdateTs(), v).build())); - //TODO 2.5 -// rpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TELEMETRY_ATTR_UPDATE_MESSAGE, builder.build().toByteArray()); - } - - private void tellRemoteTimeseriesUpdate(ServerAddress address, EntityId entityId, List ts) { - ClusterAPIProtos.TimeseriesUpdateProto.Builder builder = ClusterAPIProtos.TimeseriesUpdateProto.newBuilder(); - builder.setEntityId(entityId.getId().toString()); - builder.setEntityType(entityId.getEntityType().name()); - ts.forEach(v -> builder.addData(toKeyValueProto(v.getTs(), v).build())); - //TODO 2.5 -// rpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TELEMETRY_TS_UPDATE_MESSAGE, builder.build().toByteArray()); - } - - private void tellRemoteSessionClose(ServerAddress address, String sessionId) { - ClusterAPIProtos.SessionCloseProto proto = ClusterAPIProtos.SessionCloseProto.newBuilder().setSessionId(sessionId).build(); - //TODO 2.5 -// rpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TELEMETRY_SESSION_CLOSE_MESSAGE, proto.toByteArray()); - } - - private void tellRemoteSubClose(ServerAddress address, String sessionId, int subscriptionId) { - ClusterAPIProtos.SubscriptionCloseProto proto = ClusterAPIProtos.SubscriptionCloseProto.newBuilder().setSessionId(sessionId).setSubscriptionId(subscriptionId).build(); - //TODO 2.5 -// rpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TELEMETRY_SUBSCRIPTION_CLOSE_MESSAGE, proto.toByteArray()); - } - - private ClusterAPIProtos.KeyValueProto.Builder toKeyValueProto(long ts, KvEntry attr) { - ClusterAPIProtos.KeyValueProto.Builder dataBuilder = ClusterAPIProtos.KeyValueProto.newBuilder(); - dataBuilder.setKey(attr.getKey()); - dataBuilder.setTs(ts); - dataBuilder.setValueType(attr.getDataType().ordinal()); - switch (attr.getDataType()) { - case BOOLEAN: - Optional booleanValue = attr.getBooleanValue(); - booleanValue.ifPresent(dataBuilder::setBoolValue); - break; - case LONG: - Optional longValue = attr.getLongValue(); - longValue.ifPresent(dataBuilder::setLongValue); - break; - case DOUBLE: - Optional doubleValue = attr.getDoubleValue(); - doubleValue.ifPresent(dataBuilder::setDoubleValue); - break; - case JSON: - Optional jsonValue = attr.getJsonValue(); - jsonValue.ifPresent(dataBuilder::setJsonValue); - break; - case STRING: - Optional stringValue = attr.getStrValue(); - stringValue.ifPresent(dataBuilder::setStrValue); - break; - } - return dataBuilder; - } - - private AttributeKvEntry toAttribute(ClusterAPIProtos.KeyValueProto proto) { - return new BaseAttributeKvEntry(getKvEntry(proto), proto.getTs()); - } - - private TsKvEntry toTimeseries(ClusterAPIProtos.KeyValueProto proto) { - return new BasicTsKvEntry(proto.getTs(), getKvEntry(proto)); - } - - private KvEntry getKvEntry(ClusterAPIProtos.KeyValueProto proto) { - KvEntry entry = null; - DataType type = DataType.values()[proto.getValueType()]; - switch (type) { - case BOOLEAN: - entry = new BooleanDataEntry(proto.getKey(), proto.getBoolValue()); - break; - case LONG: - entry = new LongDataEntry(proto.getKey(), proto.getLongValue()); - break; - case DOUBLE: - entry = new DoubleDataEntry(proto.getKey(), proto.getDoubleValue()); - break; - case STRING: - entry = new StringDataEntry(proto.getKey(), proto.getStrValue()); - break; - case JSON: - entry = new JsonDataEntry(proto.getKey(), proto.getJsonValue()); - break; - } - return entry; - } - - private SubscriptionUpdate convert(ClusterAPIProtos.SubscriptionUpdateProto proto) { - if (proto.getErrorCode() > 0) { - return new SubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg()); - } else { - Map> data = new TreeMap<>(); - proto.getDataList().forEach(v -> { - List values = data.computeIfAbsent(v.getKey(), k -> new ArrayList<>()); - for (int i = 0; i < v.getTsCount(); i++) { - Object[] value = new Object[2]; - value[0] = v.getTs(i); - value[1] = v.getValue(i); - values.add(value); - } - }); - return new SubscriptionUpdate(proto.getSubscriptionId(), data); - } - } - - private Optional getSubscription(String sessionId, int subscriptionId) { - Subscription state = null; - Map subMap = subscriptionsByWsSessionId.get(sessionId); - if (subMap != null) { - state = subMap.get(subscriptionId); - } - return Optional.ofNullable(state); - } } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java index c4fe716ad9..51556602f1 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java @@ -42,24 +42,25 @@ import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.util.TenantRateLimitException; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.service.security.AccessValidator; import org.thingsboard.server.service.security.ValidationCallback; import org.thingsboard.server.service.security.ValidationResult; import org.thingsboard.server.service.security.ValidationResultCode; import org.thingsboard.server.service.security.model.UserPrincipal; import org.thingsboard.server.service.security.permission.Operation; +import org.thingsboard.server.service.subscription.LocalSubscriptionService; +import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope; +import org.thingsboard.server.service.subscription.TbAttributeSubscription; +import org.thingsboard.server.service.subscription.TbTimeseriesSubscription; import org.thingsboard.server.service.telemetry.cmd.AttributesSubscriptionCmd; import org.thingsboard.server.service.telemetry.cmd.GetHistoryCmd; import org.thingsboard.server.service.telemetry.cmd.SubscriptionCmd; import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmd; import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper; import org.thingsboard.server.service.telemetry.cmd.TimeseriesSubscriptionCmd; -import org.thingsboard.server.service.telemetry.exception.AccessDeniedException; -import org.thingsboard.server.service.telemetry.exception.EntityNotFoundException; -import org.thingsboard.server.service.telemetry.exception.InternalErrorException; import org.thingsboard.server.service.telemetry.exception.UnauthorizedException; import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode; -import org.thingsboard.server.service.telemetry.sub.SubscriptionState; import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate; import javax.annotation.Nullable; @@ -70,7 +71,6 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Optional; @@ -98,7 +98,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi private final ConcurrentMap wsSessionsMap = new ConcurrentHashMap<>(); @Autowired - private TelemetrySubscriptionService subscriptionManager; + private LocalSubscriptionService subService; @Autowired private TelemetryWebSocketMsgEndpoint msgEndpoint; @@ -112,6 +112,9 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi @Autowired private TimeseriesService tsService; + @Autowired + private TbServiceInfoProvider serviceInfoProvider; + @Value("${server.ws.limits.max_subscriptions_per_tenant:0}") private int maxSubscriptionsPerTenant; @Value("${server.ws.limits.max_subscriptions_per_customer:0}") @@ -127,9 +130,11 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi private ConcurrentMap> publicUserSubscriptionsMap = new ConcurrentHashMap<>(); private ExecutorService executor; + private String serviceId; @PostConstruct public void initExecutor() { + serviceId = serviceInfoProvider.getServiceId(); executor = Executors.newWorkStealingPool(50); } @@ -153,7 +158,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi break; case CLOSED: wsSessionsMap.remove(sessionId); - subscriptionManager.cleanupLocalWsSessionSubscriptions(sessionRef, sessionId); + subService.cancelAllSessionSubscriptions(sessionId); processSessionClose(sessionRef); break; } @@ -334,8 +339,16 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi keys.forEach(key -> subState.put(key, 0L)); attributesData.forEach(v -> subState.put(v.getKey(), v.getTs())); - SubscriptionState sub = new SubscriptionState(sessionId, cmd.getCmdId(), sessionRef.getSecurityCtx().getTenantId(), entityId, TelemetryFeature.ATTRIBUTES, false, subState, cmd.getScope()); - subscriptionManager.addLocalWsSubscription(sessionId, entityId, sub); + TbAttributeSubscription sub = TbAttributeSubscription.builder() + .serviceId(serviceId) + .sessionId(sessionId) + .subscriptionId(cmd.getCmdId()) + .tenantId(sessionRef.getSecurityCtx().getTenantId()) + .entityId(entityId) + .allKeys(false) + .keyStates(subState) + .scope(TbAttributeSubscriptionScope.valueOf(cmd.getScope())).build(); + subService.addSubscription(sub); } @Override @@ -421,8 +434,16 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi Map subState = new HashMap<>(attributesData.size()); attributesData.forEach(v -> subState.put(v.getKey(), v.getTs())); - SubscriptionState sub = new SubscriptionState(sessionId, cmd.getCmdId(), sessionRef.getSecurityCtx().getTenantId(), entityId, TelemetryFeature.ATTRIBUTES, true, subState, cmd.getScope()); - subscriptionManager.addLocalWsSubscription(sessionId, entityId, sub); + TbAttributeSubscription sub = TbAttributeSubscription.builder() + .serviceId(serviceId) + .sessionId(sessionId) + .subscriptionId(cmd.getCmdId()) + .tenantId(sessionRef.getSecurityCtx().getTenantId()) + .entityId(entityId) + .allKeys(true) + .keyStates(subState) + .scope(TbAttributeSubscriptionScope.valueOf(cmd.getScope())).build(); + subService.addSubscription(sub); } @Override @@ -494,8 +515,16 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi sendWsMsg(sessionRef, new SubscriptionUpdate(cmd.getCmdId(), data)); Map subState = new HashMap<>(data.size()); data.forEach(v -> subState.put(v.getKey(), v.getTs())); - SubscriptionState sub = new SubscriptionState(sessionId, cmd.getCmdId(), sessionRef.getSecurityCtx().getTenantId(), entityId, TelemetryFeature.TIMESERIES, true, subState, cmd.getScope()); - subscriptionManager.addLocalWsSubscription(sessionId, entityId, sub); + + TbTimeseriesSubscription sub = TbTimeseriesSubscription.builder() + .serviceId(serviceId) + .sessionId(sessionId) + .subscriptionId(cmd.getCmdId()) + .tenantId(sessionRef.getSecurityCtx().getTenantId()) + .entityId(entityId) + .allKeys(true) + .keyStates(subState).build(); + subService.addSubscription(sub); } @Override @@ -520,12 +549,19 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi @Override public void onSuccess(List data) { sendWsMsg(sessionRef, new SubscriptionUpdate(cmd.getCmdId(), data)); - Map subState = new HashMap<>(keys.size()); keys.forEach(key -> subState.put(key, startTs)); data.forEach(v -> subState.put(v.getKey(), v.getTs())); - SubscriptionState sub = new SubscriptionState(sessionId, cmd.getCmdId(), sessionRef.getSecurityCtx().getTenantId(), entityId, TelemetryFeature.TIMESERIES, false, subState, cmd.getScope()); - subscriptionManager.addLocalWsSubscription(sessionId, entityId, sub); + + TbTimeseriesSubscription sub = TbTimeseriesSubscription.builder() + .serviceId(serviceId) + .sessionId(sessionId) + .subscriptionId(cmd.getCmdId()) + .tenantId(sessionRef.getSecurityCtx().getTenantId()) + .entityId(entityId) + .allKeys(false) + .keyStates(subState).build(); + subService.addSubscription(sub); } @Override @@ -544,9 +580,9 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi private void unsubscribe(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) { if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) { - subscriptionManager.cleanupLocalWsSessionSubscriptions(sessionRef, sessionId); + subService.cancelAllSessionSubscriptions(sessionId); } else { - subscriptionManager.removeSubscription(sessionId, cmd.getCmdId()); + subService.cancelSubscription(sessionId, cmd.getCmdId()); } } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java index d77a0e50dc..e905559516 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java @@ -15,34 +15,17 @@ */ package org.thingsboard.server.service.telemetry; +import org.springframework.context.ApplicationListener; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.cluster.ServerAddress; +import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.service.telemetry.sub.SubscriptionState; /** * Created by ashvayka on 27.03.18. */ -public interface TelemetrySubscriptionService extends RuleEngineTelemetryService { +public interface TelemetrySubscriptionService extends RuleEngineTelemetryService, ApplicationListener { - void addLocalWsSubscription(String sessionId, EntityId entityId, SubscriptionState sub); - - void cleanupLocalWsSessionSubscriptions(TelemetryWebSocketSessionRef sessionRef, String sessionId); - - void removeSubscription(String sessionId, int cmdId); - - void onNewRemoteSubscription(ServerAddress serverAddress, byte[] data); - - void onRemoteSubscriptionUpdate(ServerAddress serverAddress, byte[] bytes); - - void onRemoteSubscriptionClose(ServerAddress serverAddress, byte[] bytes); - - void onRemoteSessionClose(ServerAddress serverAddress, byte[] bytes); - - void onRemoteAttributesUpdate(ServerAddress serverAddress, byte[] bytes); - - void onRemoteTsUpdate(ServerAddress serverAddress, byte[] bytes); - - void onClusterUpdate(); } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/sub/SubscriptionUpdate.java b/application/src/main/java/org/thingsboard/server/service/telemetry/sub/SubscriptionUpdate.java index 68aaca34c2..992dc26af9 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/sub/SubscriptionUpdate.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/sub/SubscriptionUpdate.java @@ -75,7 +75,7 @@ public class SubscriptionUpdate { if (data == null) { return Collections.emptyMap(); } else { - return data.entrySet().stream().collect(Collectors.toMap(e -> e.getKey(), e -> { + return data.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> { List data = e.getValue(); Object[] latest = (Object[]) data.get(data.size() - 1); return (long) latest[0]; diff --git a/application/src/main/proto/cluster.proto b/application/src/main/proto/cluster.proto index 46526e8e1c..f0ca4d6d7c 100644 --- a/application/src/main/proto/cluster.proto +++ b/application/src/main/proto/cluster.proto @@ -64,69 +64,7 @@ enum MessageType { } // Messages related to CLUSTER_TELEMETRY_MESSAGE -message SubscriptionProto { - string sessionId = 1; - int32 subscriptionId = 2; - string entityType = 3; - string tenantId = 4; - string entityId = 5; - string type = 6; - bool allKeys = 7; - repeated SubscriptionKetStateProto keyStates = 8; - string scope = 9; -} - -message SubscriptionUpdateProto { - string sessionId = 1; - int32 subscriptionId = 2; - int32 errorCode = 3; - string errorMsg = 4; - repeated SubscriptionUpdateValueListProto data = 5; -} - -message AttributeUpdateProto { - string entityType = 1; - string entityId = 2; - string scope = 3; - repeated KeyValueProto data = 4; -} - -message TimeseriesUpdateProto { - string entityType = 1; - string entityId = 2; - repeated KeyValueProto data = 4; -} - -message SessionCloseProto { - string sessionId = 1; -} - -message SubscriptionCloseProto { - string sessionId = 1; - int32 subscriptionId = 2; -} - -message SubscriptionKetStateProto { - string key = 1; - int64 ts = 2; -} - -message SubscriptionUpdateValueListProto { - string key = 1; - repeated int64 ts = 2; - repeated string value = 3; -} -message KeyValueProto { - string key = 1; - int64 ts = 2; - int32 valueType = 3; - string strValue = 4; - int64 longValue = 5; - double doubleValue = 6; - bool boolValue = 7; - string jsonValue = 8; -} message FromDeviceRPCResponseProto { int64 requestIdMSB = 1; diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index fc49716d25..8a4d735d26 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -412,7 +412,7 @@ audit-log: state: defaultInactivityTimeoutInSec: "${DEFAULT_INACTIVITY_TIMEOUT:10}" defaultStateCheckIntervalInSec: "${DEFAULT_STATE_CHECK_INTERVAL:10}" - persistToTelemetry: "${PERSIST_STATE_TO_TELEMETRY:false}" + persistToTelemetry: "${PERSIST_STATE_TO_TELEMETRY:true}" js: evaluator: "${JS_EVALUATOR:local}" # local/remote diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/kv/BaseAttributeKvEntry.java b/common/data/src/main/java/org/thingsboard/server/common/data/kv/BaseAttributeKvEntry.java index 5639f98d01..2993d81154 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/kv/BaseAttributeKvEntry.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/kv/BaseAttributeKvEntry.java @@ -32,6 +32,10 @@ public class BaseAttributeKvEntry implements AttributeKvEntry { this.lastUpdateTs = lastUpdateTs; } + public BaseAttributeKvEntry(long lastUpdateTs, KvEntry kv) { + this(kv, lastUpdateTs); + } + @Override public long getLastUpdateTs() { return lastUpdateTs; 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 new file mode 100644 index 0000000000..fb379c213a --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java @@ -0,0 +1,33 @@ +/** + * Copyright © 2016-2020 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.Set; + + +public class ClusterTopologyChangeEvent extends ApplicationEvent { + + @Getter + private final Set serviceKeys; + + public ClusterTopologyChangeEvent(Object source, Set serviceKeys) { + super(source); + this.serviceKeys = serviceKeys; + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java index 97b660a48c..f5efec30d4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java @@ -66,6 +66,10 @@ public class ConsistentHashPartitionService implements PartitionService { //TODO: Fetch this from the database, together with size of partitions for each service for each tenant. private ConcurrentMap> isolatedTenants = new ConcurrentHashMap<>(); + private Map tbCoreNotificationTopics = new HashMap<>(); + private Map tbRuleEngineNotificationTopics = new HashMap<>(); + private List currentOtherServices; + private HashFunction hashFunction; public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider, ApplicationEventPublisher applicationEventPublisher) { @@ -85,12 +89,11 @@ public class ConsistentHashPartitionService implements PartitionService { @Override public List getCurrentPartitions(ServiceType serviceType) { ServiceInfo currentService = serviceInfoProvider.getServiceInfo(); - TenantId tenantId = getTenantId(currentService); + TenantId tenantId = getSystemOrIsolatedTenantId(currentService); ServiceKey serviceKey = new ServiceKey(serviceType, tenantId); List partitions = myPartitions.get(serviceKey); List topicPartitions = new ArrayList<>(); for (Integer partition : partitions) { - TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); tpi.topic(partitionTopics.get(serviceType)); tpi.partition(partition); @@ -112,34 +115,16 @@ public class ConsistentHashPartitionService implements PartitionService { return buildTopicPartitionInfo(serviceType, tenantId, partition); } - private TopicPartitionInfo buildTopicPartitionInfo(ServiceKey serviceKey, int partition) { - return buildTopicPartitionInfo(serviceKey.getServiceType(), serviceKey.getTenantId(), partition); - } - - private TopicPartitionInfo buildTopicPartitionInfo(ServiceType serviceType, TenantId tenantId, int partition) { - boolean isolated = isolatedTenants.get(tenantId) != null && isolatedTenants.get(tenantId).contains(serviceType); - TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); - tpi.topic(partitionTopics.get(serviceType)); - tpi.partition(partition); - if (isolated) { - tpi.tenantId(tenantId); - } - return tpi.build(); - } - @Override public void recalculatePartitions(ServiceInfo currentService, List otherServices) { logServiceInfo(currentService); otherServices.forEach(this::logServiceInfo); - Map> newCircles = new HashMap<>(ServiceType.values().length); - for (ServiceType serverType : ServiceType.values()) { - newCircles.put(serverType, new ConsistentHashCircle<>()); - } - addNode(newCircles, currentService); + Map> circles = new HashMap<>(); + addNode(circles, currentService); for (ServiceInfo other : otherServices) { - addNode(newCircles, other); - TenantId tenantId = getTenantId(other); + TenantId tenantId = getSystemOrIsolatedTenantId(other); + addNode(circles, other); if (!tenantId.isNullUid()) { isolatedTenants.putIfAbsent(tenantId, new HashSet<>()); for (String serviceType : other.getServiceTypesList()) { @@ -149,12 +134,14 @@ public class ConsistentHashPartitionService implements PartitionService { } } ConcurrentMap> oldPartitions = myPartitions; + TenantId myTenantId = getSystemOrIsolatedTenantId(currentService); myPartitions = new ConcurrentHashMap<>(); partitionSizes.forEach((type, size) -> { + ServiceKey myServiceKey = new ServiceKey(type, myTenantId); for (int i = 0; i < size; i++) { - ServiceInfo serviceInfo = resolveByPartitionIdx(newCircles.get(type), i); + ServiceInfo serviceInfo = resolveByPartitionIdx(circles.get(myServiceKey), i); if (currentService.equals(serviceInfo)) { - ServiceKey serviceKey = new ServiceKey(type, getTenantId(serviceInfo)); + ServiceKey serviceKey = new ServiceKey(type, getSystemOrIsolatedTenantId(serviceInfo)); myPartitions.computeIfAbsent(serviceKey, key -> new ArrayList<>()).add(i); } } @@ -165,13 +152,81 @@ public class ConsistentHashPartitionService implements PartitionService { Set tpiList = partitions.stream() .map(partition -> buildTopicPartitionInfo(serviceKey, partition)) .collect(Collectors.toSet()); + // Adding notifications topic for every @TopicPartitionInfo list + tpiList.add(getNotificationsTopic(serviceKey.getServiceType(), serviceInfoProvider.getServiceId())); applicationEventPublisher.publishEvent(new PartitionChangeEvent(this, serviceKey, tpiList)); } + }); + + if (currentOtherServices == null) { + currentOtherServices = new ArrayList<>(otherServices); + } else { + Set changes = new HashSet<>(); + Map> currentMap = getServiceKeyListMap(currentOtherServices); + Map> newMap = getServiceKeyListMap(otherServices); + currentOtherServices = otherServices; + currentMap.forEach((key, list) -> { + if (!list.equals(newMap.get(key))) { + changes.add(key); + + } + }); + currentMap.keySet().forEach(newMap::remove); + changes.addAll(newMap.keySet()); + if (!changes.isEmpty()) { + applicationEventPublisher.publishEvent(new ClusterTopologyChangeEvent(this, changes)); + } + } + } + + private Map> getServiceKeyListMap(List services) { + final Map> currentMap = new HashMap<>(); + services.forEach(serviceInfo -> { + for (String serviceTypeStr : serviceInfo.getServiceTypesList()) { + ServiceType serviceType = ServiceType.valueOf(serviceTypeStr.toUpperCase()); + ServiceKey serviceKey = new ServiceKey(serviceType, getSystemOrIsolatedTenantId(serviceInfo)); + currentMap.computeIfAbsent(serviceKey, key -> new ArrayList<>()).add(serviceInfo); + } + }); + return currentMap; + } + + @Override + public TopicPartitionInfo getNotificationsTopic(ServiceType serviceType, String serviceId) { + switch (serviceType) { + case TB_CORE: + return tbCoreNotificationTopics.computeIfAbsent(serviceId, + id -> buildTopicPartitionInfo(serviceType, serviceId)); + case TB_RULE_ENGINE: + return tbRuleEngineNotificationTopics.computeIfAbsent(serviceId, + id -> buildTopicPartitionInfo(serviceType, serviceId)); + default: + return buildTopicPartitionInfo(serviceType, serviceId); + } + } + + private TopicPartitionInfo buildTopicPartitionInfo(ServiceType serviceType, String serviceId) { + return new TopicPartitionInfo(serviceType.name().toLowerCase() + "." + serviceId, null, null); + } + + private TopicPartitionInfo buildTopicPartitionInfo(ServiceKey serviceKey, int partition) { + return buildTopicPartitionInfo(serviceKey.getServiceType(), serviceKey.getTenantId(), partition); + } + + private TopicPartitionInfo buildTopicPartitionInfo(ServiceType serviceType, TenantId tenantId, int partition) { + boolean isolated = isolatedTenants.get(tenantId) != null && isolatedTenants.get(tenantId).contains(serviceType); + TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); + tpi.topic(partitionTopics.get(serviceType)); + tpi.partition(partition); + if (isolated) { + tpi.tenantId(tenantId); + } + return tpi.build(); } private void logServiceInfo(TransportProtos.ServiceInfo server) { - TenantId tenantId = getTenantId(server); + TenantId tenantId = getSystemOrIsolatedTenantId(server); if (tenantId.isNullUid()) { log.info("[{}] Found common server: [{}]", server.getServiceId(), server.getServiceTypesList()); } else { @@ -179,21 +234,23 @@ public class ConsistentHashPartitionService implements PartitionService { } } - private TenantId getTenantId(TransportProtos.ServiceInfo serviceInfo) { + private TenantId getSystemOrIsolatedTenantId(TransportProtos.ServiceInfo serviceInfo) { return new TenantId(new UUID(serviceInfo.getTenantIdMSB(), serviceInfo.getTenantIdLSB())); } - private void addNode(Map> circles, ServiceInfo instance) { + private void addNode(Map> circles, ServiceInfo instance) { + TenantId tenantId = getSystemOrIsolatedTenantId(instance); for (String serviceTypeStr : instance.getServiceTypesList()) { ServiceType serviceType = ServiceType.valueOf(serviceTypeStr.toUpperCase()); + ServiceKey serviceKey = new ServiceKey(serviceType, tenantId); for (int i = 0; i < virtualNodesSize; i++) { - circles.get(serviceType).put(hash(instance, i).asLong(), instance); + circles.computeIfAbsent(serviceKey, key -> new ConsistentHashCircle<>()).put(hash(instance, i).asLong(), instance); } } } private ServiceInfo resolveByPartitionIdx(ConsistentHashCircle circle, Integer partitionIdx) { - if (circle.isEmpty()) { + if (circle == null || circle.isEmpty()) { return null; } Long hash = hashFunction.newHasher().putInt(partitionIdx).hash().asLong(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java index f8613f5fb4..46864af79f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java @@ -36,4 +36,13 @@ public interface PartitionService { * @param otherServices - all other discovered services {@link org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo} */ void recalculatePartitions(TransportProtos.ServiceInfo currentService, List otherServices); + + /** + * Each Service should start a consumer for messages that target individual service instance based on serviceId. + * This topic is likely to have single partition, and is always assigned to the service. + * @param tbCore + * @param serviceId + * @return + */ + TopicPartitionInfo getNotificationsTopic(ServiceType tbCore, String serviceId); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java index 09c29aa5e1..52086b3058 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index 17f8b9c31c..19e8d64e87 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -239,6 +239,79 @@ message DeviceActorToTransportMsg { ToServerRpcResponseMsg toServerResponse = 7; } +/** + * TB Core Data Structures + */ + +message TbSubscriptionProto { + string serviceId = 1; + string sessionId = 2; + int32 subscriptionId = 3; + string entityType = 4; + int64 tenantIdMSB = 5; + int64 tenantIdLSB = 6; + int64 entityIdMSB = 7; + int64 entityIdLSB = 8; +} + +message TbTimeSeriesSubscriptionProto { + TbSubscriptionProto sub = 1; + bool allKeys = 2; + repeated TbSubscriptionKetStateProto keyStates = 3; + int64 startTime = 4; + int64 endTime = 5; +} + +message TbAttributeSubscriptionProto { + TbSubscriptionProto sub = 1; + bool allKeys = 2; + repeated TbSubscriptionKetStateProto keyStates = 3; + string scope = 4; +} + +message TbSubscriptionUpdateProto { + string sessionId = 1; + int32 subscriptionId = 2; + int32 errorCode = 3; + string errorMsg = 4; + repeated TbSubscriptionUpdateValueListProto data = 5; +} + +message TbAttributeUpdateProto { + string entityType = 1; + int64 entityIdMSB = 2; + int64 entityIdLSB = 3; + int64 tenantIdMSB = 4; + int64 tenantIdLSB = 5; + string scope = 6; + repeated TsKvProto data = 7; +} + +message TbTimeSeriesUpdateProto { + string entityType = 1; + int64 entityIdMSB = 2; + int64 entityIdLSB = 3; + int64 tenantIdMSB = 4; + int64 tenantIdLSB = 5; + repeated TsKvProto data = 6; +} + +message TbSubscriptionCloseProto { + string sessionId = 1; + int32 subscriptionId = 2; +} + +message TbSubscriptionKetStateProto { + string key = 1; + int64 ts = 2; +} + +message TbSubscriptionUpdateValueListProto { + string key = 1; + repeated int64 ts = 2; + repeated string value = 3; +} + /** * TB Core to TB Core messages */ @@ -253,27 +326,41 @@ message DeviceStateServiceMsgProto { bool deleted = 7; } +message SubscriptionMgrMsgProto { + TbTimeSeriesSubscriptionProto telemetrySub = 1; + TbAttributeSubscriptionProto attributeSub = 2; + TbSubscriptionCloseProto subClose = 3; + TbTimeSeriesUpdateProto tsUpdate = 4; + TbAttributeUpdateProto attrUpdate = 5; +} + +message LocalSubscriptionServiceMsgProto { + TbSubscriptionUpdateProto subUpdate = 1; +} + /** * Main messages; */ /* Request from Transport Service to ThingsBoard Core Service */ message TransportApiRequestMsg { - ValidateDeviceTokenRequestMsg validateTokenRequestMsg = 1; - ValidateDeviceX509CertRequestMsg validateX509CertRequestMsg = 2; - GetOrCreateDeviceFromGatewayRequestMsg getOrCreateDeviceRequestMsg = 3; + ValidateDeviceTokenRequestMsg validateTokenRequestMsg = 1; + ValidateDeviceX509CertRequestMsg validateX509CertRequestMsg = 2; + GetOrCreateDeviceFromGatewayRequestMsg getOrCreateDeviceRequestMsg = 3; } /* Response from ThingsBoard Core Service to Transport Service */ message TransportApiResponseMsg { - ValidateDeviceCredentialsResponseMsg validateTokenResponseMsg = 1; - GetOrCreateDeviceFromGatewayResponseMsg getOrCreateDeviceResponseMsg = 2; + ValidateDeviceCredentialsResponseMsg validateTokenResponseMsg = 1; + GetOrCreateDeviceFromGatewayResponseMsg getOrCreateDeviceResponseMsg = 2; } /* Messages that are handled by ThingsBoard Core Service */ message ToCoreMsg { TransportToDeviceActorMsg toDeviceActorMsg = 1; DeviceStateServiceMsgProto deviceStateServiceMsg = 2; + SubscriptionMgrMsgProto toSubscriptionMgrMsg = 3; + LocalSubscriptionServiceMsgProto toLocalSubscriptionServiceMsg = 4; } /* Messages that are handled by ThingsBoard RuleEngine Service */ diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index fcb630d041..d8d827aa52 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -136,6 +136,7 @@ public class DefaultTransportService implements TransportService { log.warn("Failed to process the notification.", e); } }); + transportNotificationsConsumer.commit(); } catch (Exception e) { log.warn("Failed to obtain messages from queue.", e); try { diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java index f57849fa35..d2e19652a2 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java @@ -44,6 +44,4 @@ public interface RuleEngineTelemetryService { void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value, FutureCallback callback); - void onSharedAttributesUpdate(TenantId tenantId, DeviceId deviceId, Set attributes); - } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index 46a68d3e26..4ddd082f3e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -65,9 +65,6 @@ public class TbMsgAttributesNode implements TbNode { String src = msg.getData(); Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(src)); ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), msg.getOriginator(), config.getScope(), new ArrayList<>(attributes), new TelemetryNodeCallback(ctx, msg)); - if (msg.getOriginator().getEntityType() == EntityType.DEVICE && DataConstants.SHARED_SCOPE.equals(config.getScope())) { - ctx.getTelemetryService().onSharedAttributesUpdate(ctx.getTenantId(), new DeviceId(msg.getOriginator().getId()), attributes); - } } @Override