minor cleanup, still working in metrics

3 files changed, 55 insertions, 2 deletions

diff --git a/src/test/scala/kamon/instrumentation/MessageQueueInstrumentationSpec.scala b/src/test/scala/kamon/instrumentation/MessageQueueInstrumentationSpec.scala
new file mode 100644
index 00000000..d672b975
--- /dev/null
+++ b/src/test/scala/kamon/instrumentation/MessageQueueInstrumentationSpec.scala
@@ -0,0 +1,53 @@
+package kamon.instrumentation
+
+import org.scalatest.WordSpec
+import com.codahale.metrics.{ExponentiallyDecayingReservoir, Histogram}
+import java.util.concurrent.ConcurrentLinkedQueue
+import akka.dispatch.{UnboundedMessageQueueSemantics, QueueBasedMessageQueue, Envelope}
+import java.util.Queue
+import akka.actor.{ActorSystem, Actor}
+
+class MessageQueueInstrumentationSpec(val actorSystem: ActorSystem) extends WordSpec {
+  def this() = this(ActorSystem("MessageQueueInstrumentationSpec"))
+
+
+  "A MonitoredMessageQueue" should {
+    "update the related histogram when a message is enqueued" in {
+      new PopulatedMessageQueueFixture {
+
+        assert(histogram.getSnapshot.getMax   === 0)
+
+        for(i <- 1 to 3) { enqueueDummyMessage }
+
+        assert(histogram.getCount === 3)
+        assert(histogram.getSnapshot.getMax === 3)
+        assert(histogram.getSnapshot.getMin === 1)
+      }
+    }
+
+    "update the related histogram when a message is dequeued" in {
+      new PopulatedMessageQueueFixture {
+        for(i <- 1 to 3) { enqueueDummyMessage }
+        assert(histogram.getSnapshot.getMax   === 3)
+
+        messageQueue.dequeue()
+        messageQueue.dequeue()
+
+        assert(histogram.getCount === 5)
+        assert(histogram.getSnapshot.getMax === 3)
+        assert(histogram.getSnapshot.getMin === 1)
+      }
+    }
+  }
+
+  trait PopulatedMessageQueueFixture {
+
+    val histogram = new Histogram(new ExponentiallyDecayingReservoir())
+    val delegate = new ConcurrentLinkedQueue[Envelope]() with QueueBasedMessageQueue with UnboundedMessageQueueSemantics {
+      final def queue: Queue[Envelope] = this
+    }
+    val messageQueue = new MonitoredMessageQueue(delegate, histogram)
+
+    def enqueueDummyMessage = messageQueue.enqueue(Actor.noSender, Envelope("", Actor.noSender, actorSystem))
+  }
+}
diff --git a/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala b/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala
index 4fe9e617..359766b6 100644
--- a/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala
+++ b/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala
@@ -12,7 +12,7 @@ import java.util.concurrent.TimeUnit
 import akka.actor.ActorSystem
 
 
-class RunnableInstrumentationSpec extends WordSpec with MustMatchers with ScalaFutures with PatienceConfiguration with OptionValues {
+class RunnableInstrumentationSpec extends WordSpec with MustMatchers with ScalaFuturesSupport with PatienceConfiguration with OptionValues {
 
   "a instrumented runnable" when {
     "created in a thread that does have a TraceContext" must {
diff --git a/src/test/scala/kamon/instrumentation/ScalaFutures.scala b/src/test/scala/kamon/instrumentation/ScalaFuturesSupport.scala
index 169b709c..cc87a7c8 100644
--- a/src/test/scala/kamon/instrumentation/ScalaFutures.scala
+++ b/src/test/scala/kamon/instrumentation/ScalaFuturesSupport.scala
@@ -6,7 +6,7 @@ import scala.util.{Failure, Success}
 import org.scalatest.concurrent.Futures
 import java.util.concurrent.TimeUnit
 
-trait ScalaFutures extends Futures {
+trait ScalaFuturesSupport extends Futures {
   implicit def scalaFutureToFutureConcept[T](future: Future[T]): FutureConcept[T] = new FutureConcept[T] {
     def eitherValue: Option[Either[Throwable, T]] = {
       if(!future.isCompleted)