mirror of
https://github.com/halo-dev/plugin-s3.git
synced 2025-10-14 14:11:15 +00:00
Compare commits
12 Commits
Author | SHA1 | Date | |
---|---|---|---|
![]() |
c109bbd61f | ||
![]() |
2320800907 | ||
![]() |
00537c164c | ||
![]() |
8be39b9898 | ||
![]() |
022ecea94f | ||
![]() |
b3bdd02e08 | ||
![]() |
5a95b4ced1 | ||
![]() |
88490bb80f | ||
![]() |
5e9b9f803b | ||
![]() |
c635ebede8 | ||
![]() |
459cc1cf94 | ||
![]() |
780258ffc1 |
14
README.md
14
README.md
@@ -35,13 +35,19 @@
|
||||
|
||||
### Bucket 桶名称
|
||||
|
||||
与服务商的控制台中的桶名称一致。
|
||||
一般与服务商控制台中的空间名称一致。
|
||||
|
||||
> 注意部分服务商 s3 空间名 ≠ 空间名称,若出现“Access Denied”报错可检查 Bucket 是否正确。
|
||||
>
|
||||
> 可通过 S3Browser 查看桶列表,七牛云也可在“开发者平台-对象存储-空间概览-s3域名”中查看 s3 空间名。
|
||||
|
||||
### Region
|
||||
|
||||
一般留空即可。
|
||||
|
||||
> 若确认过其他配置正确又不能访问,请在服务商的文档中查看并填写英文的 Region,例如 `cn-east-1`。
|
||||
>
|
||||
> Cloudflare 需要填写均为小写字母的 `auto`。
|
||||
|
||||
## 部分对象存储服务商兼容性
|
||||
|
||||
@@ -55,10 +61,12 @@
|
||||
|金山云|https://docs.ksyun.com/documents/6761|Virtual Hosted Style|✅|
|
||||
|青云|https://docsv3.qingcloud.com/storage/object-storage/s3/intro/|Virtual Hosted Style / <br>Path Style|✅|
|
||||
|网易数帆|[https://sf.163.com/help/documents/89796157866430464](https://sf.163.com/help/documents/89796157866430464)|Virtual Hosted Style|✅|
|
||||
|Cloudflare|Cloudflare S3 兼容性API<br>[https://developers.cloudflare.com/r2/data-access/s3-api/](https://developers.cloudflare.com/r2/data-access/s3-api/)|Virtual Hosted Style / <br>Path Style|✅|
|
||||
| Oracle Cloud |[https://docs.oracle.com/en-us/iaas/Content/Object/Tasks/s3compatibleapi.htm](https://docs.oracle.com/en-us/iaas/Content/Object/Tasks/s3compatibleapi.htm)|Virtual Hosted Style / <br>Path Style|✅|
|
||||
|自建minio|\-|Path Style|✅|
|
||||
|华为云|文档未说明是否兼容,工单反馈不保证兼容性,实际测试可以使用|Virtual Hosted Style|❓|
|
||||
|Ucloud|只支持 8MB 大小的分片,本插件暂不支持<br>[https://docs.ucloud.cn/ufile/s3/s3\_introduction](https://docs.ucloud.cn/ufile/s3/s3_introduction)|\-|❎|
|
||||
|又拍云|暂不支持 s3 协议|\-|❎|
|
||||
|Ucloud|只支持 8MB 大小的分片,本插件暂不支持<br>[https://docs.ucloud.cn/ufile/s3/s3\_introduction](https://docs.ucloud.cn/ufile/s3/s3_introduction)|\-|❌|
|
||||
|又拍云|暂不支持 s3 协议|\-|❌|
|
||||
|
||||
## 开发环境
|
||||
|
||||
|
19
build.gradle
19
build.gradle
@@ -1,5 +1,6 @@
|
||||
plugins {
|
||||
id "io.github.guqing.plugin-development" version "0.0.6-SNAPSHOT"
|
||||
id "io.github.guqing.plugin-development" version "0.0.7-SNAPSHOT"
|
||||
id "io.freefair.lombok" version "8.0.0-rc2"
|
||||
id 'java'
|
||||
}
|
||||
|
||||
@@ -8,7 +9,7 @@ sourceCompatibility = JavaVersion.VERSION_17
|
||||
|
||||
repositories {
|
||||
maven { url 'https://s01.oss.sonatype.org/content/repositories/releases' }
|
||||
maven { url 'https://repo.spring.io/milestone' }
|
||||
maven { url 'https://s01.oss.sonatype.org/content/repositories/snapshots/' }
|
||||
mavenCentral()
|
||||
}
|
||||
|
||||
@@ -23,9 +24,8 @@ jar {
|
||||
}
|
||||
|
||||
dependencies {
|
||||
compileOnly platform("run.halo.dependencies:halo-dependencies:1.0.0")
|
||||
|
||||
compileOnly files("lib/halo-2.0.0-SNAPSHOT-plain.jar")
|
||||
implementation platform('run.halo.tools.platform:plugin:2.5.0-SNAPSHOT')
|
||||
compileOnly 'run.halo.app:api'
|
||||
|
||||
implementation platform('software.amazon.awssdk:bom:2.19.8')
|
||||
implementation 'software.amazon.awssdk:s3'
|
||||
@@ -33,14 +33,9 @@ dependencies {
|
||||
implementation "javax.activation:activation:1.1.1"
|
||||
implementation "org.glassfish.jaxb:jaxb-runtime:2.3.3"
|
||||
|
||||
compileOnly 'org.projectlombok:lombok'
|
||||
annotationProcessor 'org.projectlombok:lombok:1.18.22'
|
||||
|
||||
testImplementation platform("run.halo.dependencies:halo-dependencies:1.0.0")
|
||||
testImplementation files("lib/halo-2.0.0-SNAPSHOT-plain.jar")
|
||||
testImplementation 'run.halo.app:api'
|
||||
testImplementation 'org.springframework.boot:spring-boot-starter-test'
|
||||
testImplementation 'org.junit.jupiter:junit-jupiter-api:5.9.0'
|
||||
testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.9.0'
|
||||
testImplementation 'io.projectreactor:reactor-test'
|
||||
}
|
||||
|
||||
test {
|
||||
|
@@ -1 +1 @@
|
||||
version=1.3.0-SNAPSHOT
|
||||
version=1.4.1-SNAPSHOT
|
||||
|
2
gradle/wrapper/gradle-wrapper.properties
vendored
2
gradle/wrapper/gradle-wrapper.properties
vendored
@@ -1,5 +1,5 @@
|
||||
distributionBase=GRADLE_USER_HOME
|
||||
distributionPath=wrapper/dists
|
||||
distributionUrl=https\://services.gradle.org/distributions/gradle-7.4-bin.zip
|
||||
distributionUrl=https\://services.gradle.org/distributions/gradle-8.0.2-bin.zip
|
||||
zipStoreBase=GRADLE_USER_HOME
|
||||
zipStorePath=wrapper/dists
|
||||
|
Binary file not shown.
44
src/main/java/run/halo/s3os/FileNameUtils.java
Normal file
44
src/main/java/run/halo/s3os/FileNameUtils.java
Normal file
@@ -0,0 +1,44 @@
|
||||
package run.halo.s3os;
|
||||
|
||||
import com.google.common.io.Files;
|
||||
import org.apache.commons.lang3.RandomStringUtils;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
|
||||
public final class FileNameUtils {
|
||||
|
||||
private FileNameUtils() {
|
||||
}
|
||||
|
||||
public static String removeFileExtension(String filename, boolean removeAllExtensions) {
|
||||
if (filename == null || filename.isEmpty()) {
|
||||
return filename;
|
||||
}
|
||||
var extPattern = "(?<!^)[.]" + (removeAllExtensions ? ".*" : "[^.]*$");
|
||||
return filename.replaceAll(extPattern, "");
|
||||
}
|
||||
|
||||
/**
|
||||
* Append random string after file name.
|
||||
* <pre>
|
||||
* Case 1: halo.run -> halo-xyz.run
|
||||
* Case 2: .run -> xyz.run
|
||||
* Case 3: halo -> halo-xyz
|
||||
* </pre>
|
||||
*
|
||||
* @param filename is name of file.
|
||||
* @param length is for generating random string with specific length.
|
||||
* @return File name with random string.
|
||||
*/
|
||||
public static String randomFileName(String filename, int length) {
|
||||
var nameWithoutExt = Files.getNameWithoutExtension(filename);
|
||||
var ext = Files.getFileExtension(filename);
|
||||
var random = RandomStringUtils.randomAlphabetic(length).toLowerCase();
|
||||
if (StringUtils.isBlank(nameWithoutExt)) {
|
||||
return random + "." + ext;
|
||||
}
|
||||
if (StringUtils.isBlank(ext)) {
|
||||
return nameWithoutExt + "-" + random;
|
||||
}
|
||||
return nameWithoutExt + "-" + random + "." + ext;
|
||||
}
|
||||
}
|
@@ -1,16 +1,35 @@
|
||||
package run.halo.s3os;
|
||||
|
||||
import lombok.Data;
|
||||
import java.net.URI;
|
||||
import java.net.URISyntaxException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.FileAlreadyExistsException;
|
||||
import java.time.Duration;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.pf4j.Extension;
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.springframework.core.io.buffer.DataBuffer;
|
||||
import org.springframework.core.io.buffer.DataBufferUtils;
|
||||
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.http.MediaTypeFactory;
|
||||
import org.springframework.lang.Nullable;
|
||||
import org.springframework.web.server.ServerErrorException;
|
||||
import org.springframework.web.server.ServerWebInputException;
|
||||
import org.springframework.web.util.UriUtils;
|
||||
import reactor.core.Exceptions;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.scheduler.Schedulers;
|
||||
import reactor.util.context.Context;
|
||||
import reactor.util.retry.Retry;
|
||||
import run.halo.app.core.extension.attachment.Attachment;
|
||||
import run.halo.app.core.extension.attachment.Attachment.AttachmentSpec;
|
||||
import run.halo.app.core.extension.attachment.Constant;
|
||||
@@ -20,22 +39,26 @@ import run.halo.app.extension.ConfigMap;
|
||||
import run.halo.app.extension.Metadata;
|
||||
import run.halo.app.infra.utils.JsonUtils;
|
||||
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
|
||||
import software.amazon.awssdk.awscore.presigner.SdkPresigner;
|
||||
import software.amazon.awssdk.core.SdkResponse;
|
||||
import software.amazon.awssdk.core.async.AsyncRequestBody;
|
||||
import software.amazon.awssdk.core.sync.RequestBody;
|
||||
import software.amazon.awssdk.http.SdkHttpResponse;
|
||||
import software.amazon.awssdk.regions.Region;
|
||||
import software.amazon.awssdk.services.s3.S3AsyncClient;
|
||||
import software.amazon.awssdk.services.s3.S3Client;
|
||||
import software.amazon.awssdk.services.s3.S3Configuration;
|
||||
import software.amazon.awssdk.services.s3.model.*;
|
||||
|
||||
import java.net.URI;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
|
||||
import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
|
||||
import software.amazon.awssdk.services.s3.model.CompletedPart;
|
||||
import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
|
||||
import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
|
||||
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
|
||||
import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
|
||||
import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
|
||||
import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
|
||||
import software.amazon.awssdk.services.s3.model.UploadPartRequest;
|
||||
import software.amazon.awssdk.services.s3.presigner.S3Presigner;
|
||||
import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest;
|
||||
import software.amazon.awssdk.utils.SdkAutoCloseable;
|
||||
|
||||
@Slf4j
|
||||
@Extension
|
||||
@@ -48,38 +71,98 @@ public class S3OsAttachmentHandler implements AttachmentHandler {
|
||||
@Override
|
||||
public Mono<Attachment> upload(UploadContext uploadContext) {
|
||||
return Mono.just(uploadContext).filter(context -> this.shouldHandle(context.policy()))
|
||||
.flatMap(context -> {
|
||||
final var properties = getProperties(context.configMap());
|
||||
return upload(context, properties).map(
|
||||
objectDetail -> this.buildAttachment(context, properties, objectDetail));
|
||||
});
|
||||
.flatMap(context -> {
|
||||
final var properties = getProperties(context.configMap());
|
||||
return upload(context, properties)
|
||||
.subscribeOn(Schedulers.boundedElastic())
|
||||
.map(objectDetail -> this.buildAttachment(properties, objectDetail));
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<Attachment> delete(DeleteContext deleteContext) {
|
||||
return Mono.just(deleteContext).filter(context -> this.shouldHandle(context.policy()))
|
||||
.flatMap(context -> {
|
||||
var annotations = context.attachment().getMetadata().getAnnotations();
|
||||
if (annotations == null || !annotations.containsKey(OBJECT_KEY)) {
|
||||
return Mono.just(context);
|
||||
}
|
||||
var objectName = annotations.get(OBJECT_KEY);
|
||||
var properties = getProperties(deleteContext.configMap());
|
||||
var client = buildS3AsyncClient(properties);
|
||||
return Mono.fromFuture(client.deleteObject(DeleteObjectRequest.builder()
|
||||
.bucket(properties.getBucket())
|
||||
.key(objectName)
|
||||
.build()))
|
||||
.doFinally(signalType -> client.close())
|
||||
.map(response -> {
|
||||
checkResult(response, "delete object");
|
||||
log.info("Delete object {} from bucket {} successfully",
|
||||
objectName, properties.getBucket());
|
||||
return context;
|
||||
});
|
||||
.flatMap(context -> {
|
||||
var objectKey = getObjectKey(context.attachment());
|
||||
if (objectKey == null) {
|
||||
return Mono.just(context);
|
||||
}
|
||||
var properties = getProperties(deleteContext.configMap());
|
||||
return Mono.using(() -> buildS3Client(properties),
|
||||
client -> Mono.fromCallable(
|
||||
() -> client.deleteObject(DeleteObjectRequest.builder()
|
||||
.bucket(properties.getBucket())
|
||||
.key(objectKey)
|
||||
.build())).subscribeOn(Schedulers.boundedElastic()),
|
||||
S3Client::close)
|
||||
.doOnNext(response -> {
|
||||
checkResult(response, "delete object");
|
||||
log.info("Delete object {} from bucket {} successfully",
|
||||
objectKey, properties.getBucket());
|
||||
})
|
||||
.thenReturn(context);
|
||||
})
|
||||
.map(DeleteContext::attachment);
|
||||
}
|
||||
|
||||
})
|
||||
.map(DeleteContext::attachment);
|
||||
@Override
|
||||
public Mono<URI> getSharedURL(Attachment attachment, Policy policy, ConfigMap configMap,
|
||||
Duration ttl) {
|
||||
if (!this.shouldHandle(policy)) {
|
||||
return Mono.empty();
|
||||
}
|
||||
var objectKey = getObjectKey(attachment);
|
||||
if (objectKey == null) {
|
||||
return Mono.error(new IllegalArgumentException(
|
||||
"Cannot obtain object key from attachment " + attachment.getMetadata().getName()));
|
||||
}
|
||||
var properties = getProperties(configMap);
|
||||
|
||||
return Mono.using(() -> buildS3Presigner(properties),
|
||||
s3Presigner -> {
|
||||
var getObjectRequest = GetObjectRequest.builder()
|
||||
.bucket(properties.getBucket())
|
||||
.key(objectKey)
|
||||
.build();
|
||||
var presignedRequest = GetObjectPresignRequest.builder()
|
||||
.signatureDuration(ttl)
|
||||
.getObjectRequest(getObjectRequest)
|
||||
.build();
|
||||
var presignedGetObjectRequest = s3Presigner.presignGetObject(presignedRequest);
|
||||
var presignedURL = presignedGetObjectRequest.url();
|
||||
try {
|
||||
return Mono.just(presignedURL.toURI());
|
||||
} catch (URISyntaxException e) {
|
||||
return Mono.error(
|
||||
new RuntimeException("Failed to convert URL " + presignedURL + " to URI."));
|
||||
}
|
||||
},
|
||||
SdkPresigner::close)
|
||||
.subscribeOn(Schedulers.boundedElastic());
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<URI> getPermalink(Attachment attachment, Policy policy, ConfigMap configMap) {
|
||||
if (!this.shouldHandle(policy)) {
|
||||
return Mono.empty();
|
||||
}
|
||||
var objectKey = getObjectKey(attachment);
|
||||
if (objectKey == null) {
|
||||
// fallback to default handler for backward compatibility
|
||||
return Mono.empty();
|
||||
}
|
||||
var properties = getProperties(configMap);
|
||||
var objectURL = getObjectURL(properties, objectKey);
|
||||
return Mono.just(URI.create(objectURL));
|
||||
}
|
||||
|
||||
@Nullable
|
||||
private String getObjectKey(Attachment attachment) {
|
||||
var annotations = attachment.getMetadata().getAnnotations();
|
||||
if (annotations == null) {
|
||||
return null;
|
||||
}
|
||||
return annotations.get(OBJECT_KEY);
|
||||
}
|
||||
|
||||
S3OsProperties getProperties(ConfigMap configMap) {
|
||||
@@ -87,164 +170,253 @@ public class S3OsAttachmentHandler implements AttachmentHandler {
|
||||
return JsonUtils.jsonToObject(settingJson, S3OsProperties.class);
|
||||
}
|
||||
|
||||
Attachment buildAttachment(UploadContext uploadContext, S3OsProperties properties,
|
||||
ObjectDetail objectDetail) {
|
||||
String externalLink;
|
||||
if (StringUtils.isBlank(properties.getDomain())) {
|
||||
var host = properties.getBucket() + "." + properties.getEndpoint();
|
||||
externalLink = properties.getProtocol() + "://" + host + "/" + objectDetail.objectKey();
|
||||
} else {
|
||||
externalLink = properties.getProtocol() + "://" + properties.getDomain() + "/" + objectDetail.objectKey();
|
||||
}
|
||||
Attachment buildAttachment(S3OsProperties properties, ObjectDetail objectDetail) {
|
||||
String externalLink = getObjectURL(properties, objectDetail.uploadState.objectKey);
|
||||
|
||||
var metadata = new Metadata();
|
||||
metadata.setName(UUID.randomUUID().toString());
|
||||
metadata.setAnnotations(
|
||||
Map.of(OBJECT_KEY, objectDetail.objectKey(), Constant.EXTERNAL_LINK_ANNO_KEY,
|
||||
UriUtils.encodePath(externalLink, StandardCharsets.UTF_8)));
|
||||
metadata.setAnnotations(new HashMap<>(
|
||||
Map.of(OBJECT_KEY, objectDetail.uploadState.objectKey,
|
||||
Constant.EXTERNAL_LINK_ANNO_KEY, externalLink)));
|
||||
|
||||
var objectMetadata = objectDetail.objectMetadata();
|
||||
var spec = new AttachmentSpec();
|
||||
spec.setSize(objectMetadata.contentLength());
|
||||
spec.setDisplayName(uploadContext.file().filename());
|
||||
spec.setDisplayName(objectDetail.uploadState.fileName);
|
||||
spec.setMediaType(objectMetadata.contentType());
|
||||
|
||||
var attachment = new Attachment();
|
||||
attachment.setMetadata(metadata);
|
||||
attachment.setSpec(spec);
|
||||
log.info("Upload object {} to bucket {} successfully", objectDetail.objectKey(), properties.getBucket());
|
||||
log.info("Upload object {} to bucket {} successfully", objectDetail.uploadState.objectKey,
|
||||
properties.getBucket());
|
||||
return attachment;
|
||||
}
|
||||
|
||||
S3AsyncClient buildS3AsyncClient(S3OsProperties properties) {
|
||||
return S3AsyncClient.builder()
|
||||
.region(Region.of(properties.getRegion()))
|
||||
.endpointOverride(URI.create(properties.getEndpointProtocol() + "://" + properties.getEndpoint()))
|
||||
.credentialsProvider(() -> AwsBasicCredentials.create(properties.getAccessKey(),
|
||||
properties.getAccessSecret()))
|
||||
.serviceConfiguration(S3Configuration.builder()
|
||||
.chunkedEncodingEnabled(false)
|
||||
.pathStyleAccessEnabled(properties.getEnablePathStyleAccess())
|
||||
.build())
|
||||
.build();
|
||||
private String getObjectURL(S3OsProperties properties, String objectKey) {
|
||||
String objectURL;
|
||||
if (StringUtils.isBlank(properties.getDomain())) {
|
||||
String host;
|
||||
if (properties.getEnablePathStyleAccess()) {
|
||||
host = properties.getEndpoint() + "/" + properties.getBucket();
|
||||
} else {
|
||||
host = properties.getBucket() + "." + properties.getEndpoint();
|
||||
}
|
||||
objectURL = properties.getProtocol() + "://" + host + "/" + objectKey;
|
||||
} else {
|
||||
objectURL = properties.getProtocol() + "://" + properties.getDomain() + "/" + objectKey;
|
||||
}
|
||||
return UriUtils.encodePath(objectURL, StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
S3Client buildS3Client(S3OsProperties properties) {
|
||||
return S3Client.builder()
|
||||
.region(Region.of(properties.getRegion()))
|
||||
.endpointOverride(
|
||||
URI.create(properties.getEndpointProtocol() + "://" + properties.getEndpoint()))
|
||||
.credentialsProvider(() -> AwsBasicCredentials.create(properties.getAccessKey(),
|
||||
properties.getAccessSecret()))
|
||||
.serviceConfiguration(S3Configuration.builder()
|
||||
.chunkedEncodingEnabled(false)
|
||||
.pathStyleAccessEnabled(properties.getEnablePathStyleAccess())
|
||||
.build())
|
||||
.build();
|
||||
}
|
||||
|
||||
private S3Presigner buildS3Presigner(S3OsProperties properties) {
|
||||
return S3Presigner.builder()
|
||||
.region(Region.of(properties.getRegion()))
|
||||
.endpointOverride(
|
||||
URI.create(properties.getEndpointProtocol() + "://" + properties.getEndpoint()))
|
||||
.credentialsProvider(() -> AwsBasicCredentials.create(properties.getAccessKey(),
|
||||
properties.getAccessSecret()))
|
||||
.serviceConfiguration(S3Configuration.builder()
|
||||
.chunkedEncodingEnabled(false)
|
||||
.pathStyleAccessEnabled(properties.getEnablePathStyleAccess())
|
||||
.build())
|
||||
.build();
|
||||
}
|
||||
|
||||
Flux<DataBuffer> reshape(Publisher<DataBuffer> content, int bufferSize) {
|
||||
var dataBufferFactory = DefaultDataBufferFactory.sharedInstance;
|
||||
return Flux.<ByteBuffer>create(sink -> {
|
||||
var byteBuffer = ByteBuffer.allocate(bufferSize);
|
||||
Flux.from(content)
|
||||
.doOnNext(dataBuffer -> {
|
||||
var count = dataBuffer.readableByteCount();
|
||||
for (var i = 0; i < count; i++) {
|
||||
byteBuffer.put(dataBuffer.read());
|
||||
// Emit the buffer when buffer
|
||||
if (!byteBuffer.hasRemaining()) {
|
||||
sink.next(deepCopy(byteBuffer));
|
||||
byteBuffer.clear();
|
||||
}
|
||||
}
|
||||
})
|
||||
.doOnComplete(() -> {
|
||||
// Emit the last part of buffer.
|
||||
if (byteBuffer.position() > 0) {
|
||||
sink.next(deepCopy(byteBuffer));
|
||||
}
|
||||
})
|
||||
.subscribe(DataBufferUtils::release, sink::error, sink::complete,
|
||||
Context.of(sink.contextView()));
|
||||
})
|
||||
.map(dataBufferFactory::wrap)
|
||||
.cast(DataBuffer.class)
|
||||
.doOnDiscard(DataBuffer.class, DataBufferUtils::release);
|
||||
}
|
||||
|
||||
ByteBuffer deepCopy(ByteBuffer src) {
|
||||
src.flip();
|
||||
var dest = ByteBuffer.allocate(src.limit());
|
||||
dest.put(src);
|
||||
src.rewind();
|
||||
dest.flip();
|
||||
return dest;
|
||||
}
|
||||
|
||||
Mono<ObjectDetail> upload(UploadContext uploadContext, S3OsProperties properties) {
|
||||
var originFilename = uploadContext.file().filename();
|
||||
var objectKey = properties.getObjectName(originFilename);
|
||||
var contentType = MediaTypeFactory.getMediaType(originFilename)
|
||||
.orElse(MediaType.APPLICATION_OCTET_STREAM).toString();
|
||||
var uploadingMapKey = properties.getBucket() + "/" + objectKey;
|
||||
// deduplication of uploading files
|
||||
if (uploadingFile.put(uploadingMapKey, uploadingMapKey) != null) {
|
||||
return Mono.error(new ServerWebInputException("文件 " + originFilename + " 已存在,建议更名后重试。"));
|
||||
}
|
||||
return Mono.using(() -> buildS3Client(properties),
|
||||
client -> {
|
||||
var uploadState = new UploadState(properties, uploadContext.file().filename());
|
||||
|
||||
var s3client = buildS3AsyncClient(properties);
|
||||
var content = uploadContext.file().content();
|
||||
|
||||
var uploadState = new UploadState(properties.getBucket(), objectKey);
|
||||
|
||||
return Mono
|
||||
// check whether file exists
|
||||
.fromFuture(s3client.headObject(HeadObjectRequest.builder()
|
||||
.bucket(properties.getBucket())
|
||||
.key(objectKey)
|
||||
.build()))
|
||||
.onErrorResume(NoSuchKeyException.class, e -> {
|
||||
var builder = HeadObjectResponse.builder();
|
||||
builder.sdkHttpResponse(SdkHttpResponse.builder().statusCode(404).build());
|
||||
return Mono.just(builder.build());
|
||||
})
|
||||
.flatMap(response -> {
|
||||
if (response != null && response.sdkHttpResponse() != null && response.sdkHttpResponse().isSuccessful()) {
|
||||
return Mono.error(new ServerWebInputException("文件 " + originFilename + " 已存在,建议更名后重试。"));
|
||||
}else {
|
||||
return Mono.just(uploadState);
|
||||
}
|
||||
})
|
||||
// init multipart upload
|
||||
.flatMap(state -> Mono.fromFuture(s3client.createMultipartUpload(
|
||||
return checkFileExistsAndRename(uploadState, client)
|
||||
// init multipart upload
|
||||
.flatMap(state -> Mono.fromCallable(() -> client.createMultipartUpload(
|
||||
CreateMultipartUploadRequest.builder()
|
||||
.bucket(properties.getBucket())
|
||||
.contentType(contentType)
|
||||
.key(objectKey)
|
||||
.build())))
|
||||
.flatMapMany((response) -> {
|
||||
checkResult(response, "createMultipartUpload");
|
||||
uploadState.setUploadId(response.uploadId());
|
||||
return uploadContext.file().content();
|
||||
})
|
||||
// buffer to part
|
||||
.windowUntil((buffer) -> {
|
||||
uploadState.buffered += buffer.readableByteCount();
|
||||
if (uploadState.buffered >= MULTIPART_MIN_PART_SIZE) {
|
||||
uploadState.buffered = 0;
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
})
|
||||
// upload part
|
||||
.concatMap((window) -> window.collectList().flatMap((bufferList) -> {
|
||||
var buffer = S3OsAttachmentHandler.concatBuffers(bufferList);
|
||||
return uploadPart(uploadState, buffer, s3client);
|
||||
}))
|
||||
.reduce(uploadState, (state, completedPart) -> {
|
||||
state.completedParts.put(completedPart.partNumber(), completedPart);
|
||||
return state;
|
||||
})
|
||||
// complete multipart upload
|
||||
.flatMap((state) -> Mono
|
||||
.fromFuture(s3client.completeMultipartUpload(CompleteMultipartUploadRequest.builder()
|
||||
.bucket(state.bucket)
|
||||
.uploadId(state.uploadId)
|
||||
.multipartUpload(CompletedMultipartUpload.builder()
|
||||
.parts(state.completedParts.values())
|
||||
.build())
|
||||
.key(state.objectKey)
|
||||
.bucket(properties.getBucket())
|
||||
.contentType(state.contentType)
|
||||
.key(state.objectKey)
|
||||
.build())))
|
||||
.doOnNext((response) -> {
|
||||
checkResult(response, "createMultipartUpload");
|
||||
uploadState.uploadId = response.uploadId();
|
||||
})
|
||||
.thenMany(reshape(content, MULTIPART_MIN_PART_SIZE))
|
||||
// buffer to part
|
||||
.windowUntil((buffer) -> {
|
||||
uploadState.buffered += buffer.readableByteCount();
|
||||
if (uploadState.buffered >= MULTIPART_MIN_PART_SIZE) {
|
||||
uploadState.buffered = 0;
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
})
|
||||
// upload part
|
||||
.concatMap((window) -> window.collectList().flatMap((bufferList) -> {
|
||||
var buffer = S3OsAttachmentHandler.concatBuffers(bufferList);
|
||||
return uploadPart(uploadState, buffer, client);
|
||||
}))
|
||||
.reduce(uploadState, (state, completedPart) -> {
|
||||
state.completedParts.put(completedPart.partNumber(), completedPart);
|
||||
return state;
|
||||
})
|
||||
// complete multipart upload
|
||||
.flatMap((state) -> Mono.just(client.completeMultipartUpload(
|
||||
CompleteMultipartUploadRequest
|
||||
.builder()
|
||||
.bucket(properties.getBucket())
|
||||
.uploadId(state.uploadId)
|
||||
.multipartUpload(CompletedMultipartUpload.builder()
|
||||
.parts(state.completedParts.values())
|
||||
.build())
|
||||
))
|
||||
// get object metadata
|
||||
.flatMap((response) -> {
|
||||
checkResult(response, "completeUpload");
|
||||
return Mono.fromFuture(s3client.headObject(
|
||||
.key(state.objectKey)
|
||||
.build())
|
||||
))
|
||||
// get object metadata
|
||||
.flatMap((response) -> {
|
||||
checkResult(response, "completeUpload");
|
||||
return Mono.just(client.headObject(
|
||||
HeadObjectRequest.builder()
|
||||
.bucket(properties.getBucket())
|
||||
.key(objectKey)
|
||||
.build()
|
||||
));
|
||||
.bucket(properties.getBucket())
|
||||
.key(uploadState.objectKey)
|
||||
.build()
|
||||
));
|
||||
})
|
||||
// build object detail
|
||||
.map((response) -> {
|
||||
checkResult(response, "getMetadata");
|
||||
return new ObjectDetail(uploadState, response);
|
||||
})
|
||||
// close client
|
||||
.doFinally((signalType) -> {
|
||||
if (uploadState.needRemoveMapKey) {
|
||||
uploadingFile.remove(uploadState.getUploadingMapKey());
|
||||
}
|
||||
});
|
||||
},
|
||||
SdkAutoCloseable::close);
|
||||
}
|
||||
|
||||
private Mono<UploadState> checkFileExistsAndRename(UploadState uploadState,
|
||||
S3Client s3client) {
|
||||
return Mono.defer(() -> {
|
||||
// deduplication of uploading files
|
||||
if (uploadingFile.put(uploadState.getUploadingMapKey(),
|
||||
uploadState.getUploadingMapKey()) != null) {
|
||||
return Mono.error(new FileAlreadyExistsException("文件 " + uploadState.objectKey
|
||||
+
|
||||
" 已存在,建议更名后重试。[local]"));
|
||||
}
|
||||
uploadState.needRemoveMapKey = true;
|
||||
// check whether file exists
|
||||
return Mono.fromSupplier(() -> s3client.headObject(HeadObjectRequest.builder()
|
||||
.bucket(uploadState.properties.getBucket())
|
||||
.key(uploadState.objectKey)
|
||||
.build()))
|
||||
.onErrorResume(NoSuchKeyException.class, e -> {
|
||||
var builder = HeadObjectResponse.builder();
|
||||
builder.sdkHttpResponse(SdkHttpResponse.builder().statusCode(404).build());
|
||||
return Mono.just(builder.build());
|
||||
})
|
||||
.flatMap(response -> {
|
||||
if (response != null && response.sdkHttpResponse() != null
|
||||
&& response.sdkHttpResponse().isSuccessful()) {
|
||||
return Mono.error(
|
||||
new FileAlreadyExistsException("文件 " + uploadState.objectKey
|
||||
+ " 已存在,建议更名后重试。[remote]"));
|
||||
} else {
|
||||
return Mono.just(uploadState);
|
||||
}
|
||||
});
|
||||
})
|
||||
.retryWhen(Retry.max(3)
|
||||
.filter(FileAlreadyExistsException.class::isInstance)
|
||||
.doAfterRetry((retrySignal) -> {
|
||||
if (uploadState.needRemoveMapKey) {
|
||||
uploadingFile.remove(uploadState.getUploadingMapKey());
|
||||
uploadState.needRemoveMapKey = false;
|
||||
}
|
||||
uploadState.randomFileName();
|
||||
})
|
||||
// build object detail
|
||||
.map((response) -> {
|
||||
checkResult(response, "getMetadata");
|
||||
return new ObjectDetail(properties.getBucket(), objectKey, response);
|
||||
})
|
||||
// close client
|
||||
.doFinally((signalType) -> {
|
||||
uploadingFile.remove(uploadingMapKey);
|
||||
s3client.close();
|
||||
});
|
||||
)
|
||||
.onErrorMap(Exceptions::isRetryExhausted,
|
||||
throwable -> new ServerWebInputException(throwable.getCause().getMessage()));
|
||||
}
|
||||
|
||||
|
||||
private Mono<CompletedPart> uploadPart(UploadState uploadState, ByteBuffer buffer, S3AsyncClient s3client) {
|
||||
private Mono<CompletedPart> uploadPart(UploadState uploadState, ByteBuffer buffer,
|
||||
S3Client s3client) {
|
||||
final int partNumber = ++uploadState.partCounter;
|
||||
return Mono
|
||||
.fromFuture(s3client.uploadPart(UploadPartRequest.builder()
|
||||
.bucket(uploadState.bucket)
|
||||
.key(uploadState.objectKey)
|
||||
.partNumber(partNumber)
|
||||
.uploadId(uploadState.uploadId)
|
||||
.contentLength((long) buffer.capacity())
|
||||
.build(),
|
||||
AsyncRequestBody.fromPublisher(Mono.just(buffer))))
|
||||
.map((uploadPartResult) -> {
|
||||
checkResult(uploadPartResult, "uploadPart");
|
||||
return CompletedPart.builder()
|
||||
.eTag(uploadPartResult.eTag())
|
||||
.partNumber(partNumber)
|
||||
.build();
|
||||
});
|
||||
return Mono.just(s3client.uploadPart(UploadPartRequest.builder()
|
||||
.bucket(uploadState.properties.getBucket())
|
||||
.key(uploadState.objectKey)
|
||||
.partNumber(partNumber)
|
||||
.uploadId(uploadState.uploadId)
|
||||
.contentLength((long) buffer.capacity())
|
||||
.build(),
|
||||
RequestBody.fromByteBuffer(buffer)))
|
||||
.map((uploadPartResult) -> {
|
||||
checkResult(uploadPartResult, "uploadPart");
|
||||
return CompletedPart.builder()
|
||||
.eTag(uploadPartResult.eTag())
|
||||
.partNumber(partNumber)
|
||||
.build();
|
||||
});
|
||||
}
|
||||
|
||||
private static void checkResult(SdkResponse result, String operation) {
|
||||
@@ -262,9 +434,7 @@ public class S3OsAttachmentHandler implements AttachmentHandler {
|
||||
}
|
||||
|
||||
ByteBuffer partData = ByteBuffer.allocate(partSize);
|
||||
buffers.forEach((buffer) -> {
|
||||
partData.put(buffer.toByteBuffer());
|
||||
});
|
||||
buffers.forEach((buffer) -> partData.put(buffer.toByteBuffer()));
|
||||
|
||||
// Reset read pointer to first byte
|
||||
partData.rewind();
|
||||
@@ -275,28 +445,44 @@ public class S3OsAttachmentHandler implements AttachmentHandler {
|
||||
|
||||
boolean shouldHandle(Policy policy) {
|
||||
if (policy == null || policy.getSpec() == null ||
|
||||
policy.getSpec().getTemplateName() == null) {
|
||||
policy.getSpec().getTemplateName() == null) {
|
||||
return false;
|
||||
}
|
||||
String templateName = policy.getSpec().getTemplateName();
|
||||
return "s3os".equals(templateName);
|
||||
}
|
||||
|
||||
record ObjectDetail(String bucketName, String objectKey, HeadObjectResponse objectMetadata) {
|
||||
record ObjectDetail(UploadState uploadState, HeadObjectResponse objectMetadata) {
|
||||
}
|
||||
|
||||
@Data
|
||||
static class UploadState {
|
||||
String bucket;
|
||||
String objectKey;
|
||||
final S3OsProperties properties;
|
||||
final String originalFileName;
|
||||
String uploadId;
|
||||
int partCounter;
|
||||
Map<Integer, CompletedPart> completedParts = new HashMap<>();
|
||||
int buffered = 0;
|
||||
String contentType;
|
||||
String fileName;
|
||||
String objectKey;
|
||||
boolean needRemoveMapKey = false;
|
||||
|
||||
UploadState(String bucket, String objectKey) {
|
||||
this.bucket = bucket;
|
||||
this.objectKey = objectKey;
|
||||
public UploadState(S3OsProperties properties, String fileName) {
|
||||
this.properties = properties;
|
||||
this.originalFileName = fileName;
|
||||
this.fileName = fileName;
|
||||
this.objectKey = properties.getObjectName(fileName);
|
||||
this.contentType = MediaTypeFactory.getMediaType(fileName)
|
||||
.orElse(MediaType.APPLICATION_OCTET_STREAM).toString();
|
||||
}
|
||||
|
||||
public String getUploadingMapKey() {
|
||||
return properties.getBucket() + "/" + objectKey;
|
||||
}
|
||||
|
||||
public void randomFileName() {
|
||||
this.fileName = FileNameUtils.randomFileName(originalFileName, 4);
|
||||
this.objectKey = properties.getObjectName(fileName);
|
||||
}
|
||||
}
|
||||
|
||||
|
@@ -4,8 +4,8 @@ metadata:
|
||||
name: PluginS3ObjectStorage
|
||||
spec:
|
||||
enabled: true
|
||||
version: 1.3.0
|
||||
requires: ">=2.0.0"
|
||||
version: 1.4.1
|
||||
requires: ">=2.5.0"
|
||||
author:
|
||||
name: longjuan
|
||||
website: https://github.com/longjuan
|
||||
|
@@ -1,12 +1,19 @@
|
||||
package run.halo.s3os;
|
||||
|
||||
import static java.nio.charset.StandardCharsets.UTF_8;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.util.List;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.core.io.buffer.DataBuffer;
|
||||
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.test.StepVerifier;
|
||||
import run.halo.app.core.extension.attachment.Policy;
|
||||
|
||||
class S3OsAttachmentHandlerTest {
|
||||
@@ -33,4 +40,57 @@ class S3OsAttachmentHandlerTest {
|
||||
// policy is null
|
||||
assertFalse(handler.shouldHandle(null));
|
||||
}
|
||||
|
||||
@Test
|
||||
void reshapeDataBufferWithSmallerBufferSize() {
|
||||
var handler = new S3OsAttachmentHandler();
|
||||
var factory = DefaultDataBufferFactory.sharedInstance;
|
||||
var content = Flux.<DataBuffer>fromIterable(List.of(factory.wrap("halo".getBytes())));
|
||||
|
||||
StepVerifier.create(handler.reshape(content, 2))
|
||||
.assertNext(dataBuffer -> {
|
||||
var str = dataBuffer.toString(UTF_8);
|
||||
assertEquals("ha", str);
|
||||
})
|
||||
.assertNext(dataBuffer -> {
|
||||
var str = dataBuffer.toString(UTF_8);
|
||||
assertEquals("lo", str);
|
||||
})
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test
|
||||
void reshapeDataBufferWithBiggerBufferSize() {
|
||||
var handler = new S3OsAttachmentHandler();
|
||||
var factory = DefaultDataBufferFactory.sharedInstance;
|
||||
var content = Flux.<DataBuffer>fromIterable(List.of(factory.wrap("halo".getBytes())));
|
||||
|
||||
StepVerifier.create(handler.reshape(content, 10))
|
||||
.assertNext(dataBuffer -> {
|
||||
var str = dataBuffer.toString(UTF_8);
|
||||
assertEquals("halo", str);
|
||||
})
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test
|
||||
void reshapeDataBuffersWithBiggerBufferSize() {
|
||||
var handler = new S3OsAttachmentHandler();
|
||||
var factory = DefaultDataBufferFactory.sharedInstance;
|
||||
var content = Flux.<DataBuffer>fromIterable(List.of(
|
||||
factory.wrap("ha".getBytes()),
|
||||
factory.wrap("lo".getBytes())
|
||||
));
|
||||
|
||||
StepVerifier.create(handler.reshape(content, 3))
|
||||
.assertNext(dataBuffer -> {
|
||||
var str = dataBuffer.toString(UTF_8);
|
||||
assertEquals("hal", str);
|
||||
})
|
||||
.assertNext(dataBuffer -> {
|
||||
var str = dataBuffer.toString(UTF_8);
|
||||
assertEquals("o", str);
|
||||
})
|
||||
.verifyComplete();
|
||||
}
|
||||
}
|
||||
|
Reference in New Issue
Block a user