changed 固件升级优化

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/entity/DeviceUpgradeHistoryEntity.java

@@ -158,4 +158,18 @@ public class DeviceUpgradeHistoryEntity extends GenericEntity<String> {
         ,accessMode = Schema.AccessMode.READ_WRITE
     )
     private FirmwareUpgradeState state;
+
+    public static DeviceUpgradeHistoryEntity of(DeviceFirmwareTaskEntity task,DeviceFirmwareEntity firmwareEntity,String deviceId){
+        DeviceUpgradeHistoryEntity historyEntity = new DeviceUpgradeHistoryEntity();
+        historyEntity.setTimeoutSeconds(task.getTimeoutSeconds());
+        historyEntity.setTaskName(task.getName());
+        historyEntity.setTaskId(task.getId());
+        historyEntity.setState(FirmwareUpgradeState.waiting);
+        historyEntity.setDeviceId(deviceId);
+        historyEntity.setFirmwareId(task.getFirmwareId());
+        historyEntity.setVersion(firmwareEntity.getVersion());
+        historyEntity.setVersionOrder(firmwareEntity.getVersionOrder());
+        historyEntity.setProductId(firmwareEntity.getProductId());
+        return historyEntity;
+    }
 }

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

@@ -1,15 +1,25 @@
 package org.jetlinks.community.device.service;
 
 import cn.hutool.core.collection.CollectionUtil;
+import javafx.concurrent.Task;
+import lombok.AllArgsConstructor;
 import org.hswebframework.web.crud.service.GenericReactiveCrudService;
 import org.jetlinks.community.device.entity.DeviceFirmwareEntity;
 import org.jetlinks.community.device.entity.DeviceFirmwareTaskEntity;
+import org.jetlinks.community.device.entity.DeviceUpgradeHistoryEntity;
 import org.jetlinks.community.device.enums.TaskDeployMode;
+import org.jetlinks.community.device.enums.TaskMode;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import reactor.core.publisher.Mono;
 
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * @author lifang
@@ -20,7 +30,12 @@ import java.util.List;
  */
 @Service
 public class LocalDeviceFirmwareTaskService extends GenericReactiveCrudService<DeviceFirmwareTaskEntity, String> {
-
+    @Autowired
+    @Lazy
+    private LocalFirmwareUpgradeHistoryService upgradeHistoryService;
+    @Autowired
+    @Lazy
+    private LocalDeviceFirmwareService firmwareService;
     public Mono<Integer> deployAll(String id) {
         return this
             .createUpdate()
@@ -29,13 +44,48 @@ public class LocalDeviceFirmwareTaskService extends GenericReactiveCrudService<D
             .execute();
     }
 
+    /**
+     * @param id
+     * @param deviceIds
+     * @return
+     */
+    @Transactional(rollbackFor = Exception.class)
     public Mono<Integer> deploy(String id,List<String> deviceIds) {
-        return this
-            .createUpdate()
-            .where(DeviceFirmwareEntity::getId,id)
-            .set(DeviceFirmwareTaskEntity::getDeployMode, TaskDeployMode.special)
-            .set(DeviceFirmwareTaskEntity::getDeployIds,deviceIds)
-            .execute();
+        return this.findById(id)
+            .flatMap(task-> Mono.zip(Mono.just(task),firmwareService.findById(task.getFirmwareId())))
+            .flatMap(tp2->{
+                DeviceFirmwareTaskEntity task = tp2.getT1();
+                DeviceFirmwareEntity firmware = tp2.getT2();
+                //当为平台推送时，所有未推送的设备应为等待升级状态
+                if(TaskMode.push.eq(task.getMode())){
+                   return  upgradeHistoryService.createQuery()
+                        .where(DeviceUpgradeHistoryEntity::getTaskId,task.getId())
+                        .where(DeviceUpgradeHistoryEntity::getFirmwareId,task.getFirmwareId())
+                        .fetch()
+                        .collectList()
+                        .map(existDevices->existDevices.stream().map(DeviceUpgradeHistoryEntity::getDeviceId).collect(Collectors.toList()))
+                        .defaultIfEmpty(new ArrayList<String>())
+                        //获取新增任务设备id
+                        .map(existDeviceIds-> CollectionUtil.subtract(deviceIds, existDeviceIds))
+                        //插入设备任务操作
+                        .flatMap(ids->
+                            upgradeHistoryService.save( ids.stream().map(deviceId-> DeviceUpgradeHistoryEntity.of(task,firmware,deviceId)).collect(Collectors.toList()))
+                        ).then(this
+                        .createUpdate()
+                        .where(DeviceFirmwareEntity::getId,id)
+                        .set(DeviceFirmwareTaskEntity::getDeployMode, TaskDeployMode.special)
+                        .set(DeviceFirmwareTaskEntity::getDeployIds,deviceIds)
+                        .execute());
+                }else {
+                    //当为设备拉取时，直接更新即可
+                    return  this
+                        .createUpdate()
+                        .where(DeviceFirmwareEntity::getId,id)
+                        .set(DeviceFirmwareTaskEntity::getDeployMode, TaskDeployMode.special)
+                        .set(DeviceFirmwareTaskEntity::getDeployIds,deviceIds)
+                        .execute();
+                }
+            });
     }
 
     /**

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/ClusterConfiguration.java

@@ -21,6 +21,7 @@ import org.jetlinks.supports.config.ClusterConfigStorageManager;
 import org.redisson.api.RedissonClient;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
@@ -36,11 +37,13 @@ import org.springframework.context.annotation.Configuration;
  */
 @Configuration
 @AllArgsConstructor
+@AutoConfigureAfter(DeviceRegistry.class)
 @ConditionalOnProperty(name = "jetlinks.cluster",havingValue = "true")
 public class ClusterConfiguration {
 
     @Getter
     private RedisProperties properties;
+
     @Bean
     public ClusterDeviceMessageBrokeMessageBroker clusterDeviceMessageBrokeMessageBroker(JetLinksProperties properties, ClusterManager clusterManager) {
         return new ClusterDeviceMessageBrokeMessageBroker(properties.getServerId(),clusterManager);

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/ClusterDeviceMessageBrokeMessageBroker.java

@@ -1,18 +1,25 @@
 package org.jetlinks.community.standalone.configuration.cluster;
 
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.extra.spring.SpringUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.core.cluster.ClusterTopic;
 import org.jetlinks.core.cluster.message.ClusterMessage;
 import org.jetlinks.core.cluster.ClusterManager;
 import org.jetlinks.core.cluster.ServerNode;
+import org.jetlinks.core.device.DeviceOperator;
+import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.device.StandaloneDeviceMessageBroker;
 import org.jetlinks.core.enums.ErrorCode;
 import org.jetlinks.core.exception.DeviceOperationException;
 import org.jetlinks.core.message.*;
 import org.reactivestreams.Publisher;
+import org.springframework.util.StringUtils;
 import reactor.core.publisher.*;
 import reactor.core.scheduler.Schedulers;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 /**
@@ -29,6 +36,7 @@ public class ClusterDeviceMessageBrokeMessageBroker extends StandaloneDeviceMess
 
     private ClusterManager clusterManager;
 
+    private DeviceRegistry deviceRegistry;
     private List<ServerNode> getAllNode(){
         return clusterManager.getHaManager().getAllNode();
     }
@@ -36,6 +44,12 @@ public class ClusterDeviceMessageBrokeMessageBroker extends StandaloneDeviceMess
     public ClusterDeviceMessageBrokeMessageBroker(String serverId, ClusterManager clusterManager) {
         this.serverId = serverId;
         this.clusterManager = clusterManager;
+        clusterManager.getTopic("_reply_"+serverId)
+            .subscribePattern()
+            .map(ClusterTopic.TopicMessage::getMessage)
+            .cast(DeviceMessageReply.class)
+            .flatMap(super::reply)
+            .subscribe();
     }
 
     @Override
@@ -48,6 +62,24 @@ public class ClusterDeviceMessageBrokeMessageBroker extends StandaloneDeviceMess
     }
 
 
+    @Override
+    public Mono<Boolean> reply(DeviceMessageReply message) {
+        if(deviceRegistry==null){
+            deviceRegistry=SpringUtil.getBean(DeviceRegistry.class);
+        }
+        String deviceId = message.getDeviceId();
+        if(StrUtil.isEmpty(deviceId)){
+            return super.reply(message);
+        }
+        return deviceRegistry.getDevice(deviceId)
+            .flatMap(DeviceOperator::getConnectionServerId)
+            .flatMap(connectionServerId->{
+                if(this.serverId.equals(connectionServerId)){
+                    return super.reply(message);
+                }
+                return clusterManager.getTopic("_reply_"+serverId).publish(Mono.just(message)).thenReturn(true);
+            });
+    }
     @Override
     public Mono<Integer> send(String serverId, Publisher<? extends Message> message) {
         if(this.serverId.equals(serverId)){