Merge pull request #34 from lloydmeta/feature/json-to-memcached

Use Json serdes for PartResponse caching duties
m3dev · Oct 1, 2014 · 4bd7b80 · 4bd7b80
2 parents 820f977 + 91da09d
commit 4bd7b80
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 15 deletions.
diff --git a/app/com/m3/octoparts/cache/PartResponseCachingSupport.scala b/app/com/m3/octoparts/cache/PartResponseCachingSupport.scala
@@ -11,6 +11,7 @@ import skinny.util.LTSV
 import com.m3.octoparts.cache.RichCacheControl._
 
 import scala.concurrent.Future
+import scala.util.control.NonFatal
 
 private[cache] object PartResponseCachingSupport {
 
@@ -43,16 +44,8 @@ trait PartResponseCachingSupport extends PartRequestServiceBase with Logging {
       super.processWithConfig(ci, partRequestInfo, params)
     } else {
       val directive = CacheDirectiveGenerator.generateDirective(ci.partId, params, ci.cacheConfig)
-      val futureMaybeFromCache = cacheOps.putIfAbsent(directive)(super.processWithConfig(ci, partRequestInfo, params)).recoverWith {
-        case ce: CacheException =>
-          ce.getCause match {
-            case te: shade.TimeoutException =>
-              warn(LTSV.dump("Memcached error" -> "timed out", "cache key" -> ce.key.toString))
-            case other =>
-              error(LTSV.dump("Memcached error" -> other.getClass.getSimpleName, "cache key" -> ce.key.toString), other)
-          }
-          super.processWithConfig(ci, partRequestInfo, params)
-      }
+      val futureMaybeFromCache = cacheOps.putIfAbsent(directive)(super.processWithConfig(ci, partRequestInfo, params)).
+        recoverWith(onCacheFailure(ci, partRequestInfo, params))
       futureMaybeFromCache.flatMap {
         partResponse =>
           // at this point, the response may come from cache and be stale.
@@ -68,6 +61,24 @@ trait PartResponseCachingSupport extends PartRequestServiceBase with Logging {
     }
   }
 
+  private def onCacheFailure(ci: HttpPartConfig,
+                             partRequestInfo: PartRequestInfo,
+                             params: Map[ShortPartParam, String]): PartialFunction[Throwable, Future[PartResponse]] = {
+    case ce: CacheException => {
+      ce.getCause match {
+        case te: shade.TimeoutException =>
+          warn(LTSV.dump("Memcached error" -> "timed out", "cache key" -> ce.key.toString))
+        case other =>
+          error(LTSV.dump("Memcached error" -> other.getClass.getSimpleName, "cache key" -> ce.key.toString), other)
+      }
+      super.processWithConfig(ci, partRequestInfo, params)
+    }
+    case NonFatal(e) => {
+      error(LTSV.dump("Memcached error" -> e.getClass.getSimpleName), e)
+      super.processWithConfig(ci, partRequestInfo, params)
+    }
+  }
+
   private[cache] def revalidate(partResponse: PartResponse,
                                 directive: CacheDirective,
                                 ci: HttpPartConfig,

diff --git a/app/com/m3/octoparts/cache/memcached/MemcachedCacheOps.scala b/app/com/m3/octoparts/cache/memcached/MemcachedCacheOps.scala
@@ -1,6 +1,6 @@
 package com.m3.octoparts.cache.memcached
 
-import com.m3.octoparts.cache.{ Cache, CacheOps }
+import com.m3.octoparts.cache._
 import com.m3.octoparts.cache.directive.CacheDirective
 import com.m3.octoparts.cache.key.{ PartCacheKey, VersionCacheKey }
 import com.m3.octoparts.cache.versioning._
@@ -10,6 +10,7 @@ import skinny.logging.Logging
 
 import scala.concurrent.duration.Duration
 import scala.concurrent.{ ExecutionContext, Future }
+import scala.util.control.NonFatal
 
 class MemcachedCacheOps(
   cache: Cache,
@@ -37,13 +38,17 @@ class MemcachedCacheOps(
   }
 
   object PartResponseCache {
+    import play.api.libs.json.Json
+    import shade.memcached.MemcachedCodecs.StringBinaryCodec
+    import com.m3.octoparts.json.format.ReqResp._
 
-    import shade.memcached.MemcachedCodecs._
-
-    def pollPartResponse(cacheKey: PartCacheKey) = cache.get[PartResponse](cacheKey)
+    def pollPartResponse(cacheKey: PartCacheKey): Future[Option[PartResponse]] = {
+      val fMaybeString = cache.get[String](cacheKey)
+      fMaybeString.map(maybeString => maybeString.map(Json.parse(_).as[PartResponse]))
+    }
 
     def insertPartResponse(cacheKey: PartCacheKey, partResponse: PartResponse, ttl: Option[Duration]): Future[Unit] =
-      cache.put(cacheKey, partResponse, calculateTtl(partResponse.cacheControl, ttl))
+      cache.put[String](cacheKey, Json.toJson(partResponse).toString, calculateTtl(partResponse.cacheControl, ttl))
   }
 
   object CombinedVersionLookup {