Cloud storage FileSystem Accessor refactor

robinhood-jim · robinhood-jim · commit ed5b79108572 · 2024-12-18T15:59:28.000+08:00
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/AbstractCloudStorageFileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/AbstractCloudStorageFileSystemAccessor.java
@@ -5,6 +5,7 @@
 import com.robin.core.fileaccess.meta.DataCollectionMeta;
 import com.robin.core.fileaccess.util.ResourceUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.util.ObjectUtils;
 
@@ -18,6 +19,7 @@
 @Slf4j
 public abstract class AbstractCloudStorageFileSystemAccessor extends AbstractFileSystemAccessor {
     protected String bucketName;
+    protected String tmpFilePath;
     @Override
     public Pair<BufferedReader, InputStream> getInResourceByReader(DataCollectionMeta meta, String resourcePath) throws IOException {
         InputStream inputStream = getInputStreamByConfig(meta);
@@ -54,12 +56,13 @@ public InputStream getRawInputStream(DataCollectionMeta meta, String resourcePat
     public void finishWrite(DataCollectionMeta meta,OutputStream outputStream) {
         try{
             putObject(getBucketName(meta),meta,outputStream);
+            if(!ObjectUtils.isEmpty(tmpFilePath)){
+                FileUtils.deleteQuietly(new File(tmpFilePath));
+            }
         }catch (IOException ex){
             log.error("{}",ex.getMessage());
         }
     }
-
-
     protected InputStream getInputStreamByConfig(DataCollectionMeta meta) {
         return getObject(getBucketName(meta), meta.getPath());
     }
@@ -70,7 +73,7 @@ protected OutputStream getOutputStream(DataCollectionMeta meta) throws IOExcepti
         OutputStream outputStream;
         if(!ObjectUtils.isEmpty(meta.getResourceCfgMap().get(ResourceConst.USETMPFILETAG)) && "true".equalsIgnoreCase(meta.getResourceCfgMap().get(ResourceConst.USETMPFILETAG).toString())){
             String tmpPath = com.robin.core.base.util.FileUtils.getWorkingPath(meta);
-            String tmpFilePath =  tmpPath + ResourceUtil.getProcessFileName(meta.getPath());
+            tmpFilePath =  tmpPath + ResourceUtil.getProcessFileName(meta.getPath());
             outputStream= Files.newOutputStream(Paths.get(tmpFilePath));
         }else {
             outputStream = new ByteArrayOutputStream();
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/BOSFileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/BOSFileSystemAccessor.java
@@ -88,7 +88,6 @@ protected InputStream getObject(String bucketName,String objectName){
     protected boolean putObject(String bucketName, DataCollectionMeta meta, OutputStream outputStream) throws IOException {
         PutObjectResponse result;
         ObjectMetadata metadata=new ObjectMetadata();
-        String tmpFilePath=null;
         if(!ObjectUtils.isEmpty(meta.getContent())){
             metadata.setContentType(meta.getContent().getContentType());
         }
@@ -98,14 +97,8 @@ protected boolean putObject(String bucketName, DataCollectionMeta meta, OutputSt
             result = client.putObject(bucketName, meta.getPath(), new ByteArrayInputStream(byteArrayOutputStream.toByteArray()),metadata);
         }else{
             outputStream.close();
-            String tmpPath = com.robin.core.base.util.FileUtils.getWorkingPath(meta);
-            tmpFilePath = tmpPath + ResourceUtil.getProcessFileName(meta.getPath());
-            metadata.setContentLength(Files.size(Paths.get(tmpFilePath)));
             result=client.putObject(bucketName,meta.getPath(), Files.newInputStream(Paths.get(tmpFilePath)),metadata);
         }
-        if(ObjectUtils.isEmpty(tmpFilePath)){
-            FileUtils.deleteQuietly(new File(tmpFilePath));
-        }
         return !ObjectUtils.isEmpty(result) && !ObjectUtils.isEmpty(result.getETag());
     }
 
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/COSFileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/COSFileSystemAccessor.java
@@ -127,7 +127,6 @@ private boolean upload(DataCollectionMeta meta, OutputStream outputStream) throw
         String bucketName= getBucketName(meta);
         TransferManager transferManager=getManager();
         PutObjectRequest request;
-        String tmpFilePath=null;
         ObjectMetadata objectMetadata = new ObjectMetadata();
         if(!ObjectUtils.isEmpty(meta.getContent())){
             objectMetadata.setContentType(meta.getContent().getContentType());
@@ -137,11 +136,8 @@ private boolean upload(DataCollectionMeta meta, OutputStream outputStream) throw
             request = new PutObjectRequest(bucketName, meta.getPath(), new ByteArrayInputStream(((ByteArrayOutputStream)outputStream).toByteArray()),objectMetadata);
         }else{
             outputStream.close();
-            String tmpPath = com.robin.core.base.util.FileUtils.getWorkingPath(meta);
-            tmpFilePath = tmpPath + ResourceUtil.getProcessFileName(meta.getPath());
             request=new PutObjectRequest(bucketName,meta.getPath(),new File(tmpFilePath));
         }
-
         try {
             Upload upload = transferManager.upload(request, null);
             UploadResult result = upload.waitForUploadResult();
@@ -152,9 +148,6 @@ private boolean upload(DataCollectionMeta meta, OutputStream outputStream) throw
             if (!ObjectUtils.isEmpty(transferManager)) {
                 transferManager.shutdownNow(true);
             }
-            if(!ObjectUtils.isEmpty(tmpFilePath)){
-                FileUtils.deleteQuietly(new File(tmpFilePath));
-            }
         }
         return false;
     }
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/MinioFileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/MinioFileSystemAccessor.java
@@ -74,14 +74,11 @@ protected InputStream getObject(String bucketName,String objectName) {
     @Override
     protected boolean putObject(String bucketName, DataCollectionMeta meta, OutputStream outputStream) throws IOException {
         String contentType=!ObjectUtils.isEmpty(meta.getContent().getContentType())?meta.getContent().getContentType():"application/octet-stream";
-        String tmpFilePath;
         if(ByteArrayOutputStream.class.isAssignableFrom(outputStream.getClass())){
             ByteArrayOutputStream byteArrayOutputStream=(ByteArrayOutputStream)outputStream;
             return MinioUtils.putBucket(client,getBucketName(meta),meta.getPath(),new ByteArrayInputStream(byteArrayOutputStream.toByteArray()),byteArrayOutputStream.size(),contentType);
         }else{
             outputStream.close();
-            String tmpPath = com.robin.core.base.util.FileUtils.getWorkingPath(meta);
-            tmpFilePath = tmpPath + ResourceUtil.getProcessFileName(meta.getPath());
             return MinioUtils.putBucket(client,getBucketName(meta),meta.getPath(), Files.newInputStream(Paths.get(tmpFilePath)),Files.size(Paths.get(tmpFilePath)),contentType);
         }
     }
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/OBSFileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/OBSFileSystemAccessor.java
@@ -11,6 +11,7 @@
 import com.robin.core.base.util.ResourceConst;
 import com.robin.core.fileaccess.meta.DataCollectionMeta;
 import com.robin.core.fileaccess.util.ResourceUtil;
+import lombok.Getter;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
@@ -22,6 +23,7 @@
 /**
  * HUAWEI OBS FileSystemAccessor,must init individual
  */
+@Getter
 public class OBSFileSystemAccessor extends AbstractCloudStorageFileSystemAccessor {
     private String endpoint;
     private String accessKeyId;
@@ -84,7 +86,6 @@ protected InputStream getObject(String bucketName, String objectName) {
 
     @Override
     protected boolean putObject(String bucketName, DataCollectionMeta meta, OutputStream outputStream) throws IOException {
-        String tmpFilePath;
         ObjectMetadata metadata=new ObjectMetadata();
         PutObjectResult result;
         if(!ObjectUtils.isEmpty(meta.getContent())){
@@ -96,9 +97,6 @@ protected boolean putObject(String bucketName, DataCollectionMeta meta, OutputSt
             result=client.putObject(bucketName,meta.getPath(),new ByteArrayInputStream(byteArrayOutputStream.toByteArray()),metadata);
         }else{
             outputStream.close();
-            String tmpPath = com.robin.core.base.util.FileUtils.getWorkingPath(meta);
-            tmpFilePath = tmpPath + ResourceUtil.getProcessFileName(meta.getPath());
-            metadata.setContentLength(Files.size(Paths.get(tmpFilePath)));
             result=client.putObject(bucketName,meta.getPath(), Files.newInputStream(Paths.get(tmpFilePath)),metadata);
         }
         return result.getStatusCode()==200;
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/OSSFileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/OSSFileSystemAccessor.java
@@ -35,7 +35,6 @@ public class OSSFileSystemAccessor extends AbstractCloudStorageFileSystemAccesso
     private String region;
     private String accessKeyId;
     private String securityAccessKey;
-    private String bucketName;
 
     private OSSFileSystemAccessor(){
         this.identifier= Const.FILESYSTEM.ALIYUN.getValue();
@@ -91,7 +90,7 @@ protected InputStream getObject(String bucketName,String objectName){
             if (object.getResponse().isSuccessful()) {
                 return object.getObjectContent();
             } else {
-                throw new RuntimeException("objectName " + objectName + " can not get!");
+                throw new MissingConfigException("objectName " + objectName + " can not get!");
             }
         }else{
             throw new MissingConfigException(" key "+objectName+" not in OSS bucket "+bucketName);
@@ -102,7 +101,6 @@ private Bucket createBucket(String bucketName){
     }
     protected boolean putObject(String bucketName,DataCollectionMeta meta,OutputStream outputStream) throws IOException{
         PutObjectResult result;
-        String tmpFilePath=null;
         ObjectMetadata metadata=new ObjectMetadata();
         if(!ObjectUtils.isEmpty(meta.getContent())){
             metadata.setContentType(meta.getContent().getContentType());
@@ -113,16 +111,9 @@ protected boolean putObject(String bucketName,DataCollectionMeta meta,OutputStre
             result = ossClient.putObject(bucketName, meta.getPath(), new ByteArrayInputStream(byteArrayOutputStream.toByteArray()),metadata);
         }else{
             outputStream.close();
-            String tmpPath = com.robin.core.base.util.FileUtils.getWorkingPath(meta);
-            tmpFilePath = tmpPath + ResourceUtil.getProcessFileName(meta.getPath());
-            metadata.setContentLength(Files.size(Paths.get(tmpFilePath)));
             result=ossClient.putObject(bucketName,meta.getPath(), Files.newInputStream(Paths.get(tmpFilePath)),metadata);
         }
-        ResponseMessage message=result.getResponse();
-        if(message.isSuccessful() && !ObjectUtils.isEmpty(tmpFilePath)){
-            FileUtils.deleteQuietly(new File(tmpFilePath));
-        }
-        return message.isSuccessful();
+        return result.getResponse().isSuccessful();
     }
 
     public static class Builder{
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/QiniuFileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/QiniuFileSystemAccessor.java
@@ -42,7 +42,6 @@ public class QiniuFileSystemAccessor extends AbstractCloudStorageFileSystemAcces
     private String accessKey;
     private String secretKey;
     private Region region;
-    private String bucketName;
     private Gson gson= GsonUtil.getGson();
     private String downDomain;
     private QiniuFileSystemAccessor(){
@@ -95,7 +94,7 @@ private boolean isKeyExist(String bucketName,String key) {
         try {
             FileInfo info = bucketManager.stat(bucketName, key);
             int status = info.status;
-            return true;
+            return status>0;
         } catch (QiniuException ex) {
             log.error("{}", ex.getMessage());
         }
@@ -112,14 +111,11 @@ private long getSize(String bucketName,String key) {
     }
     protected boolean putObject(String token,DataCollectionMeta meta,OutputStream outputStream) throws IOException{
         Response result;
-        String tmpFilePath=null;
         if(ByteArrayOutputStream.class.isAssignableFrom(outputStream.getClass())) {
             ByteArrayOutputStream byteArrayOutputStream=(ByteArrayOutputStream)outputStream;
             result= uploadManager.put(new ByteArrayInputStream(byteArrayOutputStream.toByteArray()),byteArrayOutputStream.size(),meta.getPath(),token,null,meta.getContent().getContentType(),true);
         }else{
             outputStream.close();
-            String tmpPath = com.robin.core.base.util.FileUtils.getWorkingPath(meta);
-            tmpFilePath = tmpPath + ResourceUtil.getProcessFileName(meta.getPath());
             long size=Files.size(Paths.get(tmpFilePath));
             result=uploadManager.put(Files.newInputStream(Paths.get(tmpFilePath)),size,meta.getPath(),token,null,meta.getContent().getContentType(),true);
         }
diff --git a/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/S3FileSystemAccessor.java b/hadooptool/src/main/java/com/robin/comm/fileaccess/fs/S3FileSystemAccessor.java
@@ -94,14 +94,7 @@ protected InputStream getObject(String bucketName, String objectName) {
         return AwsUtils.getObject(client,bucketName,objectName);
     }
 
-    @Override
-    public void finishWrite(DataCollectionMeta meta, OutputStream outputStream) {
-        String bucketName = getBucketName(meta);
-        ByteArrayOutputStream outputStream1=(ByteArrayOutputStream) outputStream;
-        int size=outputStream1.size();
-        String contentType=!ObjectUtils.isEmpty(meta.getContent())?meta.getContent().getContentType():"application/octet-stream";
-        AwsUtils.put(client,bucketName,meta.getPath(),contentType,new ByteArrayInputStream(outputStream1.toByteArray()),new Long(size));
-    }
+
     public static class Builder{
         private S3FileSystemAccessor accessor;
         public static Builder builder(){
