iss #22: cachable RepositoriesFetcher (with some multithread issues)

api_stubs/config.json

@@ -18,7 +18,7 @@
                     "type": ["open", "private"]
                 }
             },
-
+            "delay": 1.5,
             "response_file": "orgs/hhru/repos/page1.json",
             "headers_file": "orgs/hhru/repos/page1.headers"
         },
@@ -46,7 +46,7 @@
                     "type": ["open", "private"]
                 }
             },
-
+            "delay": 0.5,
             "response_file": "orgs/hhru/repos/page2.json",
             "headers_file": "orgs/hhru/repos/page2.headers"
         },

app/hedgehog/clients/github/RepositoriesFetcher.scala

@@ -1,8 +1,10 @@
 package hedgehog.clients.github
 
 import scala.concurrent.{Future, Promise}
+import scala.concurrent.duration._
 import scala.collection.mutable
-import com.github.nscala_time.time.Imports.DateTime
+import play.api.Logger
+import play.api.cache.Cache
 import hedgehog.models.{AccountSettings, Repo}
 
 
@@ -11,29 +13,75 @@ import hedgehog.models.{AccountSettings, Repo}
  * See LICENSE.txt for details.
  */
 class RepositoriesFetcher(client: Client) {
-  val repositories: mutable.Map[String, Option[Seq[Repo]]] = mutable.Map.empty
-  val lastFetched: mutable.Map[String, Option[DateTime]] = mutable.Map.empty
+  type Repos = Seq[Repo]
 
   private var hits_ = 0
   private var misses_ = 0
 
-  implicit val context = client.context
+  private val CACHE_TIME = 5.minutes //TODO: from configs
 
-  def getRepos(accountsSettings: Seq[AccountSettings]): Future[Seq[Repo]] = {
-    hits_ += 1
-    //TODO: cache
-    fetch(accountsSettings)
+  private val fetching = mutable.Map[AccountSettings, mutable.ListBuffer[Promise[Repos]]]() //FIXME: thread unsafe
+  private implicit val context = client.context
+  private implicit val app = play.api.Play.current  //TODO: get from client
+
+  private def getSettingsKey(settings: AccountSettings): String = s"repos-${settings.hashCode().toHexString}"
+
+  def getRepos(accountsSettings: Seq[AccountSettings]): Future[Repos] = {
+    val requests = Future.sequence(
+      accountsSettings.map {
+        accountSettings =>
+          val key = getSettingsKey(accountSettings)
+          Cache.getAs[Repos](key) match {
+            case Some(repos) =>
+              hits_ += 1
+              Logger.debug(s"Get repos for $accountSettings from cache")
+              Future.successful(repos)
+            case None =>
+              misses_ += 1
+              fetch(accountSettings)
+          }
+      }
+    )
+    requests.map(_.flatten)
   }
 
-  private def fetch(accountsSettings: Seq[AccountSettings]): Future[Seq[Repo]] = {
-    misses_ += 1
-    val requests = Future.sequence {
-      accountsSettings.map(client.getRepos)
-    }
-    val future = requests map {
-      case r => r.flatten
-    }
-    future
+  private def fetch(accountSettings: AccountSettings): Future[Repos] = {
+      val promise = Promise[Repos]()
+      // other fetch runs in parallel
+      if (fetching contains accountSettings) {
+        Logger.debug(s"Fetching ever run for $accountSettings, prepend promise")
+        fetching(accountSettings) prepend promise
+
+      // first fetch
+      } else {
+        val key = getSettingsKey(accountSettings)
+        Logger.debug(s"Fetch repos for $accountSettings")
+        fetching(accountSettings) = mutable.ListBuffer(promise)
+        val clientResult = client.getRepos(accountSettings)
+        def sendResult(res: Repos) {
+          fetching(accountSettings).foreach(_.success(res))
+          fetching -= accountSettings
+        }
+        clientResult onFailure {
+          case e =>
+            Cache.getAs[Repos](key) match {
+              case Some(repos) =>
+                Logger.warn(s"Unable to fetch repos for $accountSettings get old version from cache")
+                sendResult(repos)
+              case None =>
+                Logger.error(s"Errors when fetch repos for $accountSettings, skipping")
+                //return empty seq on errors
+                sendResult(Seq[Repo]())
+            }
+        }
+
+        clientResult onSuccess {
+          case res =>
+            Cache.set(key, res, CACHE_TIME)
+            sendResult(res)
+        }
+      }
+      promise.future
   }
 
   def reload(accountSettings: Seq[AccountSettings]): Future[Seq[(AccountSettings, Boolean)]] = {

app/hedgehog/models/Account.scala

@@ -31,6 +31,9 @@ trait Account {
     s"$webProfileUrl/${urlEncodePathSegment(repo.fullName)}"
 
   def apiReposUrl: String
+
+  override def toString: String =
+    s"Account($accountType: $name)"
 }
 
 

app/hedgehog/models/Sources.scala

@@ -43,6 +43,9 @@ class AccountSettings(
 
   override def hashCode(): Int =
     Seq(account, includePrivateRepos).map(_.hashCode).foldLeft(0)((a, b) => 31 * a + b)
+
+  override def toString: String =
+    s"AccountSettings($account, includePrivateRepos=$includePrivateRepos)"
 }
 
 

build.sbt

@@ -5,7 +5,7 @@ version := "0.0.1-SNAPSHOT"
 libraryDependencies ++= Seq(
   //jdbc,
   //anorm,
-  //cache,
+  cache,
   "org.scalatest" % "scalatest_2.10" % "2.1.7" % "test",
   "com.github.nscala-time" %% "nscala-time" % "1.2.0",
   "com.netaporter" %% "scala-uri" % "0.4.2"

conf/dev.conf

@@ -18,7 +18,7 @@ github {
     # key - user/org name, value - options
     sources: [
         # users
-        maizy
+        maizy,
         {
             name = hhru
             org = true
@@ -58,5 +58,5 @@ logger: {
     application: DEBUG
 
     # HTTP Requests & responses log
-    com.ning.http.client: DEBUG
+    # com.ning.http.client: DEBUG
 }

scripts/console.scala

@@ -9,6 +9,7 @@ val userAccount = GithubUser("user", Some("http://example.com/u.png"))
 val orgAccount = GithubOrg("org", Some("http://example.com/o.png"))
 
 val userAccountSettings = new AccountSettings(userAccount, includePrivateRepos = false)
+val userAccountSettings2 = new AccountSettings(userAccount, includePrivateRepos = false)
 val orgAccountSettings = new AccountSettings(orgAccount, includePrivateRepos = true)
 val accountSettings = userAccountSettings :: orgAccountSettings :: Nil
 
@@ -21,3 +22,7 @@ val orgRepoY = new Repo("y", userAccount, Some("some description y"), isPrivate
 val orgRepos = orgRepoZ :: orgRepoY :: Nil
 
 val repos = userRepos ++ userRepos
+
+import play.api.libs.concurrent.Execution.Implicits._
+import scala.concurrent.duration._
+import scala.concurrent._