support fetching versioned s3 files (#297)

* support fetching versioned s3 files

* bump travis

README.md

@@ -142,3 +142,4 @@ val result = program
 result should be(...)
 ```
 **TODO:** Stream send SQS messages
+

fs2-aws/src/main/scala/fs2/aws/s3/package.scala

@@ -69,6 +69,24 @@ package object s3 {
       closeAfterUse = true
     )
 
+  def readS3VersionedFile[F[_]](
+    bucket: String,
+    key: String,
+    version: String,
+    blockingEC: ExecutionContext,
+    amazonS3: AmazonS3 = AmazonS3ClientBuilder.defaultClient()
+  )(
+    implicit F: Effect[F],
+    cs: ContextShift[F],
+    s3Client: S3Client[F] = S3Client[F](amazonS3)
+  ): Stream[F, Byte] =
+    readInputStream[F](
+      s3Client.getObjectContent(new GetObjectRequest(bucket, key, version)),
+      chunkSize = 8192,
+      blocker = Blocker.liftExecutionContext(blockingEC),
+      closeAfterUse = true
+    )
+
   def uploadS3FileMultipart[F[_]](
     bucket: String,
     key: String,

fs2-aws/src/test/resources/jsontest.json_vABC

@@ -0,0 +1,4 @@
+{"this": 1}
+{"is": 2}
+{"versioned": 3}
+{"content": 4}

fs2-aws/src/test/scala/fs2/aws/S3Spec.scala

@@ -54,6 +54,19 @@ class S3Spec extends AnyFlatSpec with Matchers {
     )
   }
 
+  "Downloading the versioned JSON test file" should "return the same content" in {
+    readS3VersionedFile[IO]("resources", "jsontest.json", version = "ABC", blockingEC, mockS3)
+      .through(fs2.text.utf8Decode)
+      .through(fs2.text.lines)
+      .compile
+      .toVector
+      .unsafeRunSync
+      .reduce(_ + _)
+      .concat("") should be(
+      """{"this": 1}{"is": 2}{"versioned": 3}{"content": 4}"""
+    )
+  }
+
   "big chunk size but small entire text" should "be trimmed to content" in {
     readS3FileMultipart[IO]("resources", "jsontest1.json", 25, mockS3)
       .through(fs2.text.utf8Decode)

fs2-aws/src/test/scala/fs2/aws/utils/package.scala

@@ -54,7 +54,11 @@ package object utils {
       IO[ByteArrayInputStream] {
         val fileContent: Array[Byte] =
           try {
-            Source.fromResource(getObjectRequest.getKey).mkString.getBytes
+            val testS3Resource = Option(getObjectRequest.getVersionId) match {
+              case Some(version) => s"${getObjectRequest.getKey}_v$version"
+              case None          => getObjectRequest.getKey
+            }
+            Source.fromResource(testS3Resource).mkString.getBytes
           } catch {
             case _: FileNotFoundException => throw new AmazonS3Exception("File not found")
             case e: Throwable             => throw e