Merge pull request #4 from redbubble/slight-cache-cleanup

Moved common cache mechanics into its own class

Author: kenvandenenden

src/main/scala/com/redbubble/util/cache/Caching.scala

@@ -0,0 +1,86 @@
+package com.redbubble.util.cache
+
+import java.util.concurrent.{Executor, TimeUnit}
+
+import com.redbubble.util.async.syntax._
+import com.twitter.util.{Duration, Future}
+
+import scala.concurrent.ExecutionContext.fromExecutor
+import scala.concurrent.duration.{Duration => ScalaDuration}
+import scalacache.serialization.Codec
+import scalacache.{Flags, ScalaCache}
+
+/**
+  * Common implementations of the underlying caching mechanics, transforming of isomorphic types, etc.
+  */
+private[cache] object Caching {
+  private val flags = Flags(readsEnabled = true, writesEnabled = true)
+
+  /**
+    * Cache the result of executing `f` using the given `key`.
+    *
+    * @param cache The underlying cache.
+    * @param ttl   The TTL of the item to be cached.
+    * @param key   The key to use to cache the value.
+    * @param codec The codec to use to serialise/encide the value into the cache.
+    * @param f     The function to run, whose result will be cached.
+    * @param ex    The executor to use to run any asynchronous operations.
+    * @tparam V    The type of the value to be cached.
+    * @tparam Repr The type of the serialised/encoded form of the value, e.g. `InMemoryRepr` or `Array[Byte]`.
+    * @return A Twitter `Future` with the result of `f`.
+    */
+  def caching[V, Repr](cache: ScalaCache[Repr], ttl: Duration, key: CacheKey, codec: Codec[V, Repr])(f: => Future[V])
+      (implicit ex: Executor): Future[V] = {
+    val ec = toEc(ex)
+    val result = scalacache.cachingWithTTL(key)(toScalaTtl(ttl))(f.asScala)(cache, flags, ec, codec)
+    result.asTwitter(ec)
+  }
+
+  /**
+    * Manually put a value into the cache. This function is side-effecting. Prefer `caching` instead of this function.
+    *
+    * @param cache The underlying cache.
+    * @param ttl   The TTL of the item to be cached.
+    * @param key   The key to use to cache the value.
+    * @param codec The codec to use to serialise/encide the value into the cache.
+    * @param value The value to cache.
+    * @param ex    The executor to use to run any asynchronous operations.
+    * @tparam V    The type of the value to be cached.
+    * @tparam Repr The type of the serialised/encoded form of the value, e.g. `InMemoryRepr` or `Array[Byte]`.
+    * @return A Twitter `Future` containing no result (you can use this to know when the value has been cached).
+    */
+  def put[V, Repr](cache: ScalaCache[Repr], ttl: Duration, key: CacheKey, codec: Codec[V, Repr], value: V)
+      (implicit ex: Executor): Future[Unit] = {
+    val ec = toEc(ex)
+    val result = scalacache.put(key)(value, Some(toScalaTtl(ttl)))(cache, flags, codec)
+    result.asTwitter(ec)
+  }
+
+  /**
+    * Manually get a value from the cache. Prefer `caching` instead of this function.
+    *
+    * @param cache The underlying cache.
+    * @param key   The key to use to cache the value.
+    * @param codec The codec to use to serialise/encide the value into the cache.
+    * @param ex    The executor to use to run any asynchronous operations.
+    * @tparam V    The type of the value to be cached.
+    * @tparam Repr The type of the serialised/encoded form of the value, e.g. `InMemoryRepr` or `Array[Byte]`.
+    * @return A Twitter `Future` the value from the cache.
+    */
+  def get[V, Repr](cache: ScalaCache[Repr], key: CacheKey, codec: Codec[V, Repr])
+      (implicit ex: Executor): Future[Option[V]] = {
+    val ec = toEc(ex)
+    val result = scalacache.get(key)(cache, flags, codec)
+    result.asTwitter(ec)
+  }
+
+  /**
+    * Flush (clear) the cache of all entries.
+    */
+  def flush[Repr](cache: ScalaCache[Repr])(implicit ex: Executor): Future[Unit] =
+    cache.cache.removeAll().asTwitter(toEc(ex))
+
+  private def toEc(ex: Executor) = fromExecutor(ex)
+
+  private def toScalaTtl(ttl: Duration): ScalaDuration = ScalaDuration(ttl.inNanoseconds, TimeUnit.NANOSECONDS)
+}

src/main/scala/com/redbubble/util/cache/memory/InMemorySimpleCache.scala

@@ -3,15 +3,12 @@ package com.redbubble.util.cache.memory
 import java.util.concurrent.{Executor, TimeUnit}
 
 import com.github.benmanes.caffeine.cache.Caffeine
-import com.redbubble.util.async.syntax._
-import com.redbubble.util.cache.{CacheKey, SimpleCache}
+import com.redbubble.util.cache.{CacheKey, Caching, SimpleCache}
 import com.redbubble.util.metrics.StatsReceiver
 import com.twitter.util.{Duration, Future}
 
-import scala.concurrent.ExecutionContext.fromExecutor
-import scala.concurrent.duration.{Duration => ScalaDuration}
 import scalacache.serialization.{Codec, InMemoryRepr}
-import scalacache.{CacheConfig, Flags, ScalaCache}
+import scalacache.{CacheConfig, ScalaCache}
 
 /**
   * An in-memory cache, with metrics tracking.
@@ -24,33 +21,30 @@ import scalacache.{CacheConfig, Flags, ScalaCache}
   */
 private[cache] final class InMemorySimpleCache(name: String, maxSize: Long, ttl: Duration)
     (implicit ex: Executor, statsReceiver: StatsReceiver) extends SimpleCache {
-  private val flags = Flags(readsEnabled = true, writesEnabled = true)
-  private val scalaTtl = ScalaDuration(ttl.inNanoseconds, TimeUnit.NANOSECONDS)
-  private val cache = createCache(name, maxSize, scalaTtl, ex, statsReceiver)
-  private val ec = fromExecutor(ex)
+  private val cache = createCache(name, maxSize, ex, statsReceiver)
 
   override def caching[V](key: CacheKey)(f: => Future[V]): Future[V] = {
-    val noOpCodec = Codec.anyToNoSerialization[V]
-    scalacache.cachingWithTTL[V, InMemoryRepr](key)(scalaTtl)(f.asScala)(cache, flags, ec, noOpCodec).asTwitter(ec)
+    val codec = Codec.anyToNoSerialization[V]
+    Caching.caching(cache, ttl, key, codec)(f)(ex)
   }
 
   override def put[V, Repr](key: CacheKey, value: V): Future[Unit] = {
-    val noOpCodec = Codec.anyToNoSerialization[V]
-    scalacache.put[V, InMemoryRepr](key)(value, Some(scalaTtl))(cache, flags, noOpCodec).asTwitter(ec)
+    val codec = Codec.anyToNoSerialization[V]
+    Caching.put(cache, ttl, key, codec, value)(ex)
   }
 
   override def get[V](key: CacheKey): Future[Option[V]] = {
-    val noOpCodec = Codec.anyToNoSerialization[V]
-    scalacache.get[V, InMemoryRepr](key)(cache, flags, noOpCodec).asTwitter(ec)
+    val codec = Codec.anyToNoSerialization[V]
+    Caching.get(cache, key, codec)(ex)
   }
 
-  override def flush(): Future[Unit] = cache.cache.removeAll().asTwitter(ec)
+  override def flush(): Future[Unit] = Caching.flush(cache)(ex)
 
   private def createCache(
-      name: String, maxSize: Long, ttl: ScalaDuration, executor: Executor, statsReceiver: StatsReceiver): ScalaCache[InMemoryRepr] = {
+      name: String, maxSize: Long, executor: Executor, statsReceiver: StatsReceiver): ScalaCache[InMemoryRepr] = {
     val underlying = Caffeine.newBuilder()
         .maximumSize(maxSize)
-        .expireAfterWrite(ttl.length, ttl.unit)
+        .expireAfterWrite(ttl.inNanoseconds, TimeUnit.NANOSECONDS)
         .executor(executor)
         .recordStats(() => new StatsCounter(name, statsReceiver))
         .build[String, Object]

src/main/scala/com/redbubble/util/cache/redis/RedisSimpleCache.scala

@@ -1,15 +1,12 @@
 package com.redbubble.util.cache.redis
 
-import java.util.concurrent.{Executor, TimeUnit}
+import java.util.concurrent.Executor
 
-import com.redbubble.util.async.syntax._
-import com.redbubble.util.cache.{CacheKey, SimpleCache}
+import com.redbubble.util.cache.{CacheKey, Caching, SimpleCache}
 import com.redbubble.util.metrics.StatsReceiver
 import com.twitter.util.{Duration, Future}
 
-import scala.concurrent.ExecutionContext.fromExecutor
-import scala.concurrent.duration.{Duration => ScalaDuration}
-import scalacache.{CacheConfig, Flags, ScalaCache}
+import scalacache.{CacheConfig, ScalaCache}
 
 /**
   * An external cache, backed by a Redis instance.
@@ -23,27 +20,24 @@ import scalacache.{CacheConfig, Flags, ScalaCache}
   */
 private[cache] final class RedisSimpleCache(name: String, host: String, port: Int, ttl: Duration)
     (implicit ex: Executor, statsReceiver: StatsReceiver) extends SimpleCache {
-  private val flags = Flags(readsEnabled = true, writesEnabled = true)
-  private val scalaTtl = ScalaDuration(ttl.inNanoseconds, TimeUnit.NANOSECONDS)
-  private val cache = createCache(name, host, port, ex, statsReceiver)
-  private val ec = fromExecutor(ex)
+  private lazy val cache = createCache(name, host, port, ex, statsReceiver)
 
   override def caching[V](key: CacheKey)(f: => Future[V]): Future[V] = {
     val codec = new ScalaCacheExternaliserCodec[V]
-    scalacache.cachingWithTTL[V, Array[Byte]](key)(scalaTtl)(f.asScala)(cache, flags, ec, codec).asTwitter(ec)
+    Caching.caching(cache, ttl, key, codec)(f)(ex)
   }
 
   override def put[V, Repr](key: CacheKey, value: V): Future[Unit] = {
     val codec = new ScalaCacheExternaliserCodec[V]
-    scalacache.put[V, Array[Byte]](key)(value, Some(scalaTtl))(cache, flags, codec).asTwitter(ec)
+    Caching.put(cache, ttl, key, codec, value)(ex)
   }
 
   override def get[V](key: CacheKey): Future[Option[V]] = {
     val codec = new ScalaCacheExternaliserCodec[V]
-    scalacache.get[V, Array[Byte]](key)(cache, flags, codec).asTwitter(ec)
+    Caching.get(cache, key, codec)(ex)
   }
 
-  override def flush(): Future[Unit] = cache.cache.removeAll().asTwitter(ec)
+  override def flush(): Future[Unit] = Caching.flush(cache)
 
   private def createCache(name: String, host: String, port: Int, executor: Executor, statsReceiver: StatsReceiver): ScalaCache[Array[Byte]] = {
     val underlying = MetricsEnableRedisCache(name, host, port)(executor, statsReceiver)