#Based on modifications of the ThingsBoard Community Edition.

fix: 同步thingsboard3.8.1设备状态的逻辑


(cherry picked from commit 333ca4c93424fba4cf1d44aa3c57fe8a50f3a2e6)

@@ -18,7 +18,9 @@
  * Source: ThingsBoard Community Edition
  * Modifications made by: Chengdu Yunteng Wuzhou Technology Co., Ltd
  * Modification date: 2022-09-20
- * Description of changes:Add device upper and lower limit log records
+ * Description of changes:
+ * 1、Add device upper and lower limit log records
+ * 2、merged https://github.com/thingsboard/thingsboard/pull/11536
  */ 
 package org.thingsboard.server.service.state;
 
@@ -385,14 +387,16 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                     if (proto.getAdded()) {
                         Futures.addCallback(fetchDeviceState(device), new FutureCallback<>() {
                             @Override
-                            public void onSuccess(@Nullable DeviceStateData state) {
+                            public void onSuccess(DeviceStateData state) {
                                 TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, device.getId());
-                                if (addDeviceUsingState(tpi, state)) {
-                                    save(deviceId, ACTIVITY_STATE, false);
+                                Set<DeviceId> deviceIds = partitionedEntities.get(tpi);
+                                boolean isMyPartition = deviceIds != null;
+                                if (isMyPartition) {
+                                    deviceIds.add(state.getDeviceId());
+                                    initializeActivityState(deviceId, state);
                                     callback.onSuccess();
                                 } else {
-                                    log.debug("[{}][{}] Device belongs to external partition. Probably rebalancing is in progress. Topic: {}"
-                                            , tenantId, deviceId, tpi.getFullTopicName());
+                                    log.debug("[{}][{}] Device belongs to external partition. Probably rebalancing is in progress. Topic: {}", tenantId, deviceId, tpi.getFullTopicName());
                                     callback.onFailure(new RuntimeException("Device belongs to external partition " + tpi.getFullTopicName() + "!"));
                                 }
                             }
@@ -423,6 +427,21 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         }
     }
 
+    private void onDeviceDeleted(TenantId tenantId, DeviceId deviceId) {
+        cleanupEntity(deviceId);
+        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId);
+        Set<DeviceId> deviceIdSet = partitionedEntities.get(tpi);
+        if (deviceIdSet != null) {
+            deviceIdSet.remove(deviceId);
+        }
+    }
+
+    private void initializeActivityState(DeviceId deviceId, DeviceStateData fetchedState) {
+        DeviceStateData cachedState = deviceStates.putIfAbsent(fetchedState.getDeviceId(), fetchedState);
+        boolean activityState = Objects.requireNonNullElse(cachedState, fetchedState).getState().isActive();
+        save(deviceId, ACTIVITY_STATE, activityState);
+    }
+
     @Override
     protected Map<TopicPartitionInfo, List<ListenableFuture<?>>> onAddedPartitions(Set<TopicPartitionInfo> addedPartitions) {
         var result = new HashMap<TopicPartitionInfo, List<ListenableFuture<?>>>();
@@ -459,10 +478,16 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                         }
                         if (devicePackFutureHolder.future == null || !devicePackFutureHolder.future.isCancelled()) {
                             for (var state : states) {
-                                if (!addDeviceUsingState(entry.getKey(), state)) {
-                                    return;
+                                TopicPartitionInfo tpi = entry.getKey();
+                                Set<DeviceId> deviceIds = partitionedEntities.get(tpi);
+                                boolean isMyPartition = deviceIds != null;
+                                if (isMyPartition) {
+                                    deviceIds.add(state.getDeviceId());
+                                    deviceStates.putIfAbsent(state.getDeviceId(), state);
+                                    checkAndUpdateState(state.getDeviceId(), state);
+                                } else {
+                                    log.debug("[{}] Device belongs to external partition {}", state.getDeviceId(), tpi.getFullTopicName());
                                 }
-                                checkAndUpdateState(state.getDeviceId(), state);
                             }
                             log.info("[{}] Initialized {} out of {} device states", entry.getKey().getPartition().orElse(0), counter.addAndGet(states.size()), entry.getValue().size());
                         }
@@ -519,18 +544,6 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         }
     }
 
-    private boolean addDeviceUsingState(TopicPartitionInfo tpi, DeviceStateData state) {
-        Set<DeviceId> deviceIds = partitionedEntities.get(tpi);
-        if (deviceIds != null) {
-            deviceIds.add(state.getDeviceId());
-            deviceStates.putIfAbsent(state.getDeviceId(), state);
-            return true;
-        } else {
-            log.debug("[{}] Device belongs to external partition {}", state.getDeviceId(), tpi.getFullTopicName());
-            return false;
-        }
-    }
-
     void checkStates() {
         try {
             final long ts = getCurrentTimeMillis();
@@ -671,15 +684,6 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         return cleanup;
     }
 
-    private void onDeviceDeleted(TenantId tenantId, DeviceId deviceId) {
-        cleanupEntity(deviceId);
-        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId);
-        Set<DeviceId> deviceIdSet = partitionedEntities.get(tpi);
-        if (deviceIdSet != null) {
-            deviceIdSet.remove(deviceId);
-        }
-    }
-
     @Override
     protected void cleanupEntityOnPartitionRemoval(DeviceId deviceId) {
         cleanupEntity(deviceId);