wip: playing with akka-less implementation of subscriptions

7 files changed, 73 insertions, 38 deletions

diff --git a/kamon-core/src/main/scala/kamon/metric/EntityRecorder.scala b/kamon-core/src/main/scala/kamon/metric/EntityRecorder.scala
index a94881d2..8ce37082 100644
--- a/kamon-core/src/main/scala/kamon/metric/EntityRecorder.scala
+++ b/kamon-core/src/main/scala/kamon/metric/EntityRecorder.scala
@@ -5,6 +5,8 @@ import java.time.Duration
 import kamon.metric.instrument._
 import kamon.util.MeasurementUnit
 
+import scala.collection.concurrent.TrieMap
+
 trait EntityRecorder {
   def histogram(name: String): Histogram
   def histogram(name: String, measurementUnit: MeasurementUnit, dynamicRange: DynamicRange): Histogram
@@ -13,11 +15,55 @@ trait EntityRecorder {
   def minMaxCounter(name: String, measurementUnit: MeasurementUnit, dynamicRange: DynamicRange, sampleFrequency: Duration): MinMaxCounter
 
   def gauge(name: String): Gauge
+  def gauge(name: String, measurementUnit: MeasurementUnit): Gauge
 
   def counter(name: String): Counter
+  def counter(name: String, measurementUnit: MeasurementUnit): Counter
+}
+
+trait EntitySnapshotProducer {
+  def snapshot(): EntitySnapshot
 }
 
 
-class EntityRecorderImpl {
 
+
+class DefaultEntityRecorder(entity: Entity, instrumentFactory: InstrumentFactory) extends EntityRecorder with EntitySnapshotProducer {
+  private val histograms = TrieMap.empty[String, Histogram with DistributionSnapshotInstrument]
+  private val minMaxCounters = TrieMap.empty[String, MinMaxCounter with DistributionSnapshotInstrument]
+  private val counters = TrieMap.empty[String, Counter with SingleValueSnapshotInstrument]
+  private val gauges = TrieMap.empty[String, Gauge with SingleValueSnapshotInstrument]
+
+  def histogram(name: String): Histogram =
+    histograms.atomicGetOrElseUpdate(name, instrumentFactory.buildHistogram(entity, name))
+
+  def histogram(name: String, measurementUnit: MeasurementUnit, dynamicRange: DynamicRange): Histogram =
+    histograms.atomicGetOrElseUpdate(name, instrumentFactory.buildHistogram(entity, name, dynamicRange, measurementUnit))
+
+  def minMaxCounter(name: String): MinMaxCounter =
+    minMaxCounters.atomicGetOrElseUpdate(name, instrumentFactory.buildMinMaxCounter(entity, name))
+
+  def minMaxCounter(name: String, measurementUnit: MeasurementUnit, dynamicRange: DynamicRange, sampleFrequency: Duration): MinMaxCounter =
+    minMaxCounters.atomicGetOrElseUpdate(name, instrumentFactory.buildMinMaxCounter(entity, name, dynamicRange, sampleFrequency, measurementUnit))
+
+  def gauge(name: String): Gauge =
+    gauges.atomicGetOrElseUpdate(name, instrumentFactory.buildGauge(entity, name))
+
+  def gauge(name: String, measurementUnit: MeasurementUnit): Gauge =
+    gauges.atomicGetOrElseUpdate(name, instrumentFactory.buildGauge(entity, name, measurementUnit))
+
+  def counter(name: String): Counter =
+    counters.atomicGetOrElseUpdate(name, instrumentFactory.buildCounter(entity, name))
+
+  def counter(name: String, measurementUnit: MeasurementUnit): Counter =
+    counters.atomicGetOrElseUpdate(name, instrumentFactory.buildCounter(entity, name, measurementUnit))
+
+  def snapshot(): EntitySnapshot =
+    new EntitySnapshot(
+      entity,
+      histograms = histograms.values.map(_.snapshot()).toSeq,
+      minMaxCounters = minMaxCounters.values.map(_.snapshot()).toSeq,
+      gauges = gauges.values.map(_.snapshot()).toSeq,
+      counters = counters.values.map(_.snapshot()).toSeq
+    )
 }
\ No newline at end of file
diff --git a/kamon-core/src/main/scala/kamon/metric/EntitySnapshot.scala b/kamon-core/src/main/scala/kamon/metric/EntitySnapshot.scala
index e51e80cc..a7db93eb 100644
--- a/kamon-core/src/main/scala/kamon/metric/EntitySnapshot.scala
+++ b/kamon-core/src/main/scala/kamon/metric/EntitySnapshot.scala
@@ -2,10 +2,10 @@ package kamon.metric
 
 import kamon.metric.instrument.{DistributionSnapshot, SingleValueSnapshot}
 
-trait EntitySnapshot {
-  def entity: Entity
-  def histograms: Seq[DistributionSnapshot]
-  def minMaxCounters: Seq[DistributionSnapshot]
-  def gauges: Seq[SingleValueSnapshot]
-  def counters: Seq[SingleValueSnapshot]
-}
\ No newline at end of file
+class EntitySnapshot(
+  val entity: Entity,
+  val histograms: Seq[DistributionSnapshot],
+  val minMaxCounters: Seq[DistributionSnapshot],
+  val gauges: Seq[SingleValueSnapshot],
+  val counters: Seq[SingleValueSnapshot]
+)
\ No newline at end of file
diff --git a/kamon-core/src/main/scala/kamon/metric/MetricsSubscriber.scala b/kamon-core/src/main/scala/kamon/metric/MetricsSubscriber.scala
deleted file mode 100644
index dbdfde9d..00000000
--- a/kamon-core/src/main/scala/kamon/metric/MetricsSubscriber.scala
+++ /dev/null
@@ -1,12 +0,0 @@
-package kamon.metric
-
-import com.typesafe.config.Config
-
-trait MetricsSubscriber {
-  def reconfigure(config: Config): Unit
-
-  def start(config: Config): Unit
-  def shutdown(): Unit
-
-  def processTick(snapshot: String)
-}
diff --git a/kamon-core/src/main/scala/kamon/metric/Metrics.scala b/kamon-core/src/main/scala/kamon/metric/RecorderRegistry.scala
index f312c5b7..99974032 100644
--- a/kamon-core/src/main/scala/kamon/metric/Metrics.scala
+++ b/kamon-core/src/main/scala/kamon/metric/RecorderRegistry.scala
@@ -1,10 +1,13 @@
 package kamon
 package metric
 
+import com.typesafe.config.Config
+import kamon.metric.instrument.InstrumentFactory
+
 import scala.collection.concurrent.TrieMap
 
 
-trait Metrics {
+trait RecorderRegistry {
   def getRecorder(entity: Entity): EntityRecorder
   def getRecorder(name: String, category: String, tags: Map[String, String]): EntityRecorder
 
@@ -12,11 +15,12 @@ trait Metrics {
   def removeRecorder(name: String, category: String, tags: Map[String, String]): Boolean
 }
 
-class MetricsImpl extends Metrics{
-  private val entities = TrieMap.empty[Entity, EntityRecorder]
+class RecorderRegistryImpl(config: Config) extends RecorderRegistry {
+  private val instrumentFactory = InstrumentFactory(config.getConfig("kamon.metric.instrument-factory"))
+  private val entities = TrieMap.empty[Entity, EntityRecorder with EntitySnapshotProducer]
 
   override def getRecorder(entity: Entity): EntityRecorder = {
-    ???
+    entities.atomicGetOrElseUpdate(entity, new DefaultEntityRecorder(entity, instrumentFactory))
   }
 
   override def getRecorder(name: String, category: String, tags: Map[String, String]): EntityRecorder = ???
@@ -24,6 +28,10 @@ class MetricsImpl extends Metrics{
   override def removeRecorder(entity: Entity): Boolean = ???
 
   override def removeRecorder(name: String, category: String, tags: Map[String, String]): Boolean = ???
+
+  private[kamon] def snapshot(): Seq[EntitySnapshot] = {
+    entities.values.map(_.snapshot()).toSeq
+  }
 }
 
 
@@ -32,3 +40,5 @@ class MetricsImpl extends Metrics{
 
 
 
+
+
diff --git a/kamon-core/src/main/scala/kamon/metric/TickSnapshot.scala b/kamon-core/src/main/scala/kamon/metric/TickSnapshot.scala
index 4248180c..f4578965 100644
--- a/kamon-core/src/main/scala/kamon/metric/TickSnapshot.scala
+++ b/kamon-core/src/main/scala/kamon/metric/TickSnapshot.scala
@@ -2,16 +2,8 @@ package kamon.metric
 
 import java.time.Instant
 
-
-trait TickSnapshot {
-  def interval: Interval
-  def entities: Seq[EntitySnapshot]
-}
-
-trait Interval {
-  def from: Instant
-  def to: Instant
-}
+case class TickSnapshot(interval: Interval, entities: Seq[EntitySnapshot])
+case class Interval(from: Instant, to: Instant)
 
 
 
diff --git a/kamon-core/src/main/scala/kamon/metric/instrument/HistogramExtension.scala b/kamon-core/src/main/scala/kamon/metric/instrument/HistogramExtension.scala
index ebb82040..dc3cad08 100644
--- a/kamon-core/src/main/scala/kamon/metric/instrument/HistogramExtension.scala
+++ b/kamon-core/src/main/scala/kamon/metric/instrument/HistogramExtension.scala
@@ -6,8 +6,7 @@ import java.util.concurrent.atomic.AtomicLongArray
 import kamon.metric.instrument.DynamicRange
 
 /**
-  * This class exposes package-private members of the [[AtomicHistogram]] class that are required to properly generate
-  * snapshots of our HdrHistogram implementation.
+  * Exposes package-private members of [[org.HdrHistogram.AtomicHistogram]].
   */
 abstract class AtomicHistogramExtension(dr: DynamicRange)
     extends AtomicHistogram(dr.lowestDiscernibleValue, dr.highestTrackableValue, dr.significantValueDigits) {
@@ -22,7 +21,7 @@ abstract class AtomicHistogramExtension(dr: DynamicRange)
 }
 
 /**
-  * Exposes the package-private members of [[ZigZagEncoding]].
+  * Exposes the package-private members of [[org.HdrHistogram.ZigZagEncoding]].
   */
 object ZigZag {
   def putLong(buffer: ByteBuffer, value: Long): Unit =
diff --git a/kamon-core/src/main/scala/kamon/metric/instrument/InstrumentFactory.scala b/kamon-core/src/main/scala/kamon/metric/instrument/InstrumentFactory.scala
index fb6dfe27..4f0502f0 100644
--- a/kamon-core/src/main/scala/kamon/metric/instrument/InstrumentFactory.scala
+++ b/kamon-core/src/main/scala/kamon/metric/instrument/InstrumentFactory.scala
@@ -27,7 +27,7 @@ private[metric] class InstrumentFactory private (
   }
 
   def buildMinMaxCounter(entity: Entity, name: String, dynamicRange: DynamicRange = defaultMMCounterDynamicRange,
-      sampleInterval: Duration = defaultMMCounterSampleRate, measurementUnit: MeasurementUnit = MeasurementUnit.none): MinMaxCounter = {
+      sampleInterval: Duration = defaultMMCounterSampleRate, measurementUnit: MeasurementUnit = MeasurementUnit.none): MinMaxCounter with DistributionSnapshotInstrument = {
 
     val underlyingHistogram = buildHistogram(entity, name, dynamicRange, measurementUnit)
     new PaddedMinMaxCounter(