Making ThreadBasedDispatcher Unbounded if no capacity specced and fix a possible mem leak in it

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

@@ -77,11 +77,12 @@ object Dispatchers extends Logging {
    * <p/>
    * E.g. each actor consumes its own thread.
    */
-  def newThreadBasedDispatcher(actor: ActorRef) = new ThreadBasedDispatcher(actor, BoundedMailbox(true))
+  def newThreadBasedDispatcher(actor: ActorRef) = new ThreadBasedDispatcher(actor)
 
   /**
    * Creates an thread based dispatcher serving a single actor through the same single thread.
    * Uses the default timeout
+   * If capacity is negative, it's Integer.MAX_VALUE
    * <p/>
    * E.g. each actor consumes its own thread.
    */
@@ -89,6 +90,7 @@ object Dispatchers extends Logging {
 
   /**
    * Creates an thread based dispatcher serving a single actor through the same single thread.
+   * If capacity is negative, it's Integer.MAX_VALUE
    * <p/>
    * E.g. each actor consumes its own thread.
    */

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

@@ -10,30 +10,40 @@ import se.scalablesolutions.akka.util.Duration
 
 import java.util.Queue
 import java.util.concurrent.{ConcurrentLinkedQueue, BlockingQueue, TimeUnit, LinkedBlockingQueue}
+import se.scalablesolutions.akka.actor
+import java.util.concurrent.atomic.AtomicReference
 
 /**
  * Dedicates a unique thread for each actor passed in as reference. Served through its messageQueue.
  *
  * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
  */
-class ThreadBasedDispatcher(private val actor: ActorRef, _mailboxType: MailboxType) 
+class ThreadBasedDispatcher(_actor: ActorRef, _mailboxType: MailboxType)
   extends ExecutorBasedEventDrivenDispatcher(
-    actor.getClass.getName + ":" + actor.uuid,
+    _actor.uuid.toString,Dispatchers.THROUGHPUT,-1,_mailboxType,ThreadBasedDispatcher.oneThread) {
-    Dispatchers.THROUGHPUT,
-    -1,
-    _mailboxType,
-    ThreadBasedDispatcher.oneThread) {
 
-  def this(actor: ActorRef) = this(actor, BoundedMailbox(true)) // For Java API
+  private[akka] val owner = new AtomicReference[ActorRef](_actor)
 
-  def this(actor: ActorRef, capacity: Int) = this(actor, BoundedMailbox(true, capacity))
+  def this(actor: ActorRef) =
+    this(actor, UnboundedMailbox(true)) // For Java API
 
-  def this(actor: ActorRef, capacity: Int, pushTimeOut: Duration) = this(actor, BoundedMailbox(true, capacity, pushTimeOut))
+  def this(actor: ActorRef, capacity: Int) =
+    this(actor, BoundedMailbox(true, capacity)) //For Java API
+
+  def this(actor: ActorRef, capacity: Int, pushTimeOut: Duration) = //For Java API
+    this(actor, BoundedMailbox(true, capacity, pushTimeOut))
 
   override def register(actorRef: ActorRef) = {
-    if (actorRef != actor) throw new IllegalArgumentException("Cannot register to anyone but " + actor)
+    val actor = owner.get()
+    if ((actor ne null) && actorRef != actor) throw new IllegalArgumentException("Cannot register to anyone but " + actor)
+    owner.compareAndSet(null,actorRef) //Register if unregistered
     super.register(actorRef)
   }
+
+  override def unregister(actorRef: ActorRef) = {
+    super.unregister(actorRef)
+    owner.compareAndSet(actorRef,null) //Unregister (prevent memory leak)
+  }
 }
 
 object ThreadBasedDispatcher {

akka-remote/src/main/scala/remote/RemoteServer.scala

@@ -618,7 +618,7 @@ class RemoteServerHandler(
         log.info("Creating a new remote actor [%s:%s]", name, uuid)
         val clazz = if (applicationLoader.isDefined) applicationLoader.get.loadClass(name)
                     else Class.forName(name)
-        val actorRef = Actor.actorOf(clazz.newInstance.asInstanceOf[Actor])
+        val actorRef = Actor.actorOf(clazz.asInstanceOf[Class[_ <: Actor]])
         actorRef.uuid = uuidFrom(uuid.getHigh,uuid.getLow)
         actorRef.id = id
         actorRef.timeout = timeout