patriknw

Cluster(system) registerOnMemberUp {
  system.actorOf(Props(new FactorialFrontend(upToN, repeat = true)),
    name = "factorialFrontend")
}

patriknw

system.actorOf(Props(new ClusterSingletonManager(
  singletonProps = handOverData => Props[Master]),
  singletonName = "master",
  terminationMessage = PoisonPill)),
  name = "singleton")

patriknw

class MasterProxy extends Actor {
  // subscribe to LeaderChanged, re-subscribe when restart
  override def preStart(): Unit =
    Cluster(context.system).subscribe(self, classOf[LeaderChanged])
  override def postStop(): Unit =
    Cluster(context.system).unsubscribe(self)
 
  var leaderAddress: Option[Address] = None
 
  def receive = {

patriknw

import akka.actor._

object MessageDiscarder {
  case object WorkerAvailable
}

class MessageDiscarder extends Actor {
  import MessageDiscarder._

  var queued: Option[(Any, ActorRef)] = None

patriknw

/**
 *  Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
 */

package akka.contrib.pattern

import language.postfixOps
import scala.concurrent.duration._
import com.typesafe.config.ConfigFactory
import akka.actor.Actor

patriknw

class TransformationBackend extends Actor {
  val cluster = Cluster(context.system)

  override def preStart(): Unit =
    cluster.subscribe(self, classOf[MemberUp])
  override def postStop(): Unit =
    cluster.unsubscribe(self)

  def receive = {
    case state: CurrentClusterState ⇒

patriknw

  // sort by age, oldest first
  val ageOrdering = Ordering.fromLessThan[Member] { (a, b) => 
    a.isOlderThan(b) 
  }
  var membersByAge: SortedSet[Member] = SortedSet.empty(ageOrdering)

  def receive = {
    case state: CurrentClusterState =>
      membersByAge = SortedSet.empty(ageOrdering) ++ state.members
    }

patriknw

package akka.contrib.mailbox

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger
import com.typesafe.config.Config
import akka.actor.{ ActorContext, ActorRef, ActorSystem, ExtendedActorSystem, Extension, ExtensionId, ExtensionIdProvider }
import akka.dispatch.{ Envelope, MailboxType, MessageQueue, UnboundedMailbox, UnboundedQueueBasedMessageQueue }

object MetricsMailboxExtension extends ExtensionId[MetricsMailboxExtension] with ExtensionIdProvider {
  def lookup = this

patriknw

Cluster(system) registerOnMemberUp {
  system.actorOf(Props(classOf[FactorialFrontend], upToN, true),
    name = "factorialFrontend")
}

patriknw

class Publisher extends Actor {
  import DistributedPubSubMediator.Publish
  // activate the extension
  val mediator = DistributedPubSubExtension(context.system).mediator
 
  def receive = {
    case in: String ⇒
      val out = in.toUpperCase
      mediator ! Publish("content", out)
  }