ericacm

def selectLeader() {
  leaderLatch = new LeaderLatch(curatorFramework, leaderPath, nodeId)

  leaderLatch.start()

  // Optional, only if you need notification on cluster changes
  watchLeaderChildren()
}

ericacm

def watchLeaderChildren() {
  curatorFramework.getChildren.usingWatcher(
   new CuratorWatcher {
     def process(event: WatchedEvent) {
       val cs = clusterStatus()
       
       // Do something with cluster status (log leadership change, etc)

       // Re-set watch
       curatorFramework.getChildren.usingWatcher(this).

ericacm

def clusterStatus: ClusterStatus = {
  val isLeader = leaderLatch.hasLeadership
  val participants = leaderLatch.getParticipants.asScala
  val leader = participants.find(_.isLeader).map(_.getId).getOrElse("<none>")

  ClusterStatus(nodeId, isLeader, leader, participants.map(_.getId))
}

ericacm

  val isWindows = {
    val osName = System.getProperty("os.name")
    val isWin = osName.startsWith("Windows")
    println("os.name=" + osName + " isWindows=" + isWin)
    isWin
  }

  def currentJars: Array[String] = {
    val classloader = ClassLoader.getSystemClassLoader

ericacm

2012-10-05 16:33:59,923 | ERROR | akka.actor.ActorSystemImpl          | t-dispatcher-164 | RemoteServerError@akka://CSLServer@localhost:1997] Error[java.io.IOException:An existing connection was forcibly closed by the remote host
                at sun.nio.ch.SocketDispatcher.read0(Native Method)
                at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:43)
                at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:218)
                at sun.nio.ch.IOUtil.read(IOUtil.java:186)
                at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:359)
                at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:63)
                at org.jboss.netty.channel.socket.nio.AbstractNioWorker.processSelectedKeys(AbstractNioWorker.java:385)
                at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:256)
                at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:35)

ericacm

2012-10-05 16:20:55,932 | ERROR | akka.actor.ActorSystemImpl          | ult-dispatcher-1 | RemoteClientError@akka://CSLWorker@localhost:2000: Error[java.net.ConnectException:Connection refused: no further information
                at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
                at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:701)
                at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.connect(NioClientSocketPipelineSink.java:404)
                at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.processSelectedKeys(NioClientSocketPipelineSink.java:366)
                at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.run(NioClientSocketPipelineSink.java:282)
                at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:102)
                at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
                at java.util.c

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.
   */

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

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

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

    case UpdateCacheForPreviousBusinessDay => 
      updateCacheForPreviousBusinessDay()
  }

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)