Multiple Files Transfer Feature (#61)

* feat: multiple files transfer * feat: multiple files transfer * chore: examples configurable data
Digital-Ecosystems · Jan 8, 2024 · 1521675 · 1521675
1 parent b194299
commit 1521675
Show file tree

Hide file tree

Showing 15 changed files with 416 additions and 187 deletions.
diff --git a/...-extension/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/api/S3ConnectorApi.java b/...-extension/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/api/S3ConnectorApi.java
@@ -18,7 +18,8 @@
 
 import com.ionos.edc.extension.s3.connector.ionosapi.TemporaryKey;
 
-import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.util.List;
 
 @ExtensionPoint
 public interface S3ConnectorApi {
@@ -27,10 +28,12 @@ public interface S3ConnectorApi {
 
     boolean bucketExists(String bucketName);
 
-    void uploadParts(String bucketName, String fileName, ByteArrayInputStream  part);
-
-    byte[] getFile(String bucketName, String fileName);
-
+    void uploadObject(String bucketName, String objectName, InputStream stream);
+
+    InputStream getObject(String bucketName, String objectName);
+
+    List<S3Object> listObjects(String bucketName, String objectName);
+
     TemporaryKey createTemporaryKey();
 
     void deleteTemporaryKey(String accessKey);

diff --git a/...ension/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/api/S3ConnectorApiImpl.java b/...ension/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/api/S3ConnectorApiImpl.java
@@ -18,15 +18,13 @@
 import com.ionos.edc.extension.s3.connector.ionosapi.HttpConnector;
 import com.ionos.edc.extension.s3.connector.ionosapi.TemporaryKey;
 
-import io.minio.BucketExistsArgs;
-import io.minio.GetObjectArgs;
-import io.minio.MakeBucketArgs;
-import io.minio.MinioClient;
-import io.minio.PutObjectArgs;
+import io.minio.*;
 import org.eclipse.edc.spi.EdcException;
 
-import java.io.ByteArrayInputStream;
 import java.io.InputStream;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 
 public class S3ConnectorApiImpl implements S3ConnectorApi {
 
@@ -35,62 +33,103 @@ public class S3ConnectorApiImpl implements S3ConnectorApi {
 
     private MinioClient minioClient;
     private final String region;
-    private final String token;
+    private String token;
+    private final Integer maxFiles;
 
-
-    public S3ConnectorApiImpl(String endpoint, String accessKey, String secretKey, String token) {
-        if (accessKey != null && secretKey != null && endpoint != null)
+    public S3ConnectorApiImpl(String endpoint, String accessKey, String secretKey, int maxFiles) {
+        if(accessKey != null && secretKey  != null && endpoint != null)
             this.minioClient = minConnector.connect(endpoint, accessKey, secretKey);
         this.region = getRegion(endpoint);
+        this.token = "";
+        this.maxFiles = maxFiles;
+    }
+    public S3ConnectorApiImpl(String endpoint, String accessKey, String secretKey, String token, int maxFiles) {
+        this(endpoint, accessKey, secretKey, maxFiles);
         this.token = token;
     }
 
+
     @Override
     public void createBucket(String bucketName) {
         if (!bucketExists(bucketName.toLowerCase())) {
             try {
-                minioClient.makeBucket(MakeBucketArgs.builder().bucket(bucketName.toLowerCase()).region(region).build());
+                minioClient.makeBucket(MakeBucketArgs.builder()
+                        .bucket(bucketName.toLowerCase())
+                        .region(region)
+                        .build());
             } catch (Exception e) {
                 throw new EdcException("Creating bucket: " + e.getMessage());
             }
         }
     }
 
     @Override
-    public void uploadParts(String bucketName, String fileName, ByteArrayInputStream  part) {
+    public boolean bucketExists(String bucketName) {
+        try {
+            return minioClient.bucketExists(BucketExistsArgs.builder()
+                    .bucket(bucketName.toLowerCase())
+                    .region(this.region)
+                    .build());
+        } catch (Exception e) {
+            throw new EdcException("Verifying if bucket exists - " + e.getMessage());
+        }
+    }
+
+    @Override
+    public void uploadObject(String bucketName, String objectName, InputStream stream) {
         if (!bucketExists(bucketName.toLowerCase())) {
             createBucket(bucketName.toLowerCase());
         }
 
         try {
-            minioClient.putObject(PutObjectArgs.builder().bucket(bucketName.toLowerCase()).region(region).object(fileName).stream(part, part.available(), -1).build());
+            minioClient.putObject(PutObjectArgs.builder()
+                    .bucket(bucketName.toLowerCase())
+                    .region(region)
+                    .object(objectName)
+                    .stream(stream, stream.available(), -1)
+                    .build());
         } catch (Exception e) {
             throw new EdcException("Uploading parts: " + e.getMessage());
         }
     }
 
     @Override
-    public byte[] getFile(String bucketName, String fileName) {
+    public InputStream getObject(String bucketName, String objectName) {
         if (!bucketExists(bucketName.toLowerCase())) {
             throw new EdcException("Bucket not found - " + bucketName);
         }
 
-        InputStream stream;
         try {
-            stream = minioClient.getObject(GetObjectArgs.builder().bucket(bucketName.toLowerCase()).region(region).object(fileName).build());
-            return stream.readAllBytes();
+            return minioClient.getObject(GetObjectArgs.builder()
+                    .bucket(bucketName.toLowerCase())
+                    .region(region)
+                    .object(objectName)
+                    .build());
         } catch (Exception e) {
             throw new EdcException("Getting file - " + e.getMessage());
         }
     }
 
     @Override
-    public boolean bucketExists(String bucketName) {
-        try {
-            return minioClient.bucketExists(BucketExistsArgs.builder().bucket(bucketName.toLowerCase()).region(this.region).build());
-        } catch (Exception e) {
-            throw new EdcException("Verifying if bucket exists - " + e.getMessage());
-        }
+    public List<S3Object> listObjects(String bucketName, String objectName) {
+
+        var objects = minioClient.listObjects(ListObjectsArgs.builder()
+                .bucket(bucketName.toLowerCase())
+                .prefix(objectName)
+                .recursive(true)
+                .maxKeys(maxFiles)
+                .build());
+
+        return StreamSupport.stream(objects.spliterator(), false)
+                .map(item -> {
+                    try {
+                        return item.get();
+                    } catch (Exception e) {
+                        throw new EdcException("Error fetching object", e);
+                    }
+                })
+                .map(item -> new S3Object(item.objectName(), item.size()))
+                .collect(Collectors.toList());
     }
 
     @Override

diff --git a/edc-ionos-extension/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/api/S3Object.java b/edc-ionos-extension/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/api/S3Object.java
@@ -0,0 +1,35 @@
+/*
+ *  Copyright (c) 2023 IONOS
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *      IONOS
+ *
+ */
+
+package com.ionos.edc.extension.s3.api;
+
+public record S3Object(String objectName, long size) {
+
+    public boolean isRootObject(String blobName) {
+        return (objectName.equals(blobName) ||  objectName.equals(blobName + "/"));
+    }
+
+    public String shortObjectName(String blobName) {
+        if (isRootObject(blobName))
+            return objectName;
+
+        var shortObjectName = objectName.replaceFirst(blobName, "");
+
+        if (shortObjectName.indexOf("/") == 0)
+            return shortObjectName.replaceFirst("/", "");
+        else
+            return shortObjectName;
+    }
+
+}
diff --git a/...core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/configuration/S3CoreExtension.java b/...core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/configuration/S3CoreExtension.java
@@ -24,15 +24,17 @@
 import org.eclipse.edc.spi.system.ServiceExtension;
 import org.eclipse.edc.spi.system.ServiceExtensionContext;
 
+import static com.ionos.edc.extension.s3.schema.IonosSettingsSchema.IONOS_ACCESS_KEY;
+import static com.ionos.edc.extension.s3.schema.IonosSettingsSchema.IONOS_SECRET_KEY;
+import static com.ionos.edc.extension.s3.schema.IonosSettingsSchema.IONOS_ENDPOINT;
+import static com.ionos.edc.extension.s3.schema.IonosSettingsSchema.IONOS_TOKEN;
+import static com.ionos.edc.extension.s3.schema.IonosSettingsSchema.IONOS_MAX_FILES_DEFAULT;
+
 @Provides(S3ConnectorApi.class)
 @Extension(value = S3CoreExtension.NAME)
 public class S3CoreExtension implements ServiceExtension {
 
     public static final String NAME = "IonosS3";
-    private static final String IONOS_ACCESS_KEY = "edc.ionos.access.key";
-    private static final String IONOS_SECRET_KEY = "edc.ionos.secret.key";
-    private static final String IONOS_ENDPOINT = "edc.ionos.endpoint";
-    private static final String IONOS_TOKEN = "edc.ionos.token";
 
     @Inject
     private Vault vault;
@@ -61,7 +63,7 @@ public void initialize(ServiceExtensionContext context) {
               token = context.getSetting(IONOS_TOKEN, IONOS_TOKEN);
         }
 
-        var s3Api = new S3ConnectorApiImpl(endPoint, accessKey, secretKey, token);
+        var s3Api = new S3ConnectorApiImpl(endPoint, accessKey, secretKey, token, IONOS_MAX_FILES_DEFAULT);
         context.registerService(S3ConnectorApi.class, s3Api);
     }
 }
diff --git a/...sion/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/schema/IonosBucketSchema.java b/...sion/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/schema/IonosBucketSchema.java
@@ -21,6 +21,8 @@ public interface IonosBucketSchema {
     String STORAGE_NAME = EDC_NAMESPACE + "storage";
     String BUCKET_NAME = EDC_NAMESPACE + "bucketName";
     String BLOB_NAME = EDC_NAMESPACE + "blobName";
+    String FILTER_INCLUDES = EDC_NAMESPACE + "filter.includes";
+    String FILTER_EXCLUDES = EDC_NAMESPACE + "filter.excludes";
     String ACCESS_KEY_ID = EDC_NAMESPACE + "accessKey";
     String SECRET_ACCESS_KEY = EDC_NAMESPACE + "secretKey";
 }
diff --git a/...on/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/schema/IonosSettingsSchema.java b/...on/core-ionos-s3/src/main/java/com/ionos/edc/extension/s3/schema/IonosSettingsSchema.java
@@ -0,0 +1,23 @@
+/*
+ *  Copyright (c) 2022 IONOS
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *      IONOS
+ *
+ */
+
+package com.ionos.edc.extension.s3.schema;
+
+public interface IonosSettingsSchema {
+    String IONOS_ACCESS_KEY = "edc.ionos.access.key";
+    String IONOS_SECRET_KEY = "edc.ionos.secret.key";
+    String IONOS_ENDPOINT = "edc.ionos.endpoint";
+    String IONOS_TOKEN = "edc.ionos.token";
+    int IONOS_MAX_FILES_DEFAULT = 1000;
+}
diff --git a/...ne-ionos-s3/src/main/java/com/ionos/edc/dataplane/ionos/s3/DataPlaneIonosS3Extension.java b/...ne-ionos-s3/src/main/java/com/ionos/edc/dataplane/ionos/s3/DataPlaneIonosS3Extension.java
@@ -33,7 +33,7 @@ public class DataPlaneIonosS3Extension implements ServiceExtension {
 
     @Inject
     private S3ConnectorApi s3Api;
-    
+
     @Inject
     private DataTransferExecutorServiceContainer executorContainer;
 
@@ -50,15 +50,12 @@ public String name() {
 
     @Override
     public void initialize(ServiceExtensionContext context) {
-
-
         var monitor = context.getMonitor();
 
-        var sourceFactory = new IonosDataSourceFactory(s3Api,typeManager);
+        var sourceFactory = new IonosDataSourceFactory(s3Api, monitor);
         pipelineService.registerFactory(sourceFactory);
 
-        var sinkFactory = new IonosDataSinkFactory(s3Api, executorContainer.getExecutorService(), monitor, vault,
-            typeManager);
+        var sinkFactory = new IonosDataSinkFactory(s3Api, executorContainer.getExecutorService(), monitor, vault, typeManager);
         pipelineService.registerFactory(sinkFactory);
         context.getMonitor().info("File Transfer Extension initialized!");
     }

diff --git a/...ion/data-plane-ionos-s3/src/main/java/com/ionos/edc/dataplane/ionos/s3/IonosDataSink.java b/...ion/data-plane-ionos-s3/src/main/java/com/ionos/edc/dataplane/ionos/s3/IonosDataSink.java
@@ -21,7 +21,6 @@
 import org.eclipse.edc.spi.EdcException;
 import org.jetbrains.annotations.NotNull;
 
-import java.io.ByteArrayInputStream;
 import java.util.List;
 import java.util.Objects;
 
@@ -50,7 +49,7 @@ protected StreamResult<Object> transferParts(List<DataSource.Part> parts) {
                 if(input == null) {
                     throw new EdcException("Error transferring file");
                 }
-                s3Api.uploadParts(bucketName, blobName, new ByteArrayInputStream(input.readAllBytes()));
+                s3Api.uploadObject(bucketName, blobName, input);
             } catch (Exception e) {
                 return uploadFailure(e, blobName);
             }

diff --git a/...a-plane-ionos-s3/src/main/java/com/ionos/edc/dataplane/ionos/s3/IonosDataSinkFactory.java b/...a-plane-ionos-s3/src/main/java/com/ionos/edc/dataplane/ionos/s3/IonosDataSinkFactory.java
@@ -14,7 +14,6 @@
 
 package com.ionos.edc.dataplane.ionos.s3;
 
-
 import com.ionos.edc.dataplane.ionos.s3.validation.IonosSinkDataAddressValidationRule;
 import com.ionos.edc.extension.s3.api.S3ConnectorApi;
 import com.ionos.edc.extension.s3.api.S3ConnectorApiImpl;
@@ -34,6 +33,9 @@
 import org.jetbrains.annotations.NotNull;
 
 import java.util.concurrent.ExecutorService;
+
+import static com.ionos.edc.extension.s3.schema.IonosSettingsSchema.IONOS_MAX_FILES_DEFAULT;
+
 public class IonosDataSinkFactory implements DataSinkFactory {
 
     private static final String DEFAULT_STORAGE = "s3-eu-central-1.ionoscloud.com";
@@ -81,7 +83,11 @@ public DataSink createSink(DataFlowRequest request) {
             var token = typeManager.readValue(secret, IonosToken.class);
 
             if (destination.getStringProperty(IonosBucketSchema.STORAGE_NAME) != null) {
-                var s3ApiTemp = new S3ConnectorApiImpl(destination.getStringProperty(IonosBucketSchema.STORAGE_NAME), token.getAccessKey(), token.getSecretKey(), "");
+                var s3ApiTemp = new S3ConnectorApiImpl(destination.getStringProperty(IonosBucketSchema.STORAGE_NAME),
+                        token.getAccessKey(),
+                        token.getSecretKey(),
+                        "",
+                        IONOS_MAX_FILES_DEFAULT);
                 return IonosDataSink.Builder.newInstance()
                         .bucketName(destination.getStringProperty(IonosBucketSchema.BUCKET_NAME))
                         .blobName(destination.getStringProperty(IonosBucketSchema.BLOB_NAME))
@@ -90,7 +96,11 @@ public DataSink createSink(DataFlowRequest request) {
                         .monitor(monitor).s3Api(s3ApiTemp)
                         .build();
             } else {
-                var s3ApiTemp = new S3ConnectorApiImpl(DEFAULT_STORAGE, token.getAccessKey(), token.getSecretKey(), "");
+                var s3ApiTemp = new S3ConnectorApiImpl(DEFAULT_STORAGE,
+                        token.getAccessKey(),
+                        token.getSecretKey(),
+                        "",
+                        IONOS_MAX_FILES_DEFAULT);
                 return IonosDataSink.Builder.newInstance()
                         .bucketName(destination.getStringProperty(IonosBucketSchema.BUCKET_NAME))
                         .blobName(destination.getStringProperty(IonosBucketSchema.BLOB_NAME))