fixed 固件升级任务

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

@@ -39,6 +39,22 @@ import javax.validation.constraints.Size;
 })
 //@Builder
 public class DeviceUpgradeHistoryEntity extends GenericEntity<String> {
+    public DeviceUpgradeHistoryEntity( String productId, String deviceId, String deviceName, String firmwareId, String progress,
+                                       String taskId, String taskName, String timeoutSeconds, String version,
+                                       String versionOrder, Long lastUpdateTime, FirmwareUpgradeState state) {
+        this.productId = productId;
+        this.deviceId = deviceId;
+        this.deviceName = deviceName;
+        this.firmwareId = firmwareId;
+        this.progress = progress;
+        this.taskId = taskId;
+        this.taskName = taskName;
+        this.timeoutSeconds = timeoutSeconds;
+        this.version = version;
+        this.versionOrder = versionOrder;
+        this.lastUpdateTime = lastUpdateTime;
+        this.state = state;
+    }
 
     public DeviceUpgradeHistoryEntity() {
     }

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

@@ -7,6 +7,7 @@ import org.hswebframework.ezorm.rdb.mapping.ReactiveUpdate;
 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.DeviceInstanceEntity;
 import org.jetlinks.community.device.entity.DeviceUpgradeHistoryEntity;
 import org.jetlinks.community.device.enums.FirmwareUpgradeState;
 import org.jetlinks.community.device.enums.TaskMode;
@@ -55,22 +56,31 @@ public class LocalFirmwareUpgradeHistoryService extends GenericReactiveCrudServi
                 //平台推送
                 firmwareTaskService.createQuery()
                     .where(DeviceFirmwareTaskEntity::getMode,TaskMode.push)
-                    .where(DeviceFirmwareTaskEntity::getProductId, message.getHeader("productId").get()).fetchOne())
-                .flatMap(tp3 -> {
-                    DeviceOperator operator = tp3.getT1();
-                    DeviceFirmwareEntity firmware = tp3.getT2();
-                    if (firmwareTaskService.canUpgrade(tp3.getT3(),operator.getDeviceId())) {
+                    .where(DeviceFirmwareTaskEntity::getProductId, message.getHeader("productId").get()).fetchOne(),
+                deviceService.findById(message.getDeviceId()))
+                .flatMap(tp4 -> {
+                    DeviceOperator operator = tp4.getT1();
+                    DeviceFirmwareEntity firmware = tp4.getT2();
+                    DeviceFirmwareTaskEntity task = tp4.getT3();
+                    DeviceInstanceEntity instance = tp4.getT4();
+                    if (firmwareTaskService.canUpgrade(tp4.getT3(),operator.getDeviceId())) {
                         //该设备可以升级
                         if (!firmware.getVersion().equals(message.getVersion())) {
-                            //如果存在平台推送且版本不一致，则推送升级固件消息，等待升级
+                            //如果存在平台推送且版本不一致，则推送升级固件消息，todo 并创建等待升级任务
+                            long now = System.currentTimeMillis();
+                            DeviceUpgradeHistoryEntity history= new DeviceUpgradeHistoryEntity(firmware.getProductId(),
+                                message.getDeviceId(), instance.getName(), firmware.getId(), "0", task.getId(), task.getName(), task.getTimeoutSeconds(), firmware.getVersion(), firmware.getVersionOrder(), now, FirmwareUpgradeState.waiting);
+
                             UpgradeFirmwareMessage firmwareMessage = new UpgradeFirmwareMessage();
                             firmwareMessage.setDeviceId(message.getDeviceId());
                             firmwareMessage.setUrl(firmware.getUrl());
                             firmwareMessage.setVersion(firmware.getVersion());
                             firmwareMessage.setSign(firmware.getSign());
                             firmwareMessage.setSignMethod(firmware.getSignMethod());
-                            firmwareMessage.setTimestamp(System.currentTimeMillis());
-                            return operator.messageSender().sendAndForget(firmwareMessage);
+                            firmwareMessage.setTimestamp(now);
+                            firmwareMessage.setFirmwareId(firmware.getId());
+                            return operator.messageSender().sendAndForget(firmwareMessage)
+                                .zipWith(Mono.defer(()->this.insert(Mono.just(history))));
                         }
                     }
                     return Mono.empty();
@@ -171,18 +181,27 @@ public class LocalFirmwareUpgradeHistoryService extends GenericReactiveCrudServi
                 DeviceFirmwareEntity firmware = tp5.getT3();
                 DeviceMessageSender sender = tp5.getT2();
                 DeviceFirmwareTaskEntity task = tp5.getT5();
+                DeviceInstanceEntity instance = tp5.getT4();
                 if(firmwareTaskService.canUpgrade(task,messages.getDeviceId())){
+                    //创建升级任务
+                    long now = System.currentTimeMillis();
+                    DeviceUpgradeHistoryEntity history= new DeviceUpgradeHistoryEntity(firmware.getProductId(),
+                        messages.getDeviceId(), instance.getName(), firmware.getId(), "0", task.getId(), task.getName(), task.getTimeoutSeconds(), firmware.getVersion(), firmware.getVersionOrder(), now, FirmwareUpgradeState.waiting);
+
                     RequestFirmwareMessageReply reply = new RequestFirmwareMessageReply();
-                    reply.setTimestamp(System.currentTimeMillis());
+                    reply.setTimestamp(now);
                     reply.setDeviceId(messages.getDeviceId());
                     reply.setSign(firmware.getSign());
                     reply.setSignMethod(firmware.getSignMethod());
+                    //todo 修改文件大小
                     reply.setSize(10L);
                     reply.setUrl(firmware.getUrl());
                     reply.setMessageId(messages.getMessageId());
-                    return sender.sendAndForget(reply);
+                    return sender.sendAndForget(reply)
+                        .zipWith(Mono.defer(()->this.insert(Mono.just(history))));
                 }
                 return Mono.empty();
-            }).then();
+            })
+            .then();
     }
 }

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/DeviceCategoryController.java

@@ -6,14 +6,14 @@ import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.api.crud.entity.TreeSupportEntity;
+import org.hswebframework.web.crud.service.ReactiveCrudService;
+import org.hswebframework.web.crud.web.reactive.ReactiveServiceCrudController;
 import org.hswebframework.web.exception.BusinessException;
 import org.jetlinks.community.device.entity.DeviceCategory;
 import org.jetlinks.community.device.entity.DeviceProductEntity;
 import org.jetlinks.community.device.service.DeviceCategoryService;
 import org.jetlinks.community.device.service.LocalDeviceProductService;
-import org.reactivestreams.Subscription;
 import org.springframework.web.bind.annotation.*;
-import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -24,21 +24,26 @@ import java.util.function.Function;
 @Slf4j
 @AllArgsConstructor
 @Tag(name = "设备分类目录")
-public class DeviceCategoryController{
+public class DeviceCategoryController implements
+    ReactiveServiceCrudController<DeviceCategory, String> {
 
 
     private final LocalDeviceProductService productService;
     private final DeviceCategoryService service;
-
-    @PostMapping
-    @Operation(summary = "添加目录")
-    public Mono<?>  add(@RequestBody Mono<DeviceCategory> payload){
-        return this.service.insert(payload)
-            .thenReturn(payload)
-            .onErrorMap(e->new BusinessException("服务器繁忙，请稍后重试",e))
-            ;
+    @Override
+    public ReactiveCrudService<DeviceCategory, String> getService() {
+        return service;
     }
 
+//    @PostMapping
+//    @Operation(summary = "添加目录")
+//    public Mono<?>  add(@RequestBody Mono<DeviceCategory> payload){
+//        return this.service.insert(payload)
+//            .thenReturn(payload)
+//            .onErrorMap(e->new BusinessException("服务器繁忙，请稍后重试",e))
+//            ;
+//    }
+
     @DeleteMapping("/{id}")
     @Operation(summary = "删除目录")
     public Mono<?> remove(@PathVariable @Parameter(description = "设备ID") String id){
@@ -57,17 +62,17 @@ public class DeviceCategoryController{
     }
 
 
-    @PatchMapping
-    @Operation(summary = "更新目录")
-    public Mono<Void> updata(@RequestBody Mono<DeviceCategory> payload){
-        return payload.flatMap(data->
-            this.service.createUpdate()
-                .set(data)
-                .where(DeviceCategory::getId,data.getId())
-                .execute()
-                .then()
-        );
-    }
+//    @PatchMapping
+//    @Operation(summary = "更新目录")
+//    public Mono<Void> updata(@RequestBody Mono<DeviceCategory> payload){
+//        return payload.flatMap(data->
+//            this.service.createUpdate()
+//                .set(data)
+//                .where(DeviceCategory::getId,data.getId())
+//                .execute()
+//                .then()
+//        );
+//    }
 
     @GetMapping("/{id}/detail")
     @Operation(summary = "目录详情")
@@ -85,6 +90,7 @@ public class DeviceCategoryController{
         //        return Flux.fromIterable(statics);
     }
 
+
     @GetMapping("/_tree")
     @Operation(summary = "获取全部分类目录(树结构)")
     public Flux<DeviceCategory> getAllCategoryTree() {