create new DifferentialSource utility

2 files changed, 64 insertions, 0 deletions

diff --git a/kamon-core-tests/src/test/scala/kamon/util/DifferentialSourceSpec.scala b/kamon-core-tests/src/test/scala/kamon/util/DifferentialSourceSpec.scala
new file mode 100644
index 00000000..0a2f83e5
--- /dev/null
+++ b/kamon-core-tests/src/test/scala/kamon/util/DifferentialSourceSpec.scala
@@ -0,0 +1,40 @@
+package kamon.util
+
+import org.scalatest.{Matchers, WordSpec}
+
+class DifferentialSourceSpec extends WordSpec with Matchers {
+
+  "a Differential Source" should {
+    "get the difference between the last two observations" in {
+      val source = sourceOf(0, 0, 1, 1, 2, 3, 4, 6, 8, 10, 12, 16, 18)
+      val expectedDiffs = Seq(0, 1, 0, 1, 1, 1, 2, 2, 2, 2, 4, 2)
+
+      values(expectedDiffs.length, source) should contain theSameElementsInOrderAs(expectedDiffs)
+    }
+
+    "ignore decrements in observations" in {
+      val source = sourceOf(10, 10, 5, 5, 10, 10)
+      val expectedDiffs = Seq(0, 0, 0, 5, 0)
+
+      values(expectedDiffs.length, source) should contain theSameElementsInOrderAs(expectedDiffs)
+    }
+  }
+
+  def sourceOf(numbers: Long*): DifferentialSource = DifferentialSource(new (() => Long) {
+    var remaining = numbers.toList
+
+    override def apply(): Long = {
+      if(remaining.isEmpty) 0 else {
+        val head = remaining.head
+        remaining = remaining.tail
+        head
+      }
+    }
+  })
+
+  def values(count: Int, source: DifferentialSource): Seq[Long] = {
+    for(_ <- 1 to count) yield source.get()
+  }
+}
+
+
diff --git a/kamon-core/src/main/scala/kamon/util/DifferentialSource.scala b/kamon-core/src/main/scala/kamon/util/DifferentialSource.scala
new file mode 100644
index 00000000..f2c621b0
--- /dev/null
+++ b/kamon-core/src/main/scala/kamon/util/DifferentialSource.scala
@@ -0,0 +1,24 @@
+package kamon.util
+
+/**
+  * Keeps track of the values produced by the source and produce the difference between the last two observed values
+  * when calling get. This class assumes the source increases monotonically and any produced value that violates this
+  * assumption will be dropped.
+  *
+  */
+class DifferentialSource(source: () => Long) {
+  private var previousValue = source()
+
+  def get(): Long = synchronized {
+    val currentValue = source()
+    val diff = currentValue - previousValue
+    previousValue = currentValue
+
+    if(diff < 0) 0 else diff
+  }
+}
+
+object DifferentialSource {
+  def apply(source: () => Long): DifferentialSource =
+    new DifferentialSource(source)
+}