1 files changed, 43 insertions, 0 deletions
diff --git a/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala b/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala
index 6c79806d..1f3564d3 100644
--- a/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala
+++ b/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala
@@ -8,6 +8,8 @@ import kamon.metric.{DispatcherMetricCollector, Histogram, MetricDirectory, Exec
 import akka.dispatch.{MonitorableThreadFactory, ExecutorServiceFactory}
 import com.typesafe.config.Config
 import kamon.Kamon
+import scala.concurrent.forkjoin.ForkJoinPool
+import akka.dispatch.ForkJoinExecutorConfigurator.AkkaForkJoinPool
 
 
 @Aspect
@@ -23,6 +25,47 @@ class ActorSystemInstrumentation {
   }
 }
 
+@Aspect("perthis(forkJoinPoolInstantiation(int, scala.concurrent.forkjoin.ForkJoinPool.ForkJoinWorkerThreadFactory, java.lang.Thread.UncaughtExceptionHandler))")
+class ForkJoinPoolInstrumentation {
+  var activeThreadsHistogram: Histogram = _
+
+  @Pointcut("execution(akka.dispatch.ForkJoinExecutorConfigurator.AkkaForkJoinPool.new(..)) && args(parallelism, threadFactory, exceptionHandler)")
+  def forkJoinPoolInstantiation(parallelism: Int, threadFactory: ForkJoinPool.ForkJoinWorkerThreadFactory, exceptionHandler: Thread.UncaughtExceptionHandler) = {}
+
+  @After("forkJoinPoolInstantiation(parallelism, threadFactory, exceptionHandler)")
+  def initializeMetrics(parallelism: Int, threadFactory: ForkJoinPool.ForkJoinWorkerThreadFactory, exceptionHandler: Thread.UncaughtExceptionHandler): Unit = {
+    val (actorSystemName, dispatcherName) = threadFactory match {
+      case mtf: MonitorableThreadFactory => splitName(mtf.name, Kamon.Metric.actorSystemNames)
+      case _ => ("Unknown", "Unknown")
+    }
+
+    val metrics = Kamon.Metric.actorSystem(actorSystemName).get.registerDispatcher(dispatcherName)
+    for(m <- metrics) {
+      activeThreadsHistogram = m.activeThreadCount
+      println(s"Registered $dispatcherName for actor system $actorSystemName")
+    }
+  }
+
+  def splitName(threadFactoryName: String, knownActorSystems: List[String]): (String, String) = {
+    knownActorSystems.find(threadFactoryName.startsWith(_)).map(asName => (asName, threadFactoryName.substring(asName.length+1))).getOrElse(("Unkown", "Unkown"))
+  }
+
+
+
+
+  @Pointcut("execution(* scala.concurrent.forkjoin.ForkJoinPool.scan(..)) && this(fjp)")
+  def forkJoinScan(fjp: AkkaForkJoinPool): Unit = {}
+
+  @After("forkJoinScan(fjp)")
+  def updateMetrics(fjp: AkkaForkJoinPool): Unit = {
+    activeThreadsHistogram.update(fjp.getActiveThreadCount)
+    println("UPDATED THE COUNT TWOOOO!!!")
+  }
+
+
+
+}
+