Merge pull request #9 from StreetContxt/readme-how-to

Readme Usage Section (and minor code tweaks)

Author: anikifoss

README.md

@@ -5,6 +5,45 @@ The main benefit of this library is working with Scala-native Futures when
 interacting with KPL.
 
 
+## Installation
+
+```
+resolvers in ThisBuild += Resolver.bintrayRepo("streetcontxt", "maven")
+libraryDependencies += "com.contxt" %% "kpl-scala" % "1.0.4"
+```
+
+
+## Usage
+
+Here is a simple app that initializes the Kinesis producer and sends a string message.
+
+```
+import com.amazonaws.auth.DefaultAWSCredentialsProviderChain
+import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration
+import com.contxt.kinesis.ScalaKinesisProducer
+import java.nio.ByteBuffer
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+object Main {
+  def main(args: Array[String]): Unit = {
+    val producerConfig = new KinesisProducerConfiguration()
+      .setCredentialsProvider(new DefaultAWSCredentialsProviderChain)
+      .setRegion("us-east-1")
+
+    val producer = ScalaKinesisProducer("myStream", producerConfig)
+
+    val sendFuture = producer.send(
+      partitionKey = "myKey",
+      data = ByteBuffer.wrap("myMessage".getBytes("UTF-8"))
+    )
+    Await.result(sendFuture, 10.seconds)
+    Await.result(producer.shutdown(), Duration.Inf)
+  }
+}
+```
+
+
 ## Amazon Licensing Restrictions
 **KPL license is not compatible with open source licenses!** See
 [this discussion](https://issues.apache.org/jira/browse/LEGAL-198) for more details.

src/main/scala/com/contxt/kinesis/ScalaKinesisProducer.scala

@@ -71,13 +71,17 @@ private[kinesis] class ScalaKinesisProducerImpl(
     }
   }
 
-  def shutdown(): Future[Unit] = {
+  def shutdown(): Future[Unit] = shutdownOnce
+
+  private lazy val shutdownOnce: Future[Unit] = {
     val allFlushedFuture = flushAll()
+    val shutdownPromise = Promise[Unit]
     allFlushedFuture.onComplete { _ =>
-      producer.destroy()
-      stats.reportShutdown(streamId)
+      shutdownPromise.completeWith(destroyProducer())
     }
-    allFlushedFuture
+    val combinedFuture = allFlushedFuture.zip(shutdownPromise.future).map(_ => ())
+    combinedFuture.onComplete(_ => stats.reportShutdown(streamId))
+    combinedFuture
   }
 
   private def throwSendFailedException(result: UserRecordResult): Nothing = {
@@ -95,4 +99,12 @@ private[kinesis] class ScalaKinesisProducerImpl(
       }
     }
   }
+
+  private def destroyProducer(): Future[Unit] = {
+    Future {
+      blocking {
+        producer.destroy()
+      }
+    }
+  }
 }