Merging in Actor per Session + fixing blocking problem with remote typed actors with Future response types

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

@@ -326,32 +326,25 @@ class RemoteServer extends Logging with ListenerManagement {
   }
 
   private def register[Key](id: Key, actorRef: ActorRef, registry: ConcurrentHashMap[Key, ActorRef]) {
-    // TODO: contains is an alias for containsValue, not containsKey, so the contains is useless here
+    if (_isRunning) {
-    if (_isRunning && !registry.contains(id)) {
+      registry.put(id, actorRef) //TODO change to putIfAbsent
-      registry.put(id, actorRef);
       if (!actorRef.isRunning) actorRef.start
     }
   }
 
   private def registerPerSession[Key](id: Key, factory: () => ActorRef, registry: ConcurrentHashMap[Key,() => ActorRef]) {
-    // TODO: contains is an alias for containsValue, not containsKey, so the contains is useless here
+    if (_isRunning)
-    if (_isRunning && !registry.contains(id)) {
+      registry.put(id, factory) //TODO change to putIfAbsent
-      registry.put(id, factory)
-    }
   }
 
   private def registerTypedActor[Key](id: Key, typedActor: AnyRef, registry: ConcurrentHashMap[Key, AnyRef]) {
-    // TODO: contains is an alias for containsValue, not containsKey, so the contains is useless here
+    if (_isRunning)
-    if (_isRunning && !registry.contains(id)) {
+      registry.put(id, typedActor) //TODO change to putIfAbsent
-      registry.put(id, typedActor)
-    }
   }
 
   private def registerTypedPerSessionActor[Key](id: Key, factory: () => AnyRef, registry: ConcurrentHashMap[Key,() => AnyRef]) {
-    // TODO: contains is an alias for containsValue, not containsKey, so the contains is useless here
+    if (_isRunning)
-    if (_isRunning && !registry.contains(id)) {
+      registry.put(id, factory) //TODO change to putIfAbsent
-      registry.put(id, factory)
-    }
   }
 
   /**
@@ -531,18 +524,21 @@ class RemoteServerHandler(
     val clientAddress = getClientAddress(ctx)
     log.debug("Remote client [%s] disconnected from [%s]", clientAddress, server.name)
     // stop all session actors
-    val channelActors = sessionActors.remove(event.getChannel())
+    val channelActors = sessionActors.remove(event.getChannel)
-    val channelActorsIterator = channelActors.elements()
+    if (channelActors ne null) {
-    while (channelActorsIterator.hasMoreElements()) {
+      val channelActorsIterator = channelActors.elements
-      channelActorsIterator.nextElement().stop()
+      while (channelActorsIterator.hasMoreElements) {
+        channelActorsIterator.nextElement.stop
+      }
     }
 
-    val channelTypedActors = typedSessionActors.remove(event.getChannel())
+    val channelTypedActors = typedSessionActors.remove(event.getChannel)
-    val channelTypedActorsIterator = channelTypedActors.elements()
+    if (channelTypedActors ne null) {
-    while (channelTypedActorsIterator.hasMoreElements()) {
+      val channelTypedActorsIterator = channelTypedActors.elements
-      TypedActor.stop(channelTypedActorsIterator.nextElement())
+      while (channelTypedActorsIterator.hasMoreElements) {
+        TypedActor.stop(channelTypedActorsIterator.nextElement)
+      }
     }
-
     server.notifyListeners(RemoteServerClientDisconnected(server, clientAddress))
   }
 
@@ -629,7 +625,7 @@ class RemoteServerHandler(
               val exception = f.exception
 
               if (exception.isDefined) {
-                log.debug("Returning exception from actor invocation [%s]".format(exception.get))
+                log.debug("Returning exception from actor invocation [%s]",exception.get)
                 try {
                   channel.write(createErrorReplyMessage(exception.get, request, AkkaActorType.ScalaActor))
                 } catch {
@@ -679,26 +675,35 @@ class RemoteServerHandler(
       val messageReceiver = typedActor.getClass.getDeclaredMethod(typedActorInfo.getMethod, argClasses: _*)
       if (request.getOneWay) messageReceiver.invoke(typedActor, args: _*)
       else {
-        val result = messageReceiver.invoke(typedActor, args: _*) match {
+        //Sends the response
-          case f: Future[_] => f.await.result.get //TODO replace this with a Listener on the Future to avoid blocking
+        def sendResponse(result: Either[Any,Throwable]): Unit = try {
-          case other => other
+          val messageBuilder = RemoteActorSerialization.createRemoteMessageProtocolBuilder(
+            None,
+            Right(request.getUuid),
+            actorInfo.getId,
+            actorInfo.getTarget,
+            actorInfo.getTimeout,
+            result,
+            true,
+            None,
+            None,
+            AkkaActorType.TypedActor,
+            None)
+          if (request.hasSupervisorUuid) messageBuilder.setSupervisorUuid(request.getSupervisorUuid)
+          channel.write(messageBuilder.build)
+          log.debug("Returning result from remote typed actor invocation [%s]", result)
+        } catch {
+          case e: Throwable => server.notifyListeners(RemoteServerError(e, server))
         }
-        log.debug("Returning result from remote typed actor invocation [%s]", result)
 
-        val messageBuilder = RemoteActorSerialization.createRemoteMessageProtocolBuilder(
+        messageReceiver.invoke(typedActor, args: _*) match {
-          None,
+          case f: Future[_] => //If it's a future, we can lift on that to defer the send to when the future is completed
-          Right(request.getUuid), 
+            f.onComplete( future => {
-          actorInfo.getId, 
+              val result: Either[Any,Throwable] = if (future.exception.isDefined) Right(future.exception.get) else Left(future.result.get)
-          actorInfo.getTarget, 
+              sendResponse(result)
-          actorInfo.getTimeout, 
+            })
-          Left(result),
+          case other => sendResponse(Left(other))
-          true, 
+        }
-          None,
-          None, 
-          AkkaActorType.TypedActor, 
-          None)
-        if (request.hasSupervisorUuid) messageBuilder.setSupervisorUuid(request.getSupervisorUuid)
-        channel.write(messageBuilder.build)
       }
     } catch {
       case e: InvocationTargetException =>