优化设备状态同步逻辑

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/LocalDeviceInstanceService.java

@@ -60,6 +60,7 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
     public LocalDeviceInstanceService(DeviceRegistry registry,
                                       LocalDeviceProductService deviceProductService,
                                       DeviceConfigMetadataManager metadataManager,
+                                      @SuppressWarnings("all")
                                       ReactiveRepository<DeviceTagEntity, String> tagRepository) {
         this.registry = registry;
         this.deviceProductService = deviceProductService;
@@ -327,20 +328,20 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                 .collect(Collectors.groupingBy(Tuple2::getT1))
                 .flatMapIterable(Map::entrySet)
                 .flatMap(group -> {
-                    List<String> deviceId = group
+                    List<String> deviceIdList = group
                         .getValue()
                         .stream()
                         .map(Tuple3::getT2)
                         .collect(Collectors.toList());
                     DeviceState state = DeviceState.of(group.getKey());
-                    return Mono
-                        .zip(
+                    return Flux
+                        .concat(
                             //批量修改设备状态
-                            this.getRepository()
+                            getRepository()
                                 .createUpdate()
                                 .set(DeviceInstanceEntity::getState, state)
                                 .where()
-                                .in(DeviceInstanceEntity::getId, deviceId)
+                                .in(DeviceInstanceEntity::getId, deviceIdList)
                                 .execute()
                                 .thenReturn(group.getValue().size()),
                             //修改子设备状态
@@ -355,6 +356,8 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                                     .set(DeviceInstanceEntity::getState, state)
                                     .where()
                                     .in(DeviceInstanceEntity::getParentId, parents)
+                                    //不修改未激活的状态
+                                    .not(DeviceInstanceEntity::getState, DeviceState.notActive)
                                     .nest()
                                     /* */.accept(DeviceInstanceEntity::getFeatures, Terms.Enums.notInAny, DeviceFeature.selfManageState)
                                     /* */.or()
@@ -363,10 +366,12 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                                     .execute())
                                 .defaultIfEmpty(0)
                         )
-                        .thenReturn(deviceId
-                                        .stream()
-                                        .map(id -> DeviceStateInfo.of(id, state))
-                                        .collect(Collectors.toList()));
+                        .then(Mono.just(
+                            deviceIdList
+                                .stream()
+                                .map(id -> DeviceStateInfo.of(id, state))
+                                .collect(Collectors.toList())
+                        ));
                 }));
     }