thingsboard:开源物联网平台thingsboard源码分析

/** * Copyright © 2016-2018 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.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.stereotype.Service; import org.springframework.util.StringUtils; import org.thingsboard.rule.engine.api.util.DonAsynchron; 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.EntityIdFactory; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseTsKvQuery; 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.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.common.data.kv.TsKvQuery; import org.thingsboard.server.common.msg.cluster.ServerAddress; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.gen.cluster.ClusterAPIProtos; import org.thingsboard.server.service.cluster.routing.ClusterRoutingService; import org.thingsboard.server.service.cluster.rpc.ClusterRpcService; 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 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.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. */ @Service @Slf4j public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptionService { @Autowired private TelemetryWebSocketService wsService; @Autowired private AttributesService attrService; @Autowired private TimeseriesService tsService; @Autowired private ClusterRoutingService routingService; @Autowired private ClusterRpcService rpcService; @Autowired @Lazy private DeviceStateService stateService; private ExecutorService tsCallBackExecutor; private ExecutorService wsCallBackExecutor; @PostConstruct public void initExecutor() { tsCallBackExecutor = Executors.newSingleThreadExecutor(); wsCallBackExecutor = Executors.newSingleThreadExecutor(); } @PreDestroy public void shutdownExecutor() { if (tsCallBackExecutor != null) { tsCallBackExecutor.shutdownNow(); } if (wsCallBackExecutor != null) { wsCallBackExecutor.shutdownNow(); } } private final Map<EntityId, Set<Subscription>> subscriptionsByEntityId = new ConcurrentHashMap<>(); private final Map<String, Map<Integer, Subscription>> subscriptionsByWsSessionId = new ConcurrentHashMap<>(); @Override public void addLocalWsSubscription(String sessionId, EntityId entityId, SubscriptionState sub) { Optional<ServerAddress> server = routingService.resolveById(entityId); Subscription subscription; if (server.isPresent()) { ServerAddress address = server.get(); log.trace("[{}] Forwarding subscription [{}] for device [{}] to [{}]", sessionId, sub.getSubscriptionId(), entityId, address); subscription = new Subscription(sub, true, address); tellNewSubscription(address, sessionId, subscription); } else { log.trace("[{}] Registering local subscription [{}] for device [{}]", sessionId, sub.getSubscriptionId(), entityId); subscription = new Subscription(sub, true); } registerSubscription(sessionId, entityId, subscription); } @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<Integer, Subscription> 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); } } @Override public void saveAndNotify(EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback) { saveAndNotify(entityId, ts, 0L, callback); } @Override public void saveAndNotify(EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> callback) { ListenableFuture<List<Void>> saveFuture = tsService.save(entityId, ts, ttl); addMainCallback(saveFuture, callback); addWsCallback(saveFuture, success -> onTimeseriesUpdate(entityId, ts)); } @Override public void saveAndNotify(EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback) { ListenableFuture<List<Void>> saveFuture = attrService.save(entityId, scope, attributes); addMainCallback(saveFuture, callback); addWsCallback(saveFuture, success -> onAttributesUpdate(entityId, scope, attributes)); } @Override public void saveAttrAndNotify(EntityId entityId, String scope, String key, long value, FutureCallback<Void> callback) { saveAndNotify(entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(key, value) , System.currentTimeMillis())), callback); } @Override pub