raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Merged with master

Browse source
This commit is contained in:
Jonas Bonér 2010-05-05 13:26:31 +02:00
commit be5114e6a7
14 changed files with 629 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

88
akka-core/src/main/scala/routing/Patterns.scala
View file

@ -1,3 +1,7 @@
/**
* Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
*/
package se.scalablesolutions.akka.patterns
import se.scalablesolutions.akka.actor.{Actor, ActorID}
@ -23,16 +27,18 @@ object Patterns {
filter({case a if a.isInstanceOf[A] => interceptor(a)}, interceptee)
//FIXME 2.8, use default params with CyclicIterator
def loadBalancerActor(actors: => InfiniteIterator[ActorID]): ActorID = newActor(() => new Actor with LoadBalancer {
start
val seq = actors
})
def loadBalancerActor(actors: => InfiniteIterator[ActorID]): ActorID =
newActor(() => new Actor with LoadBalancer {
start
val seq = actors
})
def dispatcherActor(routing: PF[Any, ActorID], msgTransformer: (Any) => Any): ActorID = newActor(() => new Actor with Dispatcher {
start
override def transform(msg: Any) = msgTransformer(msg)
def routes = routing
})
def dispatcherActor(routing: PF[Any, ActorID], msgTransformer: (Any) => Any): ActorID =
newActor(() => new Actor with Dispatcher {
start
override def transform(msg: Any) = msgTransformer(msg)
def routes = routing
})
def dispatcherActor(routing: PF[Any, ActorID]): ActorID = newActor(() => new Actor with Dispatcher {
start
@ -41,68 +47,4 @@ object Patterns {
def loggerActor(actorToLog: ActorID, logger: (Any) => Unit): ActorID =
dispatcherActor({case _ => actorToLog}, logger)
}
trait Dispatcher { self: Actor =>
protected def transform(msg: Any): Any = msg
protected def routes: PartialFunction[Any, ActorID]
protected def dispatch: PartialFunction[Any, Unit] = {
case a if routes.isDefinedAt(a) =>
if (self.replyTo.isDefined) routes(a) forward transform(a)
else routes(a) ! transform(a)
}
def receive = dispatch
}
trait LoadBalancer extends Dispatcher { self: Actor =>
protected def seq: InfiniteIterator[ActorID]
protected def routes = { case x if seq.hasNext => seq.next }
}
trait InfiniteIterator[T] extends Iterator[T]
class CyclicIterator[T](items: List[T]) extends InfiniteIterator[T] {
@volatile private[this] var current: List[T] = items
def hasNext = items != Nil
def next = {
val nc = if (current == Nil) items else current
current = nc.tail
nc.head
}
}
class SmallestMailboxFirstIterator(items : List[ActorID]) extends InfiniteIterator[ActorID] {
def hasNext = items != Nil
def next = {
def actorWithSmallestMailbox(a1: ActorID, a2: ActorID) = {
if (a1.mailboxSize < a2.mailboxSize) a1 else a2
}
items.reduceLeft((actor1, actor2) => actorWithSmallestMailbox(actor1,actor2))
}
}
sealed trait ListenerMessage
case class Listen(listener : ActorID) extends ListenerMessage
case class Deafen(listener : ActorID) extends ListenerMessage
case class WithListeners(f : Set[ActorID] => Unit) extends ListenerMessage
trait Listeners { self : Actor =>
import se.scalablesolutions.akka.actor.Agent
private lazy val listeners = Agent(Set[ActorID]())
protected def listenerManagement : PartialFunction[Any,Unit] = {
case Listen(l) => listeners( _ + l)
case Deafen(l) => listeners( _ - l )
case WithListeners(f) => listeners foreach f
}
protected def gossip(msg : Any) = listeners foreach ( _ foreach ( _ ! msg ) )
}