Removing durable mailboxes from akka

akka-actor/src/main/scala/akka/dispatch/ExecutorBasedEventDrivenDispatcher.scala

@@ -68,7 +68,7 @@ class ExecutorBasedEventDrivenDispatcher(
   _name: String,
   val throughput: Int = Dispatchers.THROUGHPUT,
   val throughputDeadlineTime: Int = Dispatchers.THROUGHPUT_DEADLINE_TIME_MILLIS,
-  _mailboxType: MailboxType = Dispatchers.MAILBOX_TYPE,
+  val mailboxType: MailboxType = Dispatchers.MAILBOX_TYPE,
   val config: ThreadPoolConfig = ThreadPoolConfig())
   extends MessageDispatcher {
 
@@ -88,7 +88,6 @@ class ExecutorBasedEventDrivenDispatcher(
     this(_name, Dispatchers.THROUGHPUT, Dispatchers.THROUGHPUT_DEADLINE_TIME_MILLIS, Dispatchers.MAILBOX_TYPE) // Needed for Java API usage
 
   val name        = "akka:event-driven:dispatcher:" + _name
-  val mailboxType = Some(_mailboxType)
 
   private[akka] val threadFactory = new MonitorableThreadFactory(name)
   private[akka] val executorService = new AtomicReference[ExecutorService](config.createLazyExecutorService(threadFactory))
@@ -106,7 +105,7 @@ class ExecutorBasedEventDrivenDispatcher(
 
   override def mailboxSize(actorRef: ActorRef) = getMailbox(actorRef).size
 
-  def createTransientMailbox(actorRef: ActorRef, mailboxType: TransientMailbox): AnyRef = mailboxType match {
+  def createMailbox(actorRef: ActorRef): AnyRef = mailboxType match {
     case UnboundedMailbox(blocking) => new DefaultUnboundedMessageQueue(blocking) with ExecutableMailbox {
       def dispatcher = ExecutorBasedEventDrivenDispatcher.this
     }
@@ -117,12 +116,6 @@ class ExecutorBasedEventDrivenDispatcher(
       }
   }
 
-  /**
-   * Creates and returns a durable mailbox for the given actor.
-   */
-  private[akka] def createDurableMailbox(actorRef: ActorRef, mailboxType: DurableMailbox): AnyRef =
-    createMailbox(mailboxType.mailboxImplClassname, actorRef)
-
   private[akka] def start = log.slf4j.debug("Starting up {}\n\twith throughput [{}]", this, throughput)
 
   private[akka] def shutdown {

akka-actor/src/main/scala/akka/dispatch/ExecutorBasedEventDrivenWorkStealingDispatcher.scala

@@ -35,16 +35,13 @@ import jsr166x.{Deque, ConcurrentLinkedDeque, LinkedBlockingDeque}
  */
 class ExecutorBasedEventDrivenWorkStealingDispatcher(
   _name: String,
-  _mailboxType: MailboxType = Dispatchers.MAILBOX_TYPE,
+  val mailboxType: MailboxType = Dispatchers.MAILBOX_TYPE,
   config: ThreadPoolConfig = ThreadPoolConfig()) extends MessageDispatcher {
 
   def this(_name: String, mailboxType: MailboxType) = this(_name, mailboxType,ThreadPoolConfig())
 
   def this(_name: String) = this(_name, Dispatchers.MAILBOX_TYPE,ThreadPoolConfig())
 
-  //implicit def actorRef2actor(actorRef: ActorRef): Actor = actorRef.actor
-
-  val mailboxType = Some(_mailboxType)
   val name = "akka:event-driven-work-stealing:dispatcher:" + _name
 
   /** Type of the actors registered in this dispatcher. */
@@ -194,7 +191,7 @@ class ExecutorBasedEventDrivenWorkStealingDispatcher(
 
   override val toString = "ExecutorBasedEventDrivenWorkStealingDispatcher[" + name + "]"
 
-  private[akka] def createTransientMailbox(actorRef: ActorRef, mailboxType: TransientMailbox): AnyRef = mailboxType match {
+  private[akka] def createMailbox(actorRef: ActorRef): AnyRef = mailboxType match {
     case UnboundedMailbox(blocking) => // FIXME make use of 'blocking' in work stealer ConcurrentLinkedDeque
       new ConcurrentLinkedDeque[MessageInvocation] with MessageQueue with Runnable {
         def enqueue(handle: MessageInvocation): Unit = this.add(handle)
@@ -221,12 +218,6 @@ class ExecutorBasedEventDrivenWorkStealingDispatcher(
       }
   }
 
-  /**
-   * Creates and returns a durable mailbox for the given actor.
-   */
-  private[akka] def createDurableMailbox(actorRef: ActorRef, mailboxType: DurableMailbox): AnyRef =
-    createMailbox(mailboxType.mailboxImplClassname, actorRef)
-
   private[akka] override def register(actorRef: ActorRef) = {
     verifyActorsAreOfSameType(actorRef)
     pooledActors add actorRef

akka-actor/src/main/scala/akka/dispatch/HawtDispatcher.scala

@@ -142,8 +142,6 @@ object HawtDispatcher {
 class HawtDispatcher(val aggregate: Boolean = true, val parent: DispatchQueue = globalQueue) extends MessageDispatcher  {
   import HawtDispatcher._
 
-  val mailboxType: Option[MailboxType] = None
-
   private[akka] def start { retainNonDaemon }
 
   private[akka] def shutdown { releaseNonDaemon }
@@ -157,7 +155,7 @@ class HawtDispatcher(val aggregate: Boolean = true, val parent: DispatchQueue =
   // TODO: figure out if this can be optional in akka
   override def mailboxSize(actorRef: ActorRef) = 0
 
-  override def createMailbox(actorRef: ActorRef): AnyRef = {
+  def createMailbox(actorRef: ActorRef): AnyRef = {
     val queue = parent.createSerialQueue(actorRef.toString)
     if (aggregate) new AggregatingHawtDispatcherMailbox(queue)
     else new HawtDispatcherMailbox(queue)
@@ -166,13 +164,6 @@ class HawtDispatcher(val aggregate: Boolean = true, val parent: DispatchQueue =
   def suspend(actorRef: ActorRef) = mailbox(actorRef).suspend
   def resume(actorRef:ActorRef)   = mailbox(actorRef).resume
 
-  private[akka] def createTransientMailbox(actorRef: ActorRef, mailboxType: TransientMailbox): AnyRef = null.asInstanceOf[AnyRef]
-
-  /**
-   * Creates and returns a durable mailbox for the given actor.
-   */
-  private[akka] def createDurableMailbox(actorRef: ActorRef, mailboxType: DurableMailbox): AnyRef = null.asInstanceOf[AnyRef]
-
   override def toString = "HawtDispatcher"
 }
 

akka-actor/src/main/scala/akka/dispatch/MailboxHandling.scala

@@ -31,18 +31,15 @@ trait MessageQueue {
  */
 sealed trait MailboxType
 
-abstract class TransientMailbox(val blocking: Boolean = false) extends MailboxType
+case class UnboundedMailbox(val blocking: Boolean = false) extends MailboxType
-case class UnboundedMailbox(block: Boolean = false) extends TransientMailbox(block)
 case class BoundedMailbox(
-  block: Boolean            = false,
+  val blocking: Boolean            = false,
   val capacity: Int         = { if (Dispatchers.MAILBOX_CAPACITY < 0) Int.MaxValue else Dispatchers.MAILBOX_CAPACITY },
-  val pushTimeOut: Duration = Dispatchers.MAILBOX_PUSH_TIME_OUT) extends TransientMailbox(block) {
+  val pushTimeOut: Duration = Dispatchers.MAILBOX_PUSH_TIME_OUT) extends MailboxType {
   if (capacity < 0)        throw new IllegalArgumentException("The capacity for BoundedMailbox can not be negative")
   if (pushTimeOut eq null) throw new IllegalArgumentException("The push time-out for BoundedMailbox can not be null")
 }
 
-case class DurableMailbox(mailboxImplClassname: String) extends MailboxType
-
 class DefaultUnboundedMessageQueue(blockDequeue: Boolean)
   extends LinkedBlockingQueue[MessageInvocation] with MessageQueue {
 
@@ -69,31 +66,4 @@ class DefaultBoundedMessageQueue(capacity: Int, pushTimeOut: Duration, blockDequ
   final def dequeue(): MessageInvocation =
     if (blockDequeue) this.take()
     else this.poll()
-}
+}
-
-/**
- *  @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
- */
-trait MailboxFactory {
-
-  val mailboxType: Option[MailboxType]
-
-  /**
-   * Creates a MessageQueue (Mailbox) with the specified properties.
-   */
-  private[akka] def createMailbox(actorRef: ActorRef): AnyRef =
-    mailboxType.getOrElse(throw new IllegalStateException("No mailbox type defined")) match {
-      case mb: TransientMailbox => createTransientMailbox(actorRef, mb)
-      case mb: DurableMailbox   => createDurableMailbox(actorRef, mb)
-    }
-
-  /**
-   *  Creates and returns a transient mailbox for the given actor.
-   */
-  private[akka] def createTransientMailbox(actorRef: ActorRef, mailboxType: TransientMailbox): AnyRef
-
-  /**
-   *  Creates and returns a durable mailbox for the given actor.
-   */
-  private[akka] def createDurableMailbox(actorRef: ActorRef, mailboxType: DurableMailbox): AnyRef
-}

akka-actor/src/main/scala/akka/dispatch/MessageHandling.scala

@@ -58,7 +58,7 @@ object MessageDispatcher {
 /**
  *  @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
  */
-trait MessageDispatcher extends MailboxFactory with Logging {
+trait MessageDispatcher extends Logging {
   import MessageDispatcher._
 
   protected val uuids = new ConcurrentSkipListSet[Uuid]
@@ -70,16 +70,7 @@ trait MessageDispatcher extends MailboxFactory with Logging {
   /**
    *  Creates and returns a mailbox for the given actor.
    */
-  def createMailbox(mailboxImplClassname: String, actorRef: ActorRef): MessageQueue = {
+  private[akka] def createMailbox(actorRef: ActorRef): AnyRef
-    ReflectiveAccess.createInstance(
-      mailboxImplClassname,
-      Array(classOf[ActorRef]),
-      Array(actorRef).asInstanceOf[Array[AnyRef]],
-      ReflectiveAccess.loader)
-    .getOrElse(throw new IllegalActorStateException(
-        "Could not create mailbox [" + mailboxImplClassname + "] for actor [" + actorRef + "]"))
-    .asInstanceOf[MessageQueue]
-  }
 
   /**
    * Attaches the specified actorRef to this dispatcher
@@ -100,7 +91,9 @@ trait MessageDispatcher extends MailboxFactory with Logging {
   } else throw new IllegalActorStateException("Can't submit invocations to dispatcher since it's not started")
 
   private[akka] def register(actorRef: ActorRef) {
-    if (actorRef.mailbox eq null) actorRef.mailbox = createMailbox(actorRef)
+    if (actorRef.mailbox eq null)
+      actorRef.mailbox = createMailbox(actorRef)
+
     uuids add actorRef.uuid
     if (active.isOff) {
       active.switchOn {

akka-actor/src/main/scala/akka/util/ReflectiveAccess.scala

@@ -109,25 +109,6 @@ object ReflectiveAccess extends Logging {
 
     def ensureEnabled = if (!isEnabled) throw new ModuleNotAvailableException(
       "Feature is only available in Akka Cloud")
-
-    def createFileBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.cloud.cluster.FileBasedMailbox", actorRef)
-
-    def createZooKeeperBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.cloud.cluster.ZooKeeperBasedMailbox", actorRef)
-
-    def createBeanstalkBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.cloud.cluster.BeanstalkBasedMailbox", actorRef)
-
-    def createRedisBasedMailbox(actorRef: ActorRef): Mailbox = createMailbox("akka.cloud.cluster.RedisBasedMailbox", actorRef)
-
-    private def createMailbox(mailboxClassname: String, actorRef: ActorRef): Mailbox = {
-      ensureEnabled
-      createInstance(
-        mailboxClassname,
-        Array(classOf[ActorRef]),
-        Array(actorRef).asInstanceOf[Array[AnyRef]],
-        loader)
-        .getOrElse(throw new IllegalActorStateException("Could not create durable mailbox [" + mailboxClassname + "] for actor [" + actorRef + "]"))
-        .asInstanceOf[Mailbox]
-    }
   }
 
   val noParams = Array[Class[_]]()