Pull request #371: REPORT-100817 fix: s3插件对于分片上传的支持

Merge in PG/plugin-repository-s3 from ~AFLY/plugin-repository-s3:release/11.0 to release/11.0

* commit '42bc64567d78c2d42502d5fb9f4ef27eb575acb3':
  REPORT-100817 fix: s3插件对于分片上传的支持
  REPORT-100817 fix: s3插件对于分片上传的支持

plugin.xml

@@ -11,6 +11,7 @@
     <vendor>richie</vendor>
     <description><![CDATA[使用支持S3协议的云存储文件系统作为文件服务器。]]></description>
     <change-notes><![CDATA[
+        [2023-07-24]支持大文件分片上传。 <br/>
         [2023-06-30]修复默认配置获取错误的问题，过滤有问题的路径。 <br/>
         [2023-03-28]第三方组件升级。 <br/>
         [2023-01-03]优化写文件性能; 修复文件太多显示不全的问题。<br/>

src/main/java/com/fanruan/fs/s3/repository/core/S3ResourceRepository.java

@@ -7,25 +7,34 @@ import com.amazonaws.auth.BasicAWSCredentials;
 import com.amazonaws.client.builder.AwsClientBuilder;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.GetObjectRequest;
+import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
+import com.amazonaws.services.s3.model.InitiateMultipartUploadResult;
 import com.amazonaws.services.s3.model.ListObjectsRequest;
 import com.amazonaws.services.s3.model.ObjectListing;
 import com.amazonaws.services.s3.model.ObjectMetadata;
+import com.amazonaws.services.s3.model.PartETag;
 import com.amazonaws.services.s3.model.PutObjectRequest;
 import com.amazonaws.services.s3.model.S3Object;
 import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.amazonaws.services.s3.model.UploadPartRequest;
+import com.amazonaws.services.s3.model.UploadPartResult;
 import com.amazonaws.util.IOUtils;
 import com.fanruan.api.log.LogKit;
 import com.fanruan.api.util.StringKit;
 import com.fr.io.repository.FineFileEntry;
 import com.fr.io.repository.base.BaseResourceRepository;
+import com.fr.io.utils.ResourceIOUtils;
+import com.fr.log.FineLoggerFactory;
 import com.fr.stable.Filter;
 import com.fr.stable.StringUtils;
 import com.fr.third.org.apache.commons.io.output.NullOutputStream;
 import com.fr.workspace.resource.ResourceIOException;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.net.URL;
 import java.net.URLConnection;
@@ -41,6 +50,10 @@ public class S3ResourceRepository extends BaseResourceRepository {
 
     private static final int PAGE_SIZE = 1000;
 
+    private static final int PART_SIZE = 5 * 1024 * 1024;
+
+    private static final int MULTIPART_UPLOAD_LIMIT = 4 * PART_SIZE;
+
     private static final String DELIMITER = "/";
 
     public static final String HTTP = "http:";
@@ -158,6 +171,61 @@ public class S3ResourceRepository extends BaseResourceRepository {
         s3.putObject(bucket, path, new ByteArrayInputStream(data), metadata);
     }
 
+    @Override
+    public void write(String path, InputStream inputStream) throws ResourceIOException {
+        long dataLength = 0;
+        try {
+            dataLength = inputStream.available();
+        } catch (IOException e) {
+            FineLoggerFactory.getLogger().error(e.getMessage(), e);
+        }
+        //超过一定大小才使用分片上传，小文件来说，网络传输时间可能较短，且上传失败的风险相对较低。
+        //在网络稳定的情况下，使用分片上传可能没有太多的优势，反而增加了额外开销和复杂性
+        if (dataLength > MULTIPART_UPLOAD_LIMIT) {
+            try {
+                // Step 1: 初始化分片上传
+                InitiateMultipartUploadRequest initRequest = new InitiateMultipartUploadRequest(bucket, path);
+                InitiateMultipartUploadResult initResponse = s3.initiateMultipartUpload(initRequest);
+                String uploadId = initResponse.getUploadId();
+
+                // Step 2: 分片上传文件
+                List<PartETag> partETags = new ArrayList<>();
+                long position = 0;
+
+                for (int partNumber = 1; position < dataLength; partNumber++) {
+                    // 最后一个分片可能小于5MB
+                    long partSizeBytes = Math.min(PART_SIZE, dataLength - position);
+                    byte[] bytes = new byte[(int) partSizeBytes];
+                    inputStream.read(bytes);
+
+                    // 创建上传请求
+                    UploadPartRequest uploadRequest = new UploadPartRequest()
+                            .withBucketName(bucket)
+                            .withKey(path)
+                            .withUploadId(uploadId)
+                            .withPartNumber(partNumber)
+                            .withInputStream(new ByteArrayInputStream(bytes))
+                            .withPartSize(partSizeBytes);
+
+                    // 上传分片
+                    UploadPartResult uploadResult = s3.uploadPart(uploadRequest);
+                    partETags.add(uploadResult.getPartETag());
+
+                    position += partSizeBytes;
+                }
+
+                // Step 3: 完成分片上传
+                CompleteMultipartUploadRequest compRequest = new CompleteMultipartUploadRequest(bucket, path, uploadId, partETags);
+                s3.completeMultipartUpload(compRequest);
+            } catch (IOException e) {
+                throw new ResourceIOException(e);
+            } finally {
+                ResourceIOUtils.close(inputStream);
+            }
+        } else {
+            super.write(path, inputStream);
+        }
+    }
 
     @Override
     public boolean createFile(String path) {