优化导入

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

@@ -389,86 +389,4 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                 }));
     }
 
-    public Flux<ImportDeviceInstanceResult> doBatchImport(String fileUrl) {
-        return deviceProductService
-            .createQuery()
-            .fetch()
-            .collectList()
-            .flatMapMany(productEntities -> {
-                Map<String, String> productNameMap = productEntities.stream()
-                    .collect(Collectors.toMap(DeviceProductEntity::getName, DeviceProductEntity::getId, (_1, _2) -> _1));
-                return importExportService
-                    .doImport(DeviceInstanceImportExportEntity.class, fileUrl)
-                    .map(result -> {
-                        try {
-                            DeviceInstanceImportExportEntity importExportEntity = result.getResult();
-                            DeviceInstanceEntity entity = FastBeanCopier.copy(importExportEntity, new DeviceInstanceEntity());
-                            String productId = productNameMap.get(importExportEntity.getProductName());
-                            if (StringUtils.isEmpty(productId)) {
-                                throw new BusinessException("设备型号不存在");
-                            }
-                            if (StringUtils.isEmpty(entity.getId())) {
-                                throw new BusinessException("设备ID不能为空");
-                            }
-
-                            entity.setProductId(productId);
-                            entity.setState(DeviceState.notActive);
-                            return entity;
-                        } catch (Throwable e) {
-                            throw new BusinessException("第" +
-                                (result.getRowIndex() + 2)
-                                + "行:" + e.getMessage());
-                        }
-                    });
-            })
-            .buffer(50)
-            .flatMap(list -> this.save(Flux.fromIterable(list)))
-            .map(ImportDeviceInstanceResult::success)
-            .onErrorResume(err -> Mono.just(ImportDeviceInstanceResult.error(err)))
-            .doOnEach(ReactiveLogger.on(SignalType.CANCEL, (ctx, signal) -> {
-                log.warn("用户取消导入设备实例:{}", fileUrl);
-            }))
-            ;
-    }
-
-    private DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
-
-    @SneakyThrows
-    public Mono<Void> doExport(ServerHttpResponse response, QueryParam queryParam, String fileName) {
-        response.getHeaders()
-            .set(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=".concat(
-                URLEncoder.encode(fileName, StandardCharsets.UTF_8.displayName())));
-        queryParam.setPaging(false);
-
-        return response.writeWith(Flux.create(sink -> {
-                OutputStream outputStream = new OutputStream() {
-                    @Override
-                    public void write(byte[] b) {
-                        sink.next(bufferFactory.wrap(b));
-                    }
-
-                    @Override
-                    public void write(int b) {
-                        sink.next(bufferFactory.wrap(new byte[]{(byte) b}));
-                    }
-
-                    @Override
-                    public void close() {
-                        sink.complete();
-                    }
-                };
-                ExcelWriter excelWriter = EasyExcel.write(outputStream, DeviceInstanceImportExportEntity.class).build();
-                WriteSheet writeSheet = EasyExcel.writerSheet().build();
-                sink.onDispose(query(queryParam)
-                    .map(entity -> FastBeanCopier.copy(entity, new DeviceInstanceImportExportEntity()))
-                    .buffer(100)
-                    .doOnNext(list -> excelWriter.write(list, writeSheet))
-                    .doFinally(s -> excelWriter.finish())
-                    .doOnError(sink::error)
-                    .subscribe());
-            })
-        );
-    }
-
-
 }

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

@@ -1,7 +1,11 @@
 package org.jetlinks.community.device.web;
 
+import com.alibaba.excel.EasyExcel;
+import com.alibaba.excel.ExcelWriter;
+import com.alibaba.excel.write.metadata.WriteSheet;
 import io.swagger.annotations.ApiOperation;
 import lombok.Getter;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.ezorm.core.param.QueryParam;
 import org.hswebframework.ezorm.core.param.TermType;
@@ -9,12 +13,19 @@ import org.hswebframework.ezorm.rdb.exception.DuplicateKeyException;
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
 import org.hswebframework.web.authorization.Authentication;
+import org.hswebframework.web.authorization.Dimension;
 import org.hswebframework.web.authorization.annotation.Authorize;
 import org.hswebframework.web.authorization.annotation.QueryAction;
 import org.hswebframework.web.authorization.annotation.Resource;
 import org.hswebframework.web.authorization.annotation.SaveAction;
+import org.hswebframework.web.bean.FastBeanCopier;
 import org.hswebframework.web.crud.web.reactive.ReactiveServiceCrudController;
 import org.hswebframework.web.exception.BusinessException;
+import org.jetlinks.community.device.entity.DeviceProductEntity;
+import org.jetlinks.community.device.entity.excel.DeviceInstanceImportExportEntity;
+import org.jetlinks.community.device.enums.DeviceState;
+import org.jetlinks.community.device.service.LocalDeviceProductService;
+import org.jetlinks.community.io.excel.ImportExportService;
 import org.jetlinks.core.device.DeviceConfigKey;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
@@ -26,15 +37,25 @@ import org.jetlinks.community.device.service.LocalDeviceInstanceService;
 import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
 import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.jetlinks.community.timeseries.TimeSeriesMetric;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.StringUtils;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 @RestController
 @RequestMapping({"/device-instance", "/device/instance"})
@@ -51,10 +72,16 @@ public class DeviceInstanceController implements
 
     private final DeviceRegistry registry;
 
-    public DeviceInstanceController(LocalDeviceInstanceService service, TimeSeriesManager timeSeriesManager, DeviceRegistry registry) {
+    private final LocalDeviceProductService productService;
+
+    private final ImportExportService importExportService;
+
+    public DeviceInstanceController(LocalDeviceInstanceService service, TimeSeriesManager timeSeriesManager, DeviceRegistry registry, LocalDeviceProductService productService, ImportExportService importExportService) {
         this.service = service;
         this.timeSeriesManager = timeSeriesManager;
         this.registry = registry;
+        this.productService = productService;
+        this.importExportService = importExportService;
     }
 
     @GetMapping({
@@ -179,7 +206,7 @@ public class DeviceInstanceController implements
     @QueryAction
     public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceLog(@PathVariable String deviceId,
                                                                       QueryParamEntity entity) {
-        return service.queryDeviceLog(deviceId,entity);
+        return service.queryDeviceLog(deviceId, entity);
     }
 
     //已废弃
@@ -197,12 +224,70 @@ public class DeviceInstanceController implements
 
         return Authentication
             .currentReactive()
-            .flatMapMany(auth -> service.doBatchImport(fileUrl));
+            .flatMapMany(auth -> productService
+                .createQuery()
+                .fetch()
+                .collectList()
+                .flatMapMany(productEntities -> {
+                    Map<String, String> productNameMap = productEntities.stream()
+                        .collect(Collectors.toMap(DeviceProductEntity::getName, DeviceProductEntity::getId, (_1, _2) -> _1));
+                    return importExportService
+                        .doImport(DeviceInstanceImportExportEntity.class, fileUrl)
+                        .map(result -> {
+                            try {
+                                DeviceInstanceImportExportEntity importExportEntity = result.getResult();
+                                DeviceInstanceEntity entity = FastBeanCopier.copy(importExportEntity, new DeviceInstanceEntity());
+                                String productId = productNameMap.get(importExportEntity.getProductName());
+                                if (StringUtils.isEmpty(productId)) {
+                                    throw new BusinessException("设备型号不存在");
+                                }
+                                if (StringUtils.isEmpty(entity.getId())) {
+                                    throw new BusinessException("设备ID不能为空");
+                                }
+
+                                entity.setProductId(productId);
+                                entity.setState(DeviceState.notActive);
+                                return entity;
+                            } catch (Throwable e) {
+                                throw new BusinessException("第" +
+                                    (result.getRowIndex() + 2)
+                                    + "行:" + e.getMessage());
+                            }
+                        });
+                })
+                .buffer(20)
+                .publishOn(Schedulers.single())
+                .concatMap(list -> service.save(Flux.fromIterable(list)))
+                .map(ImportDeviceInstanceResult::success));
     }
 
+    DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
+
     @PostMapping("/export")
     @QueryAction
-    public Mono<Void> export(ServerHttpResponse response, QueryParam parameter){
-        return service.doExport(response, parameter, "设备实例.xlsx");
+    @SneakyThrows
+    public Mono<Void> export(ServerHttpResponse response, QueryParam parameter) {
+        response.getHeaders().set(HttpHeaders.CONTENT_DISPOSITION,
+            "attachment; filename=".concat(URLEncoder.encode("设备实例.xlsx", StandardCharsets.UTF_8.displayName())));
+        return Authentication
+            .currentReactive()
+            .flatMap(auth -> {
+                parameter.setPaging(false);
+                return response.writeWith(Mono.create(sink -> {
+                    ByteArrayOutputStream out = new ByteArrayOutputStream();
+                    ExcelWriter excelWriter = EasyExcel.write(out, DeviceInstanceImportExportEntity.class).build();
+                    WriteSheet writeSheet = EasyExcel.writerSheet().build();
+                    service.query(parameter)
+                        .map(entity -> FastBeanCopier.copy(entity, new DeviceInstanceImportExportEntity()))
+                        .buffer(100)
+                        .doOnNext(list -> excelWriter.write(list, writeSheet))
+                        .doFinally(s -> {
+                            excelWriter.finish();
+                            sink.success(bufferFactory.wrap(out.toByteArray()));
+                        })
+                        .doOnError(sink::error)
+                        .subscribe();
+                }));
+            });
     }
 }