Merge pull request #8 from VEuPathDB/object-stream

Implement object stream functionality
VEuPathDB · Apr 18, 2023 · e12ade7 · e12ade7
2 parents 92bf234 + deb73ad
commit e12ade7
Show file tree

Hide file tree

Showing 2 changed files with 50 additions and 2 deletions.
diff --git a/build.gradle.kts b/build.gradle.kts
@@ -8,7 +8,7 @@ plugins {
 }
 
 group = "org.veupathdb.lib.s3"
-version = "0.4.1+s34k-0.8.0"
+version = "0.5.0+s34k-0.9.0"
 
 java {
   sourceCompatibility = JavaVersion.VERSION_1_8
@@ -39,7 +39,7 @@ dependencies {
   implementation("io.minio:minio:8.4.5")
 
   @Suppress("GradlePackageVersionRange")
-  api("org.veupathdb.lib.s3:s34k-core:0.7.0+s34k-0.8.0")
+  api("org.veupathdb.lib.s3:s34k-core:0.8.0+s34k-0.9.0-SNAPSHOT")
 
   testImplementation(kotlin("test"))
 }

diff --git a/src/main/kotlin/org/veupathdb/lib/s3/s34k/minio/BucketObjectContainer.kt b/src/main/kotlin/org/veupathdb/lib/s3/s34k/minio/BucketObjectContainer.kt
@@ -15,6 +15,7 @@ import org.veupathdb.lib.s3.s34k.*
 import org.veupathdb.lib.s3.s34k.buckets.S3Bucket
 import org.veupathdb.lib.s3.s34k.core.objects.AbstractObjectContainer
 import org.veupathdb.lib.s3.s34k.core.objects.BasicObjectList
+import org.veupathdb.lib.s3.s34k.core.objects.BasicObjectStream
 import org.veupathdb.lib.s3.s34k.core.objects.BasicSubPathListing
 import org.veupathdb.lib.s3.s34k.errors.MultiObjectDeleteError
 import org.veupathdb.lib.s3.s34k.errors.ObjectDeleteError
@@ -424,6 +425,53 @@ internal class BucketObjectContainer(
     return StatObject(bucket, path, params, minio)
   }
 
+  override fun stream(params: ObjectStreamParams): ObjectStream {
+    return BasicObjectStream(
+      minio.listObjects(ListObjectsArgs.builder()
+        .bucket(bucket)
+        .prefix(params.prefix)
+        .region(params, bucket)
+        .recursive(true)
+        .headers(params.headers)
+        .queryParams(params.queryParams)
+        .build())
+        .toStream()
+        .map(Result<Item>::get)
+        .map { MObject(
+          it.objectName(),
+          it.lastModified().toOffsetDateTime(),
+          it.etag(),
+          bucket.region,
+          MHeaders(),
+          bucket,
+          minio
+        ) }
+    )
+  }
+
+  override fun streamAll(params: ObjectStreamAllParams): ObjectStream {
+    return BasicObjectStream(
+      minio.listObjects(ListObjectsArgs.builder()
+        .bucket(bucket)
+        .region(params, bucket)
+        .recursive(true)
+        .headers(params.headers)
+        .queryParams(params.queryParams)
+        .build())
+        .toStream()
+        .map(Result<Item>::get)
+        .map { MObject(
+          it.objectName(),
+          it.lastModified().toOffsetDateTime(),
+          it.etag(),
+          bucket.region,
+          MHeaders(),
+          bucket,
+          minio
+        ) }
+    )
+  }
+
   override fun touch(path: String, params: ObjectTouchParams): S3Object {
     log.debug("Attempting to touch object '{}' in {}", path, bucket)
     return ObjectToucher(bucket, path, params, minio).execute()