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

Adding support for task batching complete with tests

Browse source
This commit is contained in:
Viktor Klang 2012-08-03 23:33:45 +02:00
parent eb23b5b88d
commit 2af08127de
7 changed files with 84 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

14
akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala
View file

@ -116,7 +116,13 @@ private[akka] case class Unwatch(watchee: ActorRef, watcher: ActorRef) extends S
*/
private[akka] case object NoMessage extends SystemMessage // switched into the mailbox to signal termination
final case class TaskInvocation(eventStream: EventStream, runnable: Runnable, cleanup: () ⇒ Unit) extends Runnable {
final case class TaskInvocation(eventStream: EventStream, runnable: Runnable, cleanup: () ⇒ Unit) extends Batchable {
final override def isBatchable: Boolean = runnable match {
case b: Batchable ⇒ b.isBatchable
case _: scala.concurrent.OnCompleteRunnable ⇒ true
case _ ⇒ false
}
def run(): Unit =
try runnable.run() catch {
case NonFatal(e) ⇒ eventStream.publish(Error(e, "TaskInvocation", this.getClass, e.getMessage))
@ -163,7 +169,7 @@ private[akka] object MessageDispatcher {
implicit def defaultDispatcher(implicit system: ActorSystem): MessageDispatcher = system.dispatcher
}
abstract class MessageDispatcher(val prerequisites: DispatcherPrerequisites) extends AbstractMessageDispatcher with Executor with ExecutionContext {
abstract class MessageDispatcher(val prerequisites: DispatcherPrerequisites) extends AbstractMessageDispatcher with BatchingExecutor with ExecutionContext {
import MessageDispatcher._
import AbstractMessageDispatcher.{ inhabitantsOffset, shutdownScheduleOffset }
@ -209,8 +215,8 @@ abstract class MessageDispatcher(val prerequisites: DispatcherPrerequisites) ext
*/
final def detach(actor: ActorCell): Unit = try unregister(actor) finally ifSensibleToDoSoThenScheduleShutdown()
final override def execute(runnable: Runnable): Unit = {
val invocation = TaskInvocation(eventStream, runnable, taskCleanup)
final override protected def unbatchedExecute(r: Runnable): Unit = {
val invocation = TaskInvocation(eventStream, r, taskCleanup)
addInhabitants(+1)
try {
executeTask(invocation)