ericacm

class ProviderTokenServices(dataSource: DataSource) extends JdbcOAuth2ProviderTokenServices(dataSource) with InitializingBean {
  var log: Logger = LoggerFactory.getLogger(this.getClass.getName)

  @Autowired var domainManager: DomainManager = _
  @Autowired var schedulerService: SchedulerService = _

  @Value("${providerTokenServices.pruneSchedule:0 0 12 * * *}")
  var pruneSchedule: String = _

  override def storeAccessToken(token: OAuth2AccessToken, authentication: OAuth2Authentication[_ <: Authentication, _ <: Authentication]) {

ericacm

Scala Links

Eric Pederson

• [email protected]
• Blog
• LinkedIn
• Twitter
• StackOverflow
• GitHub

ericacm

// ©2012 Viktor Klang

object MiniActor {
  import java.util.concurrent.{ConcurrentLinkedQueue, Executor}
  import java.util.concurrent.atomic.AtomicInteger

  type Behavior = Any => Effect

  sealed trait Effect extends (Behavior => Behavior)
  case object Stay extends Effect { def apply(old: Behavior): Behavior = old }

ericacm

class CacheSystem(name: String, updateIntervalMin: Int, 
                  cacheManager: CacheManager) {

  var caches = List.empty[Cache]
  val actorSystem = ActorSystem("cache_" + name)

  val DEFAULT_TTL_SEC = 86400   // 1 day

  def addCache(name: String, size: Int, 
               ttlSeconds: Int = DEFAULT_TTL_SEC): Cache = {

ericacm

class CachingBusinessService(bizService: BusinessService) 
  extends BusinessService {

  implicit val timeout = Timeout(60 seconds)

  val service1CacheActor = 
    cacheSystem.createCacheActor("service1", DATE_CACHE_SIZE, 0 seconds, 
                                 new Service1CacheActor(_, _, bizService))
  // ... more actors created here

ericacm

object CacheActor {
  case class FindValue(params: Params)

  trait Params {
    def cacheKey: String
  }

  // Thread pool used by findValueForSender()
  val FUTURE_POOL_SIZE = 25

ericacm

abstract class CacheActor[V](cacheSystem: CacheSystem) 
    extends Actor with Logging {

  def findValueReceive: Receive = {
    case FindValue(params) => findValueForSender(params, sender)
  }

  def findValueForSender(params: Params, sender: ActorRef) {
    val key = params.cacheKey
    val elem = cache.get(key)

ericacm

abstract class DateCacheActor[V](cacheSystem: CacheSystem) 
    extends CacheActor[V](cacheSystem) {

  override def receive = findValueReceive orElse  {
    case UpdateCacheForNow => 
      updateCacheForNow()

    case UpdateCacheForPreviousBusinessDay => 
      updateCacheForPreviousBusinessDay()
  }

ericacm

class Service1CacheActor(val cache: Cache, cacheSystem: CacheSystem, 
                         bizService: BusinessService) 
    extends DateCacheActor[JList[Service1Result]](cacheSystem) {

  override def receive = super.receive

  override def updateCacheForDate(date: Date) {
    import DateCacheActor._
    Future { findObject(new Service1Params(date, true)) }
    Future { findObject(new Service1Params(date, false)) }

ericacm

import akka.util.{Duration, NonFatal}
import akka.actor.Scheduler
import akka.dispatch.{Promise, ExecutionContext, Future}
 
// Copied from Akka 2.1-M1 
trait FutureTimeoutSupport {
  /**
   * Returns a [[akka.dispatch.Future]] that will be completed with the success or failure of the provided value
   * after the specified duration.
   */