Closing #1584 and #1585

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

@@ -5,29 +5,40 @@
 package akka.remote
 
 import akka.remote.RemoteProtocol._
-import akka.serialization.Serialization
 import com.google.protobuf.ByteString
 import akka.actor.ActorSystem
 import akka.serialization.SerializationExtension
+import akka.util.ReflectiveAccess
 
 object MessageSerializer {
 
   def deserialize(system: ActorSystem, messageProtocol: MessageProtocol, classLoader: Option[ClassLoader] = None): AnyRef = {
-    val clazz = loadManifest(classLoader, messageProtocol)
-    SerializationExtension(system).deserialize(messageProtocol.getMessage.toByteArray,
-      clazz, classLoader).fold(x ⇒ throw x, identity)
+    val clazz = if (messageProtocol.hasMessageManifest) {
+      Option(ReflectiveAccess.getClassFor[AnyRef](
+        messageProtocol.getMessageManifest.toStringUtf8,
+        classLoader.getOrElse(ReflectiveAccess.loader)) match {
+          case Left(e)  ⇒ throw e
+          case Right(r) ⇒ r
+        })
+    } else None
+    SerializationExtension(system).deserialize(
+      messageProtocol.getMessage.toByteArray,
+      messageProtocol.getSerializerId,
+      clazz,
+      classLoader) match {
+        case Left(e)  ⇒ throw e
+        case Right(r) ⇒ r
+      }
   }
 
   def serialize(system: ActorSystem, message: AnyRef): MessageProtocol = {
+    val s = SerializationExtension(system)
+    val serializer = s.findSerializerFor(message)
     val builder = MessageProtocol.newBuilder
-    val bytes = SerializationExtension(system).serialize(message).fold(x ⇒ throw x, identity)
-    builder.setMessage(ByteString.copyFrom(bytes))
-    builder.setMessageManifest(ByteString.copyFromUtf8(message.getClass.getName))
+    builder.setMessage(ByteString.copyFrom(serializer.toBinary(message)))
+    builder.setSerializerId(serializer.identifier)
+    if (serializer.includeManifest)
+      builder.setMessageManifest(ByteString.copyFromUtf8(message.getClass.getName))
     builder.build
   }
-
-  private def loadManifest(classLoader: Option[ClassLoader], messageProtocol: MessageProtocol): Class[_] = {
-    val manifest = messageProtocol.getMessageManifest.toStringUtf8
-    classLoader map (_.loadClass(manifest)) getOrElse (Class.forName(manifest))
-  }
 }