suguo
/
jetlinks-community
mirror of https://gitee.com/jetlinks/jetlinks-community.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

增加文件管理

This commit is contained in:
zhouhao 2022-06-13 14:45:34 +08:00
parent aab9215293
commit 79e3db9266
10 changed files with 579 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
jetlinks-components/io-component/pom.xml
View File

@ -28,7 +28,7 @@
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>easyexcel</artifactId>
<version>3.1.0</version>
<version>3.1.1</version>
<exclusions>
<exclusion>
<groupId>org.apache.poi</groupId>
@ -42,10 +42,28 @@
<artifactId>spring-webflux</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-buffer</artifactId>
</dependency>
<dependency>
<groupId>org.hswebframework.web</groupId>
<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>
<dependency>
<groupId>org.jetlinks</groupId>
<artifactId>jetlinks-supports</artifactId>
<version>${jetlinks.version}</version>
</dependency>
</dependencies>
</project>

218
jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/ClusterFileManager.java Normal file
View File

@ -0,0 +1,218 @@
package org.jetlinks.community.io.file;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.scalecube.services.annotations.Service;
import io.scalecube.services.annotations.ServiceMethod;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.commons.codec.digest.DigestUtils;
import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
import org.hswebframework.web.exception.BusinessException;
import org.hswebframework.web.exception.NotFoundException;
import org.hswebframework.web.id.IDGenerator;
import org.jetlinks.core.rpc.RpcManager;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.buffer.*;
import org.springframework.http.codec.multipart.FilePart;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.io.File;
import java.nio.file.NoSuchFileException;
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.Objects;
import java.util.function.Function;
public class ClusterFileManager implements FileManager {
private final FileProperties properties;
private final NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT);
private final ReactiveRepository<FileEntity, String> repository;
private final RpcManager rpcManager;
public ClusterFileManager(RpcManager rpcManager,
FileProperties properties,
ReactiveRepository<FileEntity, String> repository) {
new File(properties.getStorageBasePath()).mkdirs();
this.properties = properties;
this.rpcManager = rpcManager;
this.repository = repository;
rpcManager.registerService(new ServiceImpl());
}
@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> 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 = rpcManager.currentServerId();
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) {
return doSaveFile(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,
(int) properties.getReadBufferSize().toBytes())
.onErrorMap(NoSuchFileException.class, e -> new NotFoundException());
}
private Flux<DataBuffer> readFile(FileEntity file, long position) {
if (Objects.equals(file.getServerNodeId(), rpcManager.currentServerId())) {
return readFile(file.getStoragePath(), position);
}
return readFromAnotherServer(file, position);
}
protected Flux<DataBuffer> readFromAnotherServer(FileEntity file, long position) {
return rpcManager
.getService(file.getServerNodeId(), Service.class)
.switchIfEmpty(Mono.error(NotFoundException::new))
.flatMapMany(service -> service.read(new ReadRequest(file.getId(), position)))
.<DataBuffer>map(bufferFactory::wrap)
.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
}
@Override
public Flux<DataBuffer> read(String id) {
return read(id, 0);
}
@Override
public Flux<DataBuffer> read(String id, long position) {
return repository
.findById(id)
.switchIfEmpty(Mono.error(NotFoundException::new))
.flatMapMany(file -> readFile(file, position));
}
@Override
public Flux<DataBuffer> read(String id, Function<ReaderContext, Mono<Void>> beforeRead) {
return repository
.findById(id)
.switchIfEmpty(Mono.error(NotFoundException::new))
.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;
}
}
@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
public static class ReadRequest {
private String id;
private long position;
}
@io.scalecube.services.annotations.Service
public interface Service {
@ServiceMethod
Flux<ByteBuf> read(ReadRequest request);
}
public class ServiceImpl implements Service {
@Override
public Flux<ByteBuf> read(ReadRequest request) {
return ClusterFileManager
.this
.read(request.id, request.position)
.map(buf -> {
if (buf instanceof NettyDataBuffer) {
return ((NettyDataBuffer) buf).getNativeBuffer();
}
return Unpooled.wrappedBuffer(buf.asByteBuffer());
});
}
}
}

73
jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileEntity.java Normal file
View File

@ -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;
}
}

55
jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileInfo.java Normal file
View File

@ -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;
}
}

31
jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileManager.java Normal file
View File

@ -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);
}
}

23
jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileManagerConfiguration.java Normal file
View File

@ -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.jetlinks.core.rpc.RpcManager;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@EnableConfigurationProperties(FileProperties.class)
@EnableEasyormRepository("org.jetlinks.community.io.file.FileEntity")
public class FileManagerConfiguration {
@Bean
public FileManager fileManager(RpcManager rpcManager,
FileProperties properties,
ReactiveRepository<FileEntity, String> repository){
return new ClusterFileManager(rpcManager,properties,repository);
}
}

6
jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileOption.java Normal file
View File

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

17
jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/FileProperties.java Normal file
View File

@ -0,0 +1,17 @@
package org.jetlinks.community.io.file;
import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.util.unit.DataSize;
@Getter
@Setter
@ConfigurationProperties("file.manager")
public class FileProperties {
private String storageBasePath = "./data/files";
private DataSize readBufferSize = DataSize.ofKilobytes(64);
}

67
jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/file/web/FileManagerController.java Normal file
View File

@ -0,0 +1,67 @@
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.jetlinks.community.io.file.FileInfo;
import org.jetlinks.community.io.file.FileManager;
import org.springframework.http.ContentDisposition;
import org.springframework.http.HttpRange;
import org.springframework.http.MediaType;
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;
@RestController
@RequestMapping("/file")
@AllArgsConstructor
public class FileManagerController {
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();
}));
}
}

88
jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/protocol/AutoDownloadJarProtocolSupportLoader.java Normal file → Executable file
View File

@ -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"));
public AutoDownloadJarProtocolSupportLoader(WebClient.Builder builder) {
private final FileManager fileManager;
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);
}
}