Merge pull request #103 from guardian/support-logging-on-changed-values

Support logging on updated values

Author: rtyley

core/src/main/scala/com/gu/etagcaching/Loading.scala

@@ -1,5 +1,6 @@
 package com.gu.etagcaching
 
+import com.gu.etagcaching.Loading.{OnUpdate, Update}
 import com.gu.etagcaching.fetching.{ETaggedData, Fetching, MissingOrETagged}
 
 import scala.concurrent.{ExecutionContext, Future}
@@ -25,6 +26,11 @@ trait Loading[K, V] {
    * any new data, and can just reuse our old data, saving us CPU time and network bandwidth.
    */
   def fetchThenParseIfNecessary(k: K, oldV: ETaggedData[V])(implicit ec: ExecutionContext): Future[MissingOrETagged[V]]
+
+  /**
+   * Add a handler for doing side-effectful logging of updates.
+   */
+  def onUpdate(handler: Update[K,V] => Unit): Loading[K, V] = OnUpdate(this)(handler)
 }
 
 object Loading {
@@ -38,5 +44,32 @@ object Loading {
         case Some(freshResponse) => freshResponse.map(parse)
       }
   }
+
+  /**
+   * Represents an update event for a given key.
+   */
+  case class Update[K, V](key: K, oldV: Option[V], newV: Option[V])
+
+  /**
+   * Wrapper round an underlying instance of Loading which adds handler for doing side-effectful logging of updates.
+   */
+  case class OnUpdate[K, V](underlying: Loading[K, V])(handler: Update[K,V] => Unit)
+    extends Loading[K, V] {
+
+    override def fetchAndParse(key: K)(implicit ec: ExecutionContext): Future[MissingOrETagged[V]] =
+      handle(key, None, underlying.fetchAndParse(key))
+
+    override def fetchThenParseIfNecessary(key: K, oldV: ETaggedData[V])(implicit ec: ExecutionContext): Future[MissingOrETagged[V]] =
+      handle(key, oldV.toOption, underlying.fetchThenParseIfNecessary(key, oldV))
+
+    private def handle(key: K, oldV: Option[V], fut: Future[MissingOrETagged[V]])(implicit ec: ExecutionContext): Future[MissingOrETagged[V]] = {
+      for {
+        wrappedNewV <- fut
+      } {
+        handler(Update(key, oldV, wrappedNewV.toOption))
+      }
+      fut
+    }
+  }
 }
 

core/src/test/scala/com/gu/etagcaching/FreshnessPolicyTest.scala

@@ -2,18 +2,22 @@ package com.gu.etagcaching
 
 import com.github.blemale.scaffeine.{AsyncLoadingCache, Scaffeine}
 import com.gu.etagcaching.FreshnessPolicy.{AlwaysWaitForRefreshedValue, TolerateOldValueWhileRefreshing}
+import com.gu.etagcaching.fetching.{ETaggedData, Fetching, MissingOrETagged}
+import org.scalatest.OptionValues
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.concurrent.TimeLimits.failAfter
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.time.SpanSugar._
 
-import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.Future
+import java.util.concurrent.atomic.AtomicInteger
+import scala.concurrent.{ExecutionContext, Future}
 
-class FreshnessPolicyTest extends AnyFlatSpec with Matchers with ScalaFutures {
+class FreshnessPolicyTest extends AnyFlatSpec with Matchers with ScalaFutures with OptionValues {
 
   case class DemoCache(policy: FreshnessPolicy) {
+    import scala.concurrent.ExecutionContext.Implicits.global
+
     lazy val exampleCache: AsyncLoadingCache[String, Int] = {
       def simulateWork(task: String, key: String, millis: Long): Unit = {
         println(s"$task begin: $key")
@@ -54,4 +58,24 @@ class FreshnessPolicyTest extends AnyFlatSpec with Matchers with ScalaFutures {
       demo.read() shouldBe 0
     }
   }
+
+  it should "mean that ETagCache won't make additional requests if a value is cached locally" in {
+    val fetching: Fetching[String, Int] = TestFetching.withIncrementingValues
+
+    val eTagCache = new ETagCache[String, Int](
+      fetching.thenParsing(identity),
+      TolerateOldValueWhileRefreshing,
+      _.maximumSize(1).expireAfterWrite(100.millis)
+    )(ExecutionContext.Implicits.global)
+
+    eTagCache.get("KEY").futureValue.value shouldBe 0
+    eTagCache.get("KEY").futureValue.value shouldBe 0
+    eTagCache.get("KEY").futureValue.value shouldBe 0
+    Thread.sleep(200)
+    eTagCache.get("KEY").futureValue.value shouldBe 1
+    Thread.sleep(30)
+    eTagCache.get("KEY").futureValue.value shouldBe 1
+    Thread.sleep(30)
+    eTagCache.get("KEY").futureValue.value shouldBe 1
+  }
 }

core/src/test/scala/com/gu/etagcaching/LoadingTest.scala

@@ -0,0 +1,45 @@
+package com.gu.etagcaching
+
+import com.gu.etagcaching.FreshnessPolicy.TolerateOldValueWhileRefreshing
+import com.gu.etagcaching.Loading.Update
+import com.gu.etagcaching.fetching.Fetching
+import org.scalatest.OptionValues
+import org.scalatest.concurrent.{Eventually, ScalaFutures}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.collection.mutable
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration._
+
+class LoadingTest extends AnyFlatSpec with Matchers with ScalaFutures with OptionValues with Eventually {
+  "onUpdate" should "give callbacks that allow logging updates" in {
+    val updates: mutable.Buffer[Update[String, Int]] = mutable.Buffer.empty
+
+    val fetching: Fetching[String, Int] = TestFetching.withIncrementingValues
+
+    val cache = new ETagCache(
+      fetching.thenParsing(identity).onUpdate { update =>
+        updates.append(update)
+      },
+      TolerateOldValueWhileRefreshing,
+      _.maximumSize(1).refreshAfterWrite(100.millis)
+    )
+
+    val expectedUpdates = Seq(
+      Update("key", None, Some(0)),
+      Update("key", Some(0), Some(1))
+    )
+
+    cache.get("key").futureValue shouldBe Some(0)
+    updates shouldBe expectedUpdates.take(1)
+
+    Thread.sleep(105)
+
+    eventually { cache.get("key").futureValue shouldBe Some(1) }
+    updates.toSeq shouldBe expectedUpdates
+
+    Thread.sleep(105)
+    updates.toSeq shouldBe expectedUpdates // No updates if we're not requesting the key from the cache
+  }
+}

core/src/test/scala/com/gu/etagcaching/TestFetching.scala

@@ -0,0 +1,20 @@
+package com.gu.etagcaching
+
+import com.gu.etagcaching.fetching.{ETaggedData, Fetching, MissingOrETagged}
+
+import java.util.concurrent.atomic.AtomicInteger
+import scala.concurrent.{ExecutionContext, Future}
+
+object TestFetching {
+
+  def withIncrementingValues: Fetching[String, Int] = new Fetching[String, Int] {
+    val counter = new AtomicInteger()
+
+    override def fetch(key: String)(implicit ec: ExecutionContext): Future[MissingOrETagged[Int]] = {
+      val count = counter.getAndIncrement()
+      Future.successful(ETaggedData(count.toString, count))
+    }
+    override def fetchOnlyIfETagChanged(key: String, eTag: String)(implicit ec: ExecutionContext): Future[Option[MissingOrETagged[Int]]] =
+      fetch(key)(ec).map(Some(_))
+  }
+}