promises.scala

package com.gu.mobileaggregator.memcached

import java.util.concurrent.ConcurrentHashMap
import scala.collection.JavaConverters._
import dispatch.Promise
import com.gu.common.Logging
import com.gu.mobileaggregator.management.MemcachedMetrics

class PromisesInProgress extends Logging {
  val updates = new ConcurrentHashMap[String, Promise[Any]]().asScala

  def updateAndCache[A](key: String, softExpire: Int, hardExpire: Int, f: () => Promise[A]): Promise[A] = {
    val promiseInProgress = updates.get(key)

    promiseInProgress map { x =>
      log.debug("Promise in progress for %s".format(key))

      log.info("Other promise available")

      x.asInstanceOf[Promise[A]]
    } getOrElse {
      log.debug("Initiating promise for %s".format(key))

      MemcachedMetrics.updates.increment()

      val promiseNow = f()

      updates.put(key, promiseNow.asInstanceOf[Promise[Any]])

      promiseNow onSuccess {
        case a: A => {
          log.debug("Promise completed for %s".format(key))
          Memcached.set[CachedStale[A]](key, hardExpire, CachedStale(a, softExpire))
          updates.remove(key)
        }
      }

      promiseNow recover {
        case error => {
          log.debug("Promise error for %s: %s".format(key, error.getMessage))
          updates.remove(key)

          throw error
        }
      }
    }
  }
}