增加文件管理

jetlinks-components/io-component/pom.xml

@@ -47,5 +47,11 @@
             <artifactId>hsweb-core</artifactId>
             <version>${hsweb.framework.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.hswebframework.web</groupId>
+            <artifactId>hsweb-commons-crud</artifactId>
+            <version>${hsweb.framework.version}</version>
+        </dependency>
     </dependencies>
 </project>

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/DefaultFileManager.java

@@ -0,0 +1,213 @@
+package org.jetlinks.community.io.file;
+
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
+import lombok.AllArgsConstructor;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
+import org.hswebframework.web.exception.BusinessException;
+import org.hswebframework.web.id.IDGenerator;
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.NettyDataBufferFactory;
+import org.springframework.http.ContentDisposition;
+import org.springframework.http.HttpRange;
+import org.springframework.http.MediaType;
+import org.springframework.http.client.MultipartBodyBuilder;
+import org.springframework.http.codec.multipart.FilePart;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.security.MessageDigest;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.function.Function;
+
+
+public class DefaultFileManager implements FileManager {
+
+    private final FileProperties properties;
+
+    private final DataBufferFactory bufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT);
+
+    private final ReactiveRepository<FileEntity, String> repository;
+
+
+    private final WebClient client;
+
+    public DefaultFileManager(WebClient.Builder builder,
+                              FileProperties properties,
+                              ReactiveRepository<FileEntity, String> repository) {
+        new File(properties.getStorageBasePath()).mkdirs();
+        this.properties = properties;
+        this.client = builder
+            .clone()
+            .filter(this.properties.createWebClientRute())
+            .build();
+        this.repository = repository;
+    }
+
+    @Override
+    public Mono<FileInfo> saveFile(FilePart filePart) {
+        return saveFile(filePart.filename(), filePart.content());
+    }
+
+    private DataBuffer updateDigest(MessageDigest digest, DataBuffer dataBuffer) {
+        dataBuffer = DataBufferUtils.retain(dataBuffer);
+        digest.update(dataBuffer.asByteBuffer());
+        DataBufferUtils.release(dataBuffer);
+        return dataBuffer;
+    }
+
+    public Mono<FileInfo> saveFileToCluster(String name, Flux<DataBuffer> stream) {
+        String serverId = properties.selectServerNode();
+        MultipartBodyBuilder builder = new MultipartBodyBuilder();
+        builder.asyncPart("file", stream, DataBuffer.class)
+               .headers(header -> header
+                   .setContentDisposition(ContentDisposition
+                                              .builder("form-data")
+                                              .name("file")
+                                              .filename(name)
+                                              .build()))
+               .contentType(MediaType.APPLICATION_OCTET_STREAM);
+        return client
+            .post()
+            .uri("http://" + serverId + "/file/" +serverId)
+            .attribute(FileProperties.serverNodeIdAttr, serverId)
+            .contentType(MediaType.MULTIPART_FORM_DATA)
+            .body(BodyInserters.fromMultipartData(builder.build()))
+            .retrieve()
+            .bodyToMono(FileInfo.class);
+    }
+
+    public Mono<FileInfo> doSaveFile(String name, Flux<DataBuffer> stream) {
+        LocalDate now = LocalDate.now();
+        FileInfo fileInfo = new FileInfo();
+        fileInfo.setId(IDGenerator.MD5.generate());
+        fileInfo.withFileName(name);
+
+        String storagePath = now.format(DateTimeFormatter.BASIC_ISO_DATE)
+            + "/" + fileInfo.getId() + "." + fileInfo.getExtension();
+
+        MessageDigest md5 = DigestUtils.getMd5Digest();
+        MessageDigest sha256 = DigestUtils.getSha256Digest();
+        String storageBasePath = properties.getStorageBasePath();
+        String serverNodeId = properties.getServerNodeId();
+        Path path = Paths.get(storageBasePath, storagePath);
+        path.toFile().getParentFile().mkdirs();
+        return stream
+            .map(buffer -> updateDigest(md5, updateDigest(sha256, buffer)))
+            .as(buf -> DataBufferUtils
+                .write(buf, path,
+                       StandardOpenOption.WRITE,
+                       StandardOpenOption.CREATE_NEW,
+                       StandardOpenOption.TRUNCATE_EXISTING))
+            .then(Mono.defer(() -> {
+                File savedFile = Paths.get(storageBasePath, storagePath).toFile();
+                if (!savedFile.exists()) {
+                    return Mono.error(new BusinessException("error.file_storage_failed"));
+                }
+                fileInfo.setMd5(ByteBufUtil.hexDump(md5.digest()));
+                fileInfo.setSha256(ByteBufUtil.hexDump(sha256.digest()));
+                fileInfo.setLength(savedFile.length());
+                fileInfo.setCreateTime(System.currentTimeMillis());
+                FileEntity entity = FileEntity.of(fileInfo, storagePath, serverNodeId);
+                return repository
+                    .insert(entity)
+                    .then(Mono.fromSupplier(entity::toInfo));
+            }));
+    }
+
+    @Override
+    public Mono<FileInfo> saveFile(String name, Flux<DataBuffer> stream) {
+        if (properties.getClusterRute().isEmpty()
+            || properties.getClusterRute().containsKey(properties.getServerNodeId())) {
+            return doSaveFile(name, stream);
+        }
+        //配置里集群,但是并不支持本节点,则保存到其他节点
+        return saveFileToCluster(name, stream);
+    }
+
+    @Override
+    public Mono<FileInfo> getFile(String id) {
+        return repository
+            .findById(id)
+            .map(FileEntity::toInfo);
+    }
+
+    private Flux<DataBuffer> readFile(String filePath, long position) {
+        return DataBufferUtils
+            .read(new FileSystemResource(Paths.get(properties.getStorageBasePath(), filePath)),
+                  position,
+                  bufferFactory,
+                  properties.getReadBufferSize());
+    }
+
+    private Flux<DataBuffer> readFile(FileEntity file, long position) {
+        if (Objects.equals(file.getServerNodeId(), properties.getServerNodeId())) {
+            return readFile(file.getStoragePath(), position);
+        }
+        return readFromAnotherServer(file, position);
+    }
+
+    protected Flux<DataBuffer> readFromAnotherServer(FileEntity file, long position) {
+        return client
+            .get()
+            .uri("http://" + file.getServerNodeId() + "/file/{serverNodeId}/{fileId}", file.getServerNodeId(), file.getId())
+            .attribute(FileProperties.serverNodeIdAttr, file.getServerNodeId())
+            .headers(header -> header.setRange(Collections.singletonList(HttpRange.createByteRange(position))))
+            .retrieve()
+            .bodyToFlux(DataBuffer.class);
+    }
+
+    @Override
+    public Flux<DataBuffer> read(String id) {
+        return read(id, 0);
+    }
+
+    @Override
+    public Flux<DataBuffer> read(String id, long position) {
+        return repository
+            .findById(id)
+            .flatMapMany(file -> readFile(file, position));
+    }
+
+    @Override
+    public Flux<DataBuffer> read(String id, Function<ReaderContext, Mono<Void>> beforeRead) {
+        return repository
+            .findById(id)
+            .flatMapMany(file -> {
+                DefaultReaderContext context = new DefaultReaderContext(file.toInfo(), 0);
+                return beforeRead
+                    .apply(context)
+                    .thenMany(Flux.defer(() -> readFile(file, context.position)));
+            });
+    }
+
+    @AllArgsConstructor
+    private static class DefaultReaderContext implements ReaderContext {
+        private final FileInfo info;
+        private long position;
+
+        @Override
+        public FileInfo info() {
+            return info;
+        }
+
+        @Override
+        public void position(long position) {
+            this.position = position;
+        }
+    }
+
+}

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileEntity.java

@@ -0,0 +1,73 @@
+package org.jetlinks.community.io.file;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.hswebframework.ezorm.rdb.mapping.annotation.ColumnType;
+import org.hswebframework.ezorm.rdb.mapping.annotation.DefaultValue;
+import org.hswebframework.ezorm.rdb.mapping.annotation.EnumCodec;
+import org.hswebframework.ezorm.rdb.mapping.annotation.JsonCodec;
+import org.hswebframework.web.api.crud.entity.GenericEntity;
+import org.hswebframework.web.api.crud.entity.RecordCreationEntity;
+import org.hswebframework.web.crud.generator.Generators;
+
+import javax.persistence.Column;
+import javax.persistence.Table;
+import java.sql.JDBCType;
+import java.util.Map;
+
+@Getter
+@Setter
+@Table(name = "s_file")
+public class FileEntity extends GenericEntity<String> implements RecordCreationEntity {
+
+    @Column(nullable = false)
+    private String name;
+
+    @Column(nullable = false)
+    private String extension;
+
+    @Column(nullable = false)
+    private Long length;
+
+    @Column(nullable = false, length = 32)
+    private String md5;
+
+    @Column(nullable = false, length = 64)
+    private String sha256;
+
+    @Column(nullable = false)
+    @DefaultValue(generator = Generators.CURRENT_TIME)
+    private Long createTime;
+
+    @Column(length = 64)
+    private String creatorId;
+
+    @Column(length = 64, nullable = false)
+    private String serverNodeId;
+
+    @Column(length = 512, nullable = false)
+    private String storagePath;
+
+    @Column
+    @EnumCodec(toMask = true)
+    @ColumnType(jdbcType = JDBCType.BIGINT, javaType = Long.class)
+    private FileOption[] options;
+
+    @Column
+    @JsonCodec
+    @ColumnType(jdbcType = JDBCType.LONGVARCHAR, javaType = String.class)
+    private Map<String, Object> others;
+
+
+    public FileInfo toInfo() {
+        return copyTo(new FileInfo());
+    }
+
+    public static FileEntity of(FileInfo fileInfo,String storagePath,String serverNodeId) {
+        FileEntity fileEntity = new FileEntity().copyFrom(fileInfo);
+        fileEntity.setStoragePath(storagePath);
+        fileEntity.setServerNodeId(serverNodeId);
+        return fileEntity;
+    }
+
+}

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileInfo.java

@@ -0,0 +1,55 @@
+package org.jetlinks.community.io.file;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.commons.io.FilenameUtils;
+import org.springframework.http.MediaType;
+import org.springframework.util.StringUtils;
+
+@Getter
+@Setter
+public class FileInfo {
+
+    private String id;
+
+    private String name;
+
+    private String extension;
+
+    private long length;
+
+    private String md5;
+
+    private String sha256;
+
+    private long createTime;
+
+    private String creatorId;
+
+    private FileOption[] options;
+
+    public MediaType mediaType() {
+        if (!StringUtils.hasText(extension)) {
+            return MediaType.APPLICATION_OCTET_STREAM;
+        }
+        switch (extension.toLowerCase()) {
+            case "jpg":
+            case "jpeg":
+                return MediaType.IMAGE_JPEG;
+            case "text":
+            case "txt":
+                return MediaType.TEXT_PLAIN;
+            case "js":
+                return MediaType.APPLICATION_JSON;
+            default:
+                return MediaType.APPLICATION_OCTET_STREAM;
+        }
+    }
+
+    public FileInfo withFileName(String fileName) {
+        name = fileName;
+        extension = FilenameUtils.getExtension(fileName);
+        return this;
+    }
+
+}

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileManager.java

@@ -0,0 +1,31 @@
+package org.jetlinks.community.io.file;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.http.codec.multipart.FilePart;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+
+public interface FileManager {
+
+    Mono<FileInfo> saveFile(FilePart filePart);
+
+    Mono<FileInfo> saveFile(String name, Flux<DataBuffer> stream);
+
+    Mono<FileInfo> getFile(String id);
+
+    Flux<DataBuffer> read(String id);
+
+    Flux<DataBuffer> read(String id, long position);
+
+    Flux<DataBuffer> read(String id,
+                          Function<ReaderContext,Mono<Void>> beforeRead);
+
+    interface ReaderContext{
+        FileInfo info();
+
+        void position(long position);
+    }
+}

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileManagerConfiguration.java

@@ -0,0 +1,23 @@
+package org.jetlinks.community.io.file;
+
+import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
+import org.hswebframework.web.crud.annotation.EnableEasyormRepository;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@Configuration
+@EnableConfigurationProperties(FileProperties.class)
+@EnableEasyormRepository("org.jetlinks.community.io.file.FileEntity")
+public class FileManagerConfiguration {
+
+
+    @Bean
+    public FileManager fileManager(WebClient.Builder builder,
+                                   FileProperties properties,
+                                   ReactiveRepository<FileEntity, String> repository){
+        return new DefaultFileManager(builder,properties,repository);
+    }
+
+}

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileOption.java

@@ -0,0 +1,6 @@
+package org.jetlinks.community.io.file;
+
+public enum FileOption {
+
+    publicAccess
+}

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileProperties.java

@@ -0,0 +1,77 @@
+package org.jetlinks.community.io.file;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.hswebframework.web.exception.NotFoundException;
+import org.hswebframework.web.utils.DigestUtils;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.util.unit.DataSize;
+import org.springframework.web.reactive.function.client.ClientRequest;
+import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
+import org.springframework.web.util.UriComponentsBuilder;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+
+@Getter
+@Setter
+@ConfigurationProperties("file.manager")
+public class FileProperties {
+
+    public static final String clusterKeyHeader = "cluster-key";
+
+    public static final String serverNodeIdAttr = "server-node-id";
+
+    private String clusterKey = DigestUtils.md5Hex("_JetLinks_FM_K");
+
+    private String storageBasePath = "./data/files";
+
+    private int readBufferSize = (int) DataSize.ofKilobytes(64).toBytes();
+
+    private String serverNodeId = "default";
+
+    /**
+     * server1: 192.168.33.222:3322
+     */
+    private Map<String, String> clusterRute = new HashMap<>();
+
+
+    public String selectServerNode() {
+        int size = clusterRute.size();
+        if (size == 0) {
+            throw new NotFoundException("error.server_node_notfound");
+        }
+        return new ArrayList<>(clusterRute.keySet())
+            .get(ThreadLocalRandom.current().nextInt(size));
+    }
+
+    public ExchangeFilterFunction createWebClientRute() {
+        return (clientRequest, exchangeFunction) -> {
+            String target = clientRequest
+                .attribute(serverNodeIdAttr)
+                .map(String::valueOf)
+                .map(clusterRute::get)
+                .orElseThrow(() -> new NotFoundException("error.server_node_notfound"));
+            int idx = target.lastIndexOf(":");
+            String host = target.substring(0, idx).trim();
+            String port = target.substring(idx + 1).trim();
+            return exchangeFunction
+                .exchange(
+                    ClientRequest
+                        .from(clientRequest)
+                        .header(clusterKeyHeader, clusterKey)
+                        .url(UriComponentsBuilder
+                                 .fromUri(clientRequest.url())
+                                 .host(host)
+                                 .port(port)
+                                 .build()
+                                 .toUri())
+                        .build()
+                );
+        };
+    }
+
+
+}

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/web/FileManagerController.java

@@ -0,0 +1,110 @@
+package org.jetlinks.community.io.file.web;
+
+import io.swagger.v3.oas.annotations.Operation;
+import lombok.AllArgsConstructor;
+import org.hswebframework.web.authorization.annotation.Authorize;
+import org.hswebframework.web.authorization.exception.AccessDenyException;
+import org.jetlinks.community.io.file.FileInfo;
+import org.jetlinks.community.io.file.FileManager;
+import org.jetlinks.community.io.file.FileProperties;
+import org.springframework.http.ContentDisposition;
+import org.springframework.http.HttpRange;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.codec.multipart.FilePart;
+import org.springframework.web.bind.annotation.*;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Mono;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Objects;
+
+@RestController
+@RequestMapping("/file")
+@AllArgsConstructor
+public class FileManagerController {
+
+    private final FileProperties properties;
+
+    private final FileManager fileManager;
+
+    @PostMapping("/upload")
+    @Authorize(merge = false)
+    @Operation(summary = "上传文件")
+    public Mono<FileInfo> upload(@RequestPart("file") Mono<FilePart> partMono) {
+        return partMono.flatMap(fileManager::saveFile);
+    }
+
+    @GetMapping("/{fileId}")
+    @Authorize(merge = false)
+    @Operation(summary = "获取文件")
+    public Mono<Void> read(@PathVariable String fileId,
+                           ServerWebExchange exchange) {
+
+        return exchange
+            .getResponse()
+            .writeWith(fileManager
+                           .read(fileId, ctx -> {
+                               List<HttpRange> ranges = exchange
+                                   .getRequest()
+                                   .getHeaders()
+                                   .getRange();
+                               long position = 0;
+                               if (ranges.size() != 0) {
+                                   position = ranges.get(0).getRangeStart(ctx.info().getLength());
+                               }
+                               ctx.position(position);
+                               MediaType mediaType = ctx.info().mediaType();
+                               exchange.getResponse().getHeaders().setContentType(mediaType);
+                               exchange.getResponse().getHeaders().setContentLength(ctx.info().getLength());
+                               //文件流时下载文件
+                               if (mediaType.includes(MediaType.APPLICATION_OCTET_STREAM)) {
+                                   exchange.getResponse().getHeaders().setContentDisposition(
+                                       ContentDisposition
+                                           .builder("attachment")
+                                           .filename(ctx.info().getName(), StandardCharsets.UTF_8)
+                                           .build()
+                                   );
+                               }
+                               return Mono.empty();
+                           }));
+    }
+
+    //用于集群间获取文件
+    @GetMapping("/{clusterNodeId}/{fileId}")
+    @Authorize(ignore = true)
+    @Operation(summary = "集群间获取文件", hidden = true)
+    public Mono<Void> readFromCluster(@PathVariable String clusterNodeId,
+                                      @PathVariable String fileId,
+                                      ServerWebExchange exchange) {
+        if (Objects.equals(clusterNodeId, properties.getServerNodeId())) {
+            //读取自己
+            return Mono.error(new IllegalArgumentException("error.file_read_loop"));
+        }
+        //校验key
+        if (!Objects.equals(exchange.getRequest().getHeaders().getFirst(FileProperties.clusterKeyHeader),
+                            properties.getClusterKey())) {
+            return Mono.error(new AccessDenyException());
+        }
+        return read(fileId, exchange);
+    }
+
+    //用于集群间保存文件
+    @PostMapping("/{clusterNodeId}")
+    @Authorize(ignore = true)
+    @Operation(summary = "集群间获取文件", hidden = true)
+    public Mono<ResponseEntity<FileInfo>> saveFromCluster(@PathVariable String clusterNodeId,
+                                                          @RequestPart("file") Mono<FilePart> partMono,
+                                                          @RequestHeader(FileProperties.clusterKeyHeader) String key) {
+        if (!Objects.equals(clusterNodeId, properties.getServerNodeId())) {
+            return Mono.error(new IllegalArgumentException("error.file_read_loop"));
+        }
+        //校验key
+        if (!Objects.equals(key, properties.getClusterKey())) {
+            return Mono.error(new AccessDenyException());
+        }
+        return upload(partMono)
+            .map(ResponseEntity::ok);
+    }
+}

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/protocol/AutoDownloadJarProtocolSupportLoader.java

@@ -1,11 +1,13 @@
 package org.jetlinks.community.standalone.configuration.protocol;
 
+import lombok.Generated;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.hswebframework.web.bean.FastBeanCopier;
-import org.jetlinks.community.utils.TimeUtils;
 import org.jetlinks.core.ProtocolSupport;
 import org.jetlinks.core.spi.ServiceContext;
+import org.jetlinks.community.io.file.FileManager;
+import org.jetlinks.community.utils.TimeUtils;
 import org.jetlinks.supports.protocol.management.ProtocolSupportDefinition;
 import org.jetlinks.supports.protocol.management.jar.JarProtocolSupportLoader;
 import org.jetlinks.supports.protocol.management.jar.ProtocolClassLoader;
@@ -13,45 +15,63 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
 import org.springframework.web.reactive.function.client.WebClient;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
 import javax.annotation.PreDestroy;
 import java.io.File;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.TimeoutException;
 
-import static java.nio.file.StandardOpenOption.CREATE;
-import static java.nio.file.StandardOpenOption.WRITE;
-
+import static java.nio.file.StandardOpenOption.*;
+
+/**
+ * 自动下载并缓存协议包，
+ * <pre>
+ *     1. 下载的协议包报错在./data/protocols目录下，可通过启动参数-Djetlinks.protocol.temp.path进行配置
+ *     2. 文件名规则: 协议ID+"_"+md5(文件地址)
+ *     3. 如果文件不存在则下载协议
+ * </pre>
+ *
+ * @author zhouhao
+ * @since 1.3
+ */
 @Component
 @Slf4j
 public class AutoDownloadJarProtocolSupportLoader extends JarProtocolSupportLoader {
 
-
     final WebClient webClient;
 
     final File tempPath;
 
-    private final Duration loadTimeout = TimeUtils.parse(System.getProperty("jetlinks.protocol.load.timeout", "10s"));
+    private final Duration loadTimeout = TimeUtils.parse(System.getProperty("jetlinks.protocol.load.timeout", "30s"));
+
+    private final FileManager fileManager;
 
-    public AutoDownloadJarProtocolSupportLoader(WebClient.Builder builder) {
+    public AutoDownloadJarProtocolSupportLoader(WebClient.Builder builder,
+                                                FileManager fileManager) {
         this.webClient = builder.build();
-        tempPath = new File(System.getProperty("jetlinks.protocol.temp.path","./data/protocols"));
+        this.fileManager = fileManager;
+        tempPath = new File(System.getProperty("jetlinks.protocol.temp.path", "./data/protocols"));
         tempPath.mkdirs();
     }
 
     @Override
     @Autowired
+    @Generated
     public void setServiceContext(ServiceContext serviceContext) {
         super.setServiceContext(serviceContext);
     }
 
     @Override
     @PreDestroy
+    @Generated
     protected void closeAll() {
         super.closeAll();
     }
@@ -59,33 +79,31 @@ public class AutoDownloadJarProtocolSupportLoader extends JarProtocolSupportLoad
     @Override
     protected void closeLoader(ProtocolClassLoader loader) {
         super.closeLoader(loader);
-//        for (URL url : loader.getUrls()) {
-//            if (new File(url.getFile()).delete()) {
-//                log.debug("delete old protocol:{}", url);
-//            }
-//        }
     }
 
     @Override
     public Mono<? extends ProtocolSupport> load(ProtocolSupportDefinition definition) {
 
+        //复制新的配置信息
         ProtocolSupportDefinition newDef = FastBeanCopier.copy(definition, new ProtocolSupportDefinition());
 
         Map<String, Object> config = newDef.getConfiguration();
         String location = Optional
             .ofNullable(config.get("location"))
             .map(String::valueOf)
-            .orElseThrow(() -> new IllegalArgumentException("configuration.location不能为空"));
-
-        if (location.startsWith("http")) {
+            .orElse(null);
+        //远程文件则先下载再加载
+        if (StringUtils.hasText(location) && location.startsWith("http")) {
             String urlMd5 = DigestUtils.md5Hex(location);
             //地址没变则直接加载本地文件
             File file = new File(tempPath, (newDef.getId() + "_" + urlMd5) + ".jar");
             if (file.exists()) {
+                //设置文件地址文本地文件
                 config.put("location", file.getAbsolutePath());
                 return super
                     .load(newDef)
-                    .subscribeOn(Schedulers.elastic())
+                    .subscribeOn(Schedulers.boundedElastic())
+                    //加载失败则删除文件,防止文件内容错误时,一直无法加载
                     .doOnError(err -> file.delete());
             }
             return webClient
@@ -95,17 +113,51 @@ public class AutoDownloadJarProtocolSupportLoader extends JarProtocolSupportLoad
                 .bodyToFlux(DataBuffer.class)
                 .as(dataStream -> {
                     log.debug("download protocol file {} to {}", location, file.getAbsolutePath());
+                    //写出文件
                     return DataBufferUtils
                         .write(dataStream, file.toPath(), CREATE, WRITE)
                         .thenReturn(file.getAbsolutePath());
                 })
-                .subscribeOn(Schedulers.elastic())
+                //使用弹性线程池来写出文件
+                .subscribeOn(Schedulers.boundedElastic())
+                //设置本地文件路径
                 .doOnNext(path -> config.put("location", path))
                 .then(super.load(newDef))
                 .timeout(loadTimeout, Mono.error(() -> new TimeoutException("获取协议文件失败:" + location)))
+                //失败时删除文件
                 .doOnError(err -> file.delete())
                 ;
         }
-        return super.load(newDef);
+
+        //使用文件管理器获取文件
+        String fileId = (String) config.getOrDefault("fileId", null);
+        if (!StringUtils.hasText(fileId)) {
+            return Mono.error(new IllegalArgumentException("location or fileId can not be empty"));
+        }
+        return loadFromFileManager(newDef.getId(), fileId)
+            .flatMap(file -> {
+                config.put("location", file.getAbsolutePath());
+                return super
+                    .load(newDef)
+                    .subscribeOn(Schedulers.boundedElastic())
+                    //加载失败则删除文件,防止文件内容错误时,一直无法加载
+                    .doOnError(err -> file.delete());
+            });
+
     }
+
+    private Mono<File> loadFromFileManager(String protocolId, String fileId) {
+        Path path = Paths.get(tempPath.getPath(), (protocolId + "_" + fileId) + ".jar");
+
+        File file = path.toFile();
+        if (file.exists()) {
+            return Mono.just(file);
+        }
+
+        return DataBufferUtils
+            .write(fileManager.read(fileId),
+                   path, CREATE_NEW, TRUNCATE_EXISTING, WRITE)
+            .thenReturn(file);
+    }
+
 }