#1111
- Created on2011-08-15 02:07
- Reported by
- StatusFixed
- PriorityNormal (3)
- Assigned to
- Milestone
- Plan Level-
- Due Date-
Consider adding ScatterGather router
From discussion http://groups.google.com/group/akka-user/browse_thread/thread/f035dd55e703b466:
package akka.routing
import akka.actor.ActorRef
import akka.actor.Timeout
import akka.dispatch.{Future, Futures}
import akka.event.slf4j.Logging
trait ScatterGatherRouter extends Router with Serializable with Logging{
@volatile
protected var connections: RouterConnections = _
protected def gather[T](results: Iterable[Future[T]]): Future[T]
def init(connections: RouterConnections) = {
this.connections = connections
}
private def broadcast(message: Any)(implicit sender: Option[ActorRef]) =
connections.versionedIterator._2.foreach(actor ⇒
try {
actor.!(message)(sender)
} catch {
case e: Exception ⇒
connections.signalDeadActor(actor)
throw e
})
private def scatterGather[T](message: Any, timeout: Timeout)(implicit sender: Option[ActorRef]): Future[T] =
gather(connections
.versionedIterator._2
.map{_.?(message, timeout).asInstanceOf[Future[T]]
})
def route(message: Any)(implicit sender: Option[ActorRef]) = message match {
case Routing.Broadcast(message) ⇒ broadcast(message)
case message ⇒ broadcast(message)
}
def route[T](message: Any, timeout: Timeout)(implicit sender: Option[ActorRef]): Future[T] = message match {
case Routing.Broadcast(message) ⇒ scatterGather(message, timeout)
case message ⇒ scatterGather(message, timeout)
}
}
class ScatterGatherFirstComer extends ScatterGatherRouter {
protected def gather[T](results: Iterable[Future[T]]) = Futures.firstCompletedOf(results)
}
Leave a comment
on 2011-08-15 02:08 *
By Jonas Bonér
Description changed from From user:
package ... to From discussion http://grou...
Fix for the ticket can be found here ~> https://github.com/jboner/akka/compare/master...wip-1111-remeniuk
(In revision:b121da7d2b0624ae94bf994ad350f8b33eeeca66) Scatter-gather router. ScatterGatherRouter boradcasts the message to all connections of a clustered actor, and aggregates responses due to a specified strategy. Fixes #1111
Branch: master
- ScatterGatherRouter can be mixed in with other router types, and inherit their behavior on processing standard and broadcast messages
- ScatterGatherFirstCompletedRouter is a sample implementation of a ScatterGatherRouter that gathers the first reply came from connections
- Multi-JVM tests are commented out so far (custom routers dooesn't work well in a clustered mode, so far), until #1109 is closed
Branch: master
on 2011-08-17 06:12 *
By vasil.remeniuk
(In revision:b121da7d2b0624ae94bf994ad350f8b33eeeca66) Scatter-gather router. ScatterGatherRouter boradcasts the message to all connections of a clustered actor, and aggregates responses due to a specified strategy. Fixes #1111
Branch: wip-props
- ScatterGatherRouter can be mixed in with other router types, and inherit their behavior on processing standard and broadcast messages
- ScatterGatherFirstCompletedRouter is a sample implementation of a ScatterGatherRouter that gathers the first reply came from connections
- Multi-JVM tests are commented out so far (custom routers dooesn't work well in a clustered mode, so far), until #1109 is closed
Branch: wip-props
Updating tickets (#967, #974, #975, #976, #980, #981, #989, #990, #992, #993, #994, #999, #1000, #1004, #1008, #1011, #1015, #1018, #1022, #1023, #1024, #1025, #1027, #1028, #1029, #1030, #1032, #1033, #1036, #1047, #1053, #1062, #1067, #1068, #1069, #1072, #1075, #1078, #1082, #1102, #1107, #1110, #1111, #1115, #1116, #1121, #1122, #1123, #1124)