Merge pull request #3 from StreetContxt/tweak-stats

Tweaked stats

Author: anikifoss

build.sbt

@@ -1,6 +1,6 @@
 organization in ThisBuild := "com.contxt"
 scalaVersion in ThisBuild := "2.11.8"
-version in ThisBuild := "1.0.0-SNAPSHOT"
+version in ThisBuild := "1.0.1-SNAPSHOT"
 
 val slf4j = "org.slf4j" % "slf4j-api" % "1.7.21"
 val amazonKinesisProducer = "com.amazonaws" % "amazon-kinesis-producer" % "0.12.8"

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

@@ -8,6 +8,7 @@ import scala.util.control.NonFatal
 
 trait ProducerStats {
   def trackSend(streamId: StreamId, size: Int)(closure: => Future[UserRecordResult]): Future[UserRecordResult]
+  def reportInitialization(streamId: StreamId): Unit
   def reportShutdown(streamId: StreamId): Unit
 }
 
@@ -29,5 +30,6 @@ object ProducerStats {
 
 class NoopProducerStats extends ProducerStats {
   def trackSend(streamId: StreamId, size: Int)(closure: => Future[UserRecordResult]): Future[UserRecordResult] = closure
+  def reportInitialization(streamId: StreamId): Unit = {}
   def reportShutdown(streamId: StreamId): Unit = {}
 }

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

@@ -61,6 +61,8 @@ private[kinesis] class ScalaKinesisProducerImpl(
 ) extends ScalaKinesisProducer {
   import ScalaKinesisProducer.listenableToScalaFuture
 
+  stats.reportInitialization(streamId)
+
   def send(partitionKey: String, data: ByteBuffer, explicitHashKey: Option[String]): Future[UserRecordResult] = {
     stats.trackSend(streamId, data.remaining) {
       producer.addUserRecord(streamId.streamName, partitionKey, explicitHashKey.orNull, data).map { result =>