Stream a file to AWS S3 using Akka Streams

Table of contents

Reading Time: 2 minutes

In this blog, I’ll show you how to stream a file to AWS S3 using the library –
gfc-aws-s3,

To use the above, add to your dependencies

"com.gilt" %% "gfc-aws-s3" % "0.1.0"

The library contains akka-stream Sources and Sinks to Stream data from and to S3

From my blog on “Upload file in Akka Http“, you can see how to obtain the stream of a file, So once you get that source its pretty simple to stream it to AWS S3.

Let’s first create our AWS S3 client, here we are using Amazon SDK

<br>val s3ClientBuilderBase = AmazonS3ClientBuilder.standard()<br>.withRegion("region")<br>.withPathStyleAccessEnabled(true)
<p>val credentials = new BasicAWSCredentials("accessKey", "secretKey")</p>
<p>s3ClientBuilderBase.withCredentials(new AWSStaticCredentialsProvider(credentials)).build()<br></p>

So we pass this s3Client, in our S3ClientService as shown below, with the name of the bucket that we want the service to work with.

<br>class S3ClientService(<br>s3Client: AmazonS3,<br>bucketName: String<br>)(implicit system: ActorSystem, materializer: Materializer) {
<p>def getSink(path: String): Sink[ByteString, Future[Long]] = {<br>val chunkSize = 10 * 1024 * 1024 // 10 Megabytes<br>val sink: Sink[Byte, Future[Long]] = S3MultipartUploaderSink(s3Client, bucketName, path, chunkSize)<br>Flow[ByteString].mapConcat(identity).toMat(sink)(Keep.right)<br>}</p>
<p>def write(<br>source: Source[ByteString, Any],<br>path: String<br>)(implicit ec: ExecutionContext,<br>materializer: Materializer<br>): Future[Long] =<br>source.runWith(getSink(path))<br>}<br></p>

In this service, we have a write method, just supply the path where you want to store the file and source to file to this, it, in turn, is calling getSink method that takes path as parameter and it will upload your file as a stream to AWS S3, so here the getSink method creates a Sink using S3MultipartUploaderSink of gfc-aws-s3 library, and as that Sink accepts Byte we modify our stream of ByteString to Byte using mapConcat with identity.

Now, Let’s see how we will use this service to stream a file from the browser to Amazon S3 using Akka-Http route.

<br>path("uploadPicture") {<br>(post & withoutSizeLimit) {<br>extractRequestContext { ctx =><br>import ctx.{ executionContext, materializer }<br>entity(as[Multipart.FormData]) { data =>
<p>val extractedData: Future[Map[String, Any]] = data.parts.mapAsyncUnordered(1) {<br>case picture: BodyPart if picture.name == "picture" =><br>val path = "pictures"<br>s3ClientService.write(picture.entity.dataBytes, path).map { _ =><br>"picture" -> path<br>}<br>case expectedPart if expectedPart.name == "pictureName" =><br>expectedPart.toStrict(2.seconds).map(field => field.name -> field.entity.data.utf8String)</p>
<p>}.runFold(Map.empty[String, Any])((map, tuple) => map + tuple)<br>complete(<br>extractedData.map { data =><br>// do some processing and send response<br>}.recover {<br>// handle exception in case of missing fields, or any other exception that can occur<br>}<br>}<br>}<br>}<br>}<br> </p>

In this example, we are uploading a picture to S3 using the service we created above.
The library also has S3MultipartDownload and S3ChunkedDownload to download files from S3, will go through them in our next blog.
Hopefully, this was helpful and you guys can use it if stuck in a similar use case.