Improved implementation for FutureTransfer.listenFor

Cast to internal type `AbstractTransfer` so that a
low-level state change listener can be attached.
In light of issue #30, this provides a more foolproof
implementation of listening to transfer events.
There is now a three-way race to complete the
promise that signals the completion/termination of
the transfer.

This adds an integration test for s3, using the
fakes3 ruby gem.

closes #30

Author: dwhjames

.gitignore

@@ -1,7 +1,10 @@
 *.class
 *.log
 
-/dynamodb
+/dynamodb-local
+/fakes3
+
+/Gemfile.lock
 
 # sbt specific
 dist/*

.travis.yml

@@ -9,5 +9,6 @@ jdk:
 cache:
   directories:
     - $HOME/.ivy2/cache
+install: bundle install
 script:
 - sbt ++$TRAVIS_SCALA_VERSION --warn update compile doc awsWrapTest/it:compile && sbt ++$TRAVIS_SCALA_VERSION awsWrapTest/it:test

Gemfile

@@ -0,0 +1,2 @@
+source 'https://rubygems.org'
+gem 'fakes3'

integration/build.sbt

@@ -4,6 +4,7 @@ name := "aws-wrap-test"
 
 libraryDependencies ++= Seq(
   Dependencies.Compile.awsJavaSDK_dynamodb % "it",
+  Dependencies.Compile.awsJavaSDK_s3 % "it",
   Dependencies.Compile.jodaTime % "it",
   Dependencies.Compile.jodaConvert % "it",
   Dependencies.Compile.logback % "it",
@@ -20,10 +21,14 @@ testOptions in IntegrationTest += Tests.Setup { () =>
     println("Start DynamoDB Local")
     System.setProperty("DynamoDB.localMode", "true")
     Process("bash start-dynamodb-local.sh").!
+    println("Start fakes3")
+    Process("bash start-fakes3.sh").!
 }
 
 testOptions in IntegrationTest += Tests.Cleanup { () =>
     println("Stop DynamoDB Local")
     System.clearProperty("DynamoDB.localMode")
     Process("bash stop-dynamodb-local.sh").!
+    println("Stop fakes3")
+    Process("bash stop-fakes3.sh").!
 }

integration/src/it/scala/dynamodb/AwaitHelper.scala renamed to integration/src/it/scala/AwaitHelper.scala

@@ -15,10 +15,9 @@
  * limitations under the License.
  */
 
-package com.github.dwhjames.awswrap.dynamodb
+package com.github.dwhjames.awswrap
 
 import scala.concurrent._
-import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration._
 
 trait AwaitHelper {

integration/src/it/scala/dynamodb/DynamoDBClient.scala

@@ -15,7 +15,8 @@
  * limitations under the License.
  */
 
-package com.github.dwhjames.awswrap.dynamodb
+package com.github.dwhjames.awswrap
+package dynamodb
 
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration._

integration/src/it/scala/s3/FutureTransferSpec.scala

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2015 Daniel W. H. James
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.github.dwhjames.awswrap
+package s3
+
+import scala.concurrent.ExecutionContext.Implicits.global
+
+import java.io.File
+
+import org.scalatest.{ FlatSpec, Matchers }
+
+import com.amazonaws.AmazonClientException
+
+
+class FutureTransferSpec
+  extends FlatSpec
+     with Matchers
+     with S3ClientHelper
+{
+
+  val bucketName = "my-s3-bucket-98bfdf06-9475-4d1a-a235-e0eddd5859f9"
+  override val bucketNames = Seq(bucketName)
+  val objectKey = "test"
+
+  override def afterAll(): Unit = {
+    try {
+      client.client.deleteObject(bucketName, objectKey)
+    } finally {
+      super.afterAll()
+    }
+  }
+
+  "FutureTransfer" should "upload a file" in {
+    val file = new File(
+        this.getClass()
+            .getClassLoader()
+            .getResource("logback-test.xml")
+            .toURI())
+
+    val upload = transferManager.upload(bucketName, objectKey, file)
+
+    await {
+      FutureTransfer.listenFor(upload)
+    }
+    upload.waitForUploadResult()
+    ()
+  }
+
+  it should "download a file" in {
+    val file = File.createTempFile("logback-test", ".xml")
+
+    try {
+      val download = transferManager.download(bucketName, objectKey, file)
+
+      await {
+        FutureTransfer.listenFor(download)
+      }
+      download.waitForCompletion()
+      ()
+    } finally {
+      file.delete()
+      ()
+    }
+  }
+
+}

integration/src/it/scala/s3/S3ClientHelper.scala

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2015 Daniel W. H. James
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.github.dwhjames.awswrap
+package s3
+
+import org.scalatest.{Suite, BeforeAndAfterAll}
+
+import com.amazonaws.auth.BasicAWSCredentials
+import com.amazonaws.services.s3._
+import com.amazonaws.services.s3.transfer._
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+trait S3ClientHelper
+  extends BeforeAndAfterAll
+     with AwaitHelper
+{ self: Suite =>
+
+  private val logger: Logger = LoggerFactory.getLogger(self.getClass)
+
+  val client = {
+    val c = new AmazonS3ScalaClient(new BasicAWSCredentials("FAKE_ACCESS_KEY", "FAKE_SECRET_KEY"))
+    c.client.setEndpoint("http://localhost:4000")
+    c.client.setS3ClientOptions(new S3ClientOptions().withPathStyleAccess(true))
+    c
+  }
+
+  val transferManager = new TransferManager(client.client)
+
+  val bucketNames: Seq[String]
+
+  override def beforeAll(): Unit = {
+    bucketNames foreach { name =>
+      logger.info(s"Creating bucket $name")
+      client.client.createBucket(name)
+    }
+
+    super.beforeAll()
+  }
+
+  override def afterAll(): Unit = {
+    try {
+      super.afterAll()
+      bucketNames foreach { name =>
+        logger.info(s"Deleting bucket $name")
+        client.client.deleteBucket(name)
+      }
+    } finally {
+      transferManager.shutdownNow()
+      client.shutdown()
+    }
+  }
+}

src/main/scala/s3/s3.scala

@@ -508,8 +508,8 @@ class AmazonS3ScalaClient(
   * interface, and [[FutureTransfer.listenFor]] adapts this interface to
   * Scala futures.
   *
-  * @see [[http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/transfer/TransferManager.html TransferManager]]
-  * @see [[http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/transfer/Transfer.html Transfer]]
+  * @see [[com.amazonaws.services.s3.transfer.TransferManager TransferManager]]
+  * @see [[com.amazonaws.services.s3.transfer.Transfer Transfer]]
   */
 object FutureTransfer {
 
@@ -518,7 +518,7 @@ object FutureTransfer {
   /**
     * Attach a listener to an S3 Transfer and return it as a Future.
     *
-    * This helper method attaches a progress listener to the given
+    * This helper method attaches a progress and state change listeners to the given
     * Transfer object. The returned future is completed with the
     * same transfer when the transfer is ‘done’ (canceled, completed,
     * or failed). The future will always been completed successfully
@@ -536,14 +536,33 @@ object FutureTransfer {
     * @param transfer
     *     an S3 Transfer to listen for progress.
     * @return the transfer in a future.
-    * @see [[http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/transfer/Transfer.html Transfer]]
-    * @see [[http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/event/ProgressListener.html ProgressListener]]
+    * @see [[com.amazonaws.services.s3.transfer.Transfer Transfer]]
+    * @see [[com.amazonaws.event.ProgressListener ProgressListener]]
     */
   def listenFor[T <: Transfer](transfer: T): Future[transfer.type] = {
+    import com.amazonaws.services.s3.transfer.internal.{ AbstractTransfer, TransferStateChangeListener }
     val transferDescription = transfer.getDescription
     def debugLog(eventType: String): Unit = {
       logger.debug(s"$eventType : $transferDescription")
     }
+    def logTransferState(state: Transfer.TransferState): Unit = {
+      if (logger.isDebugEnabled) {
+        state match {
+          case Transfer.TransferState.Waiting =>
+            debugLog("Waiting")
+          case Transfer.TransferState.InProgress =>
+            debugLog("InProgress")
+          case Transfer.TransferState.Completed =>
+            debugLog("Completed")
+          case Transfer.TransferState.Canceled =>
+            debugLog("Canceled")
+          case Transfer.TransferState.Failed =>
+            debugLog("Failed")
+          case _ =>
+            logger.warn(s"unrecognized transfer state for transfer $transferDescription")
+        }
+      }
+    }
     def logProgressEvent(progressEvent: ProgressEvent): Unit = {
       if (logger.isDebugEnabled) {
         progressEvent.getEventType match {
@@ -601,6 +620,35 @@ object FutureTransfer {
 
     val p = Promise[transfer.type]
 
+    if (transfer.isInstanceOf[AbstractTransfer]) {
+      /* Attach a state change listener to the transfer.
+       * At this point, the transfer is already in progress
+       * and may even have already completed. We will have
+       * missed any state change events that have already been
+       * fired, including the completion event!
+       */
+      transfer.asInstanceOf[AbstractTransfer].addStateChangeListener(new TransferStateChangeListener {
+        /* Note that the transferStateChanged will be called in the Java SDK’s
+         * special thread for callbacks, so any blocking calls here have
+         * the potential to induce deadlock.
+         */
+        override def transferStateChanged(t: Transfer, state: Transfer.TransferState): Unit = {
+          logTransferState(state)
+
+          if (state == Transfer.TransferState.Completed ||
+              state == Transfer.TransferState.Canceled ||
+              state == Transfer.TransferState.Failed) {
+            val success = p trySuccess transfer
+            if (logger.isDebugEnabled) {
+              if (success) {
+                logger.debug(s"promise successfully completed from transfer state change listener for $transferDescription")
+              }
+            }
+          }
+        }
+      })
+    }
+
     /* Attach a progress listener to the transfer.
      * At this point, the transfer is already in progress
      * and may even have already completed. We will have
@@ -624,8 +672,6 @@ object FutureTransfer {
           if (logger.isDebugEnabled) {
             if (success) {
               logger.debug(s"promise successfully completed from progress listener for $transferDescription")
-            } else {
-              logger.debug(s"promise was found to be already completed from progress listener for $transferDescription")
             }
           }
         }
@@ -639,9 +685,7 @@ object FutureTransfer {
       val success = p trySuccess transfer
       if (logger.isDebugEnabled) {
         if (success) {
-          logger.debug(s"promise successfully completed outside of progress listener for $transferDescription")
-        } else {
-          logger.debug(s"promise was found to be already completed outside of progress listener for $transferDescription")
+          logger.debug(s"promise successfully completed from outside of callbacks for $transferDescription")
         }
       }
     }

start-dynamodb-local.sh

@@ -4,7 +4,7 @@
 set -euo pipefail
 IFS=$'\n\t'
 
-WORKING_DIR="dynamodb"
+WORKING_DIR="dynamodb-local"
 
 mkdir -p $WORKING_DIR
 cd $WORKING_DIR
@@ -26,7 +26,7 @@ mkdir -p $LOG_DIR
 echo "DynamoDB Local output will save to ${WORKING_DIR}/${LOG_DIR}/"
 
 NOW=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
-nohup java -Djava.library.path=./DynamoDBLocal_lib -jar DynamoDBLocal.jar -port 8000 -inMemory 1>"${LOG_DIR}/${NOW}.out.log" 2>"${LOG_DIR}/${NOW}.err.log" &
+exec java -Djava.library.path=./DynamoDBLocal_lib -jar DynamoDBLocal.jar -port 8000 -inMemory 1>"${LOG_DIR}/${NOW}.out.log" 2>"${LOG_DIR}/${NOW}.err.log" &
 PID=$!
 
 echo "DynamoDB Local started with pid ${PID}"

start-fakes3.sh

@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+
+# http://redsymbol.net/articles/unofficial-bash-strict-mode/
+set -euo pipefail
+IFS=$'\n\t'
+
+bundle check
+
+WORKING_DIR="fakes3"
+DATA_DIR="${WORKING_DIR}/data"
+LOG_DIR="${WORKING_DIR}/logs"
+mkdir -p "$DATA_DIR" "$LOG_DIR"
+
+PORT="4000"
+NOW=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
+exec bundle exec fakes3 -r "$DATA_DIR" -p "$PORT" 1>"${LOG_DIR}/${NOW}.out.log" 2>"${LOG_DIR}/${NOW}.err.log" &
+PID=$!
+
+echo "fakes3 started with pid ${PID}"
+echo $PID >"${WORKING_DIR}/PID"
+
+echo "Pausing for 2 seconds..."
+sleep 2

stop-dynamodb-local.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-WORKING_DIR="dynamodb"
+WORKING_DIR="dynamodb-local"
 
 PID_FILE="${WORKING_DIR}/PID"
 
@@ -10,7 +10,7 @@ then
   exit 1
 else
   PID=$(cat $PID_FILE)
-  kill -s KILL $PID
+  kill -s TERM $PID
   rm $PID_FILE
   echo "DynamoDB Local stopped"
 fi