-
Notifications
You must be signed in to change notification settings - Fork 74
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Replace alpakka S3 with fs2-aws and minio with localstack (#4852)
* Read file using fs2-aws * Use S3 storage client to fetch files * Use same group size as the previous implementation * WIP - multipart S3 upload * WIP - check object existence, fetch metadata for file size, refactor * Use S3 base endpoint for absolute path * Remove alpakka s3 dependency * Tidy * Tidy integration test debugging * Don't error when uploading an empty file (calling .lastOrError) * Fix bug in file size calculation --------- Co-authored-by: Daniel Bell <[email protected]> Co-authored-by: Daniel Bell <[email protected]>
- Loading branch information
3 people
authored
Apr 9, 2024
1 parent
ea14433
commit 96a1c94
Showing
19 changed files
with
370 additions
and
312 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
46 changes: 19 additions & 27 deletions
46
...pfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/S3StorageFetchFile.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,45 +1,37 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3 | ||
|
|
||
| import akka.actor.ActorSystem | ||
| import akka.http.scaladsl.model.Uri | ||
| import akka.stream.alpakka.s3.scaladsl.S3 | ||
| import akka.stream.alpakka.s3.{S3Attributes, S3Exception} | ||
| import akka.stream.scaladsl.Sink | ||
| import akka.stream.scaladsl.Source | ||
| import akka.util.ByteString | ||
| import cats.effect.IO | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileAttributes | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StoragesConfig.StorageTypeConfig | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.StorageValue.S3StorageValue | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.FetchFile | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.StorageFileRejection.FetchFileRejection._ | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.client.S3StorageClient | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.AkkaSource | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.stream.StreamConverter | ||
|
|
||
| import java.net.URLDecoder | ||
| import java.nio.charset.StandardCharsets.UTF_8 | ||
| import scala.concurrent.duration.DurationInt | ||
|
|
||
| final class S3StorageFetchFile(value: S3StorageValue, config: StorageTypeConfig)(implicit | ||
| as: ActorSystem | ||
| ) extends FetchFile { | ||
|
|
||
| private val s3Attributes = S3Attributes.settings(value.alpakkaSettings(config)) | ||
| final class S3StorageFetchFile(client: S3StorageClient, bucket: String) extends FetchFile { | ||
|
|
||
| override def apply(attributes: FileAttributes): IO[AkkaSource] = | ||
| apply(attributes.path) | ||
|
|
||
| override def apply(path: Uri.Path): IO[AkkaSource] = | ||
| IO.fromFuture( | ||
| IO.delay( | ||
| S3.download(value.bucket, URLDecoder.decode(path.toString, UTF_8.toString)) | ||
| .withAttributes(s3Attributes) | ||
| .runWith(Sink.head) | ||
| override def apply(path: Uri.Path): IO[AkkaSource] = { | ||
| IO.delay( | ||
| Source.fromGraph( | ||
| StreamConverter( | ||
| client | ||
| .readFile(bucket, URLDecoder.decode(path.toString, UTF_8.toString)) | ||
| .groupWithin(8192, 1.second) | ||
| .map(bytes => ByteString(bytes.toArray)) | ||
| ) | ||
| ) | ||
| ).redeemWith( | ||
| { | ||
| case err: S3Exception => IO.raiseError(UnexpectedFetchError(path.toString, err.toString())) | ||
| case err => IO.raiseError(UnexpectedFetchError(path.toString, err.getMessage)) | ||
| }, | ||
| { | ||
| case Some((source, _)) => IO.pure(source: AkkaSource) | ||
| case None => IO.raiseError(FileNotFound(path.toString())) | ||
| } | ||
| ) | ||
| ).recoverWith { err => | ||
| IO.raiseError(UnexpectedFetchError(path.toString, err.getMessage)) | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.