upgrading to akka 2.2

5 files changed, 236 insertions, 0 deletions

diff --git a/kamon-core/src/test/scala/akka/instrumentation/ActorInstrumentationSpec.scala b/kamon-core/src/test/scala/akka/instrumentation/ActorInstrumentationSpec.scala
new file mode 100644
index 00000000..0026d953
--- /dev/null
+++ b/kamon-core/src/test/scala/akka/instrumentation/ActorInstrumentationSpec.scala
@@ -0,0 +1,45 @@
+package akka.instrumentation
+
+import org.scalatest.{WordSpecLike, Matchers}
+import akka.actor.{Actor, Props, ActorSystem}
+
+import akka.testkit.{ImplicitSender, TestKit}
+import kamon.{TraceContext, Kamon}
+
+
+class ActorInstrumentationSpec extends TestKit(ActorSystem("ActorInstrumentationSpec")) with WordSpecLike with Matchers with ImplicitSender {
+
+  "an instrumented actor ref" when {
+    "used inside the context of a transaction" should {
+      "propagate the trace context using bang" in new TraceContextEchoFixture {
+        echo ! "test"
+
+        expectMsg(Some(testTraceContext))
+      }
+
+      "propagate the trace context using tell" in {
+
+      }
+
+      "propagate the trace context using ask" in {
+
+      }
+    }
+  }
+
+  trait TraceContextEchoFixture {
+    val testTraceContext = Kamon.newTraceContext()
+    val echo = system.actorOf(Props[TraceContextEcho])
+
+    Kamon.set(testTraceContext)
+  }
+
+}
+
+class TraceContextEcho extends Actor {
+  def receive = {
+    case msg ⇒ sender ! Kamon.context()
+  }
+}
+
+
diff --git a/kamon-core/src/test/scala/kamon/instrumentation/ActorSystemInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/ActorSystemInstrumentationSpec.scala
new file mode 100644
index 00000000..1eab6355
--- /dev/null
+++ b/kamon-core/src/test/scala/kamon/instrumentation/ActorSystemInstrumentationSpec.scala
@@ -0,0 +1,22 @@
+package kamon.instrumentation
+
+import org.scalatest.{Matchers, WordSpec}
+import akka.actor.ActorSystem
+import kamon.Kamon
+
+class ActorSystemInstrumentationSpec extends WordSpec with Matchers {
+
+  // TODO: Selection filters to exclude unwanted actor systems. Read from configuration.
+
+  "the actor system instrumentation" should {
+    "register all actor systems created" in {
+      val as1 = ActorSystem("as1")
+      val as2 = ActorSystem("as2")
+
+
+      Kamon.Metric.actorSystem("as1") should not be (None)
+      Kamon.Metric.actorSystem("as2") should not be (None)
+      Kamon.Metric.actorSystem("unknown") should be (None)
+    }
+  }
+}
diff --git a/kamon-core/src/test/scala/kamon/instrumentation/DispatcherInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/DispatcherInstrumentationSpec.scala
new file mode 100644
index 00000000..89ef61f3
--- /dev/null
+++ b/kamon-core/src/test/scala/kamon/instrumentation/DispatcherInstrumentationSpec.scala
@@ -0,0 +1,34 @@
+package kamon.instrumentation
+
+import org.scalatest.{Matchers, WordSpec}
+import akka.actor.{Actor, Props, ActorSystem}
+import kamon.metric.MetricDirectory
+import kamon.Kamon
+
+class DispatcherInstrumentationSpec extends WordSpec with Matchers{
+
+
+  "the dispatcher instrumentation" should {
+    "instrument a dispatcher that belongs to a non-filtered actor system" in new SingleDispatcherActorSystem {
+      val x = Kamon.Metric.actorSystem("single-dispatcher").get.dispatchers
+      (1 to 10).foreach(actor ! _)
+
+      val active = x.get("akka.actor.default-dispatcher").get.activeThreadCount.snapshot
+      println("Active max: "+active.max)
+      println("Active min: "+active.min)
+
+    }
+  }
+
+
+  trait SingleDispatcherActorSystem {
+    val actorSystem = ActorSystem("single-dispatcher")
+    val actor = actorSystem.actorOf(Props(new Actor {
+      def receive = {
+        case a => sender ! a;
+      }
+    }))
+
+  }
+}
+
diff --git a/kamon-core/src/test/scala/kamon/instrumentation/MessageQueueInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/MessageQueueInstrumentationSpec.scala
new file mode 100644
index 00000000..cc55ec92
--- /dev/null
+++ b/kamon-core/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/kamon-core/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala
new file mode 100644
index 00000000..de65aaca
--- /dev/null
+++ b/kamon-core/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala
@@ -0,0 +1,82 @@
+package kamon.instrumentation
+
+import scala.concurrent.{Await, Promise, Future}
+import org.scalatest.{Matchers, OptionValues, WordSpec}
+import org.scalatest.concurrent.{ScalaFutures, PatienceConfiguration}
+import kamon.{Kamon, TraceContext}
+import java.util.UUID
+import scala.util.Success
+import scala.concurrent.duration._
+import java.util.concurrent.TimeUnit
+import akka.actor.ActorSystem
+
+
+class RunnableInstrumentationSpec extends WordSpec with Matchers with ScalaFutures with PatienceConfiguration with OptionValues {
+
+  "a instrumented runnable" when {
+    "created in a thread that does have a TraceContext" must {
+      "preserve the TraceContext" which {
+        "should be available during the run method execution" in { new FutureWithContextFixture {
+
+            whenReady(futureWithContext) { result =>
+              result.value should equal(testContext)
+            }
+        }}
+
+        "should be available during the execution of onComplete callbacks" in { new FutureWithContextFixture {
+            val onCompleteContext = Promise[TraceContext]()
+
+            futureWithContext.onComplete({
+              case _ => onCompleteContext.complete(Success(Kamon.context.get))
+            })
+
+            whenReady(onCompleteContext.future) { result =>
+              result should equal(testContext)
+            }
+        }}
+      }
+    }
+
+    "created in a thread that doest have a TraceContext" must {
+      "not capture any TraceContext for the body execution" in { new FutureWithoutContextFixture{
+
+          whenReady(futureWithoutContext) { result =>
+            result should equal(None)
+          }
+      }}
+
+      "not make any TraceContext available during the onComplete callback" in { new FutureWithoutContextFixture {
+        val onCompleteContext = Promise[Option[TraceContext]]()
+
+        futureWithoutContext.onComplete({
+          case _ => onCompleteContext.complete(Success(Kamon.context))
+        })
+
+        whenReady(onCompleteContext.future) { result =>
+          result should equal(None)
+        }
+      }}
+    }
+  }
+
+
+  /**
+   *  We are using Futures for the test since they exercise Runnables in the back and also resemble the real use case we have.
+   */
+  implicit val testActorSystem = ActorSystem("test-actorsystem")
+  implicit val execContext = testActorSystem.dispatcher
+
+  class FutureWithContextFixture {
+    val testContext = TraceContext()
+    Kamon.set(testContext)
+
+    val futureWithContext = Future { Kamon.context}
+  }
+
+  trait FutureWithoutContextFixture {
+    Kamon.clear // Make sure no TraceContext is available
+    val futureWithoutContext = Future { Kamon.context }
+  }
+}
+
+