change serialization to strictly rely on subtyping

- when encountering new message type, check all bindings which map apply
- if multiple are found, choose the most specific one if that exists or
  verify that all mappings yield the same serializer
- in case of remaining ambiguity, throw exception
- also add special handling for “none” serializer mapping: turn off a
  default

akka-actor-tests/src/test/scala/akka/serialization/SerializeSpec.scala

@@ -179,14 +179,10 @@ class SerializeSpec extends AkkaSpec(SerializeSpec.config) {
       ser.serializerFor(classOf[ExtendedPlainMessage]).getClass must be(classOf[TestSerializer])
     }
 
-    "resolve serializer for message with several bindings" in {
-      ser.serializerFor(classOf[Both]).getClass must be(classOf[TestSerializer])
+    "throw exception for message with several bindings" in {
+      intercept[java.io.NotSerializableException] {
+        ser.serializerFor(classOf[Both])
       }
-
-    "resolve serializer in the order of the bindings" in {
-      ser.serializerFor(classOf[A]).getClass must be(classOf[JavaSerializer])
-      ser.serializerFor(classOf[B]).getClass must be(classOf[TestSerializer])
-      ser.serializerFor(classOf[C]).getClass must be(classOf[JavaSerializer])
     }
 
     "resolve serializer in the order of most specific binding first" in {

akka-actor/src/main/resources/reference.conf

@@ -269,15 +269,16 @@ akka {
     # Entries for pluggable serializers and their bindings.
     serializers {
       java = "akka.serialization.JavaSerializer"
-      # proto = "akka.serialization.ProtobufSerializer"
     }
 
     # Class to Serializer binding. You only need to specify the name of an interface
-    # or abstract base class of the messages. In case of ambiguity it is primarily
-    # using the most specific configured class, and secondly the entry configured first.
+    # or abstract base class of the messages. In case of ambiguity it is 
+    # using the most specific configured class, throwing an exception otherwise.
+    #
+    # To disable one of the default serializers, assign its class to "none", like
+    # "java.io.Serializable" = none
     serialization-bindings {
       "java.io.Serializable" = java
-      #"com.google.protobuf.Message" = proto
     }
   }
 

akka-actor/src/main/scala/akka/serialization/Serialization.scala

@@ -125,18 +125,24 @@ class Serialization(val system: ExtendedActorSystem) extends Extension {
   def serializerFor(clazz: Class[_]): Serializer =
     serializerMap.get(clazz) match {
       case null ⇒
-        val ser = bindings.collectFirst {
-          case (c, s) if c.isAssignableFrom(clazz) ⇒ s
-        } getOrElse (throw new NotSerializableException(
-          "No configured serialization-bindings for class [%s]" format clazz.getName))
+        // bindings are ordered from most specific to least specific
+        def unique(cs: Seq[Class[_]], ser: Set[Serializer]): Boolean = (cs forall (_ isAssignableFrom cs(0))) || ser.size == 1
 
-        // memorize for performance
+        val possible = bindings filter { _._1 isAssignableFrom clazz }
+        possible.size match {
+          case 0 ⇒
+            throw new NotSerializableException("No configured serialization-bindings for class [%s]" format clazz.getName)
+          case x if x == 1 || unique(possible map (_._1), possible.map(_._2)(scala.collection.breakOut)) ⇒
+            val ser = possible(0)._2
             serializerMap.putIfAbsent(clazz, ser) match {
               case null ⇒
                 log.debug("Using serializer[{}] for message [{}]", ser.getClass.getName, clazz.getName)
                 ser
               case some ⇒ some
             }
+          case _ ⇒
+            throw new NotSerializableException("Multiple serializers found for " + clazz + ": " + possible)
+        }
       case ser ⇒ ser
     }
 
@@ -160,7 +166,7 @@ class Serialization(val system: ExtendedActorSystem) extends Extension {
    *  It is primarily ordered by the most specific classes first, and secondly in the configured order.
    */
   private[akka] val bindings: Seq[ClassSerializer] = {
-    val configuredBindings = for ((k: String, v: String) ← settings.SerializationBindings) yield {
+    val configuredBindings = for ((k: String, v: String) ← settings.SerializationBindings if v != "none") yield {
       val c = ReflectiveAccess.getClassFor(k, system.internalClassLoader).fold(throw _, identity[Class[_]])
       (c, serializers(v))
     }

akka-docs/java/serialization.rst

@@ -32,14 +32,26 @@ should be serialized using which ``Serializer``, this is done in the "akka.actor
 
 .. includecode:: ../scala/code/akka/docs/serialization/SerializationDocSpec.scala#serialization-bindings-config
 
-You only need to specify the name of an interface or abstract base class of the messages. In case of ambiguity,
-i.e. the message implements several of the configured classes, it is primarily using the most specific
-configured class, and secondly the entry configured first.
+You only need to specify the name of an interface or abstract base class of the
+messages. In case of ambiguity, i.e. the message implements several of the
+configured classes, the most specific configured class will be used, i.e. the
+one of which all other candidates are superclasses. If this condition cannot be
+met, because e.g. ``java.io.Serializable`` and ``MyOwnSerializable`` both apply
+and neither is a subtype of the other, an exception will be thrown during
+serialization.
 
-Akka provides serializers for ``java.io.Serializable`` and `protobuf <http://code.google.com/p/protobuf/>`_
-``com.google.protobuf.Message`` by default, so normally you don't need to add configuration for that, but
-it can be done to force a specific serializer in case messages implements both ``java.io.Serializable``
-and ``com.google.protobuf.Message``.
+Akka provides serializers for :class:`java.io.Serializable` and `protobuf
+<http://code.google.com/p/protobuf/>`_
+:class:`com.google.protobuf.GeneratedMessage` by default (the latter only if
+depending on the akka-remote module), so normally you don't need to add
+configuration for that; since :class:`com.google.protobuf.GeneratedMessage`
+implements :class:`java.io.Serializable`, protobuf messages will always by
+serialized using the protobuf protocol unless specifically overridden. In order
+to disable a default serializer, map its marker type to “none”::
+
+  akka.actor.serialization-bindings {
+    "java.io.Serializable" = none
+  }
 
 Verification
 ------------

akka-docs/scala/serialization.rst

@@ -32,14 +32,26 @@ should be serialized using which ``Serializer``, this is done in the "akka.actor
 
 .. includecode:: code/akka/docs/serialization/SerializationDocSpec.scala#serialization-bindings-config
 
-You only need to specify the name of an interface or abstract base class of the messages. In case of ambiguity,
-i.e. the message implements several of the configured classes, it is primarily using the most specific
-configured class, and secondly the entry configured first.
+You only need to specify the name of an interface or abstract base class of the
+messages. In case of ambiguity, i.e. the message implements several of the
+configured classes, the most specific configured class will be used, i.e. the
+one of which all other candidates are superclasses. If this condition cannot be
+met, because e.g. ``java.io.Serializable`` and ``MyOwnSerializable`` both apply
+and neither is a subtype of the other, an exception will be thrown during
+serialization.
 
-Akka provides serializers for ``java.io.Serializable`` and `protobuf <http://code.google.com/p/protobuf/>`_
-``com.google.protobuf.Message`` by default, so normally you don't need to add configuration for that, but
-it can be done to force a specific serializer in case messages implements both ``java.io.Serializable``
-and ``com.google.protobuf.Message``.
+Akka provides serializers for :class:`java.io.Serializable` and `protobuf
+<http://code.google.com/p/protobuf/>`_
+:class:`com.google.protobuf.GeneratedMessage` by default (the latter only if
+depending on the akka-remote module), so normally you don't need to add
+configuration for that; since :class:`com.google.protobuf.GeneratedMessage`
+implements :class:`java.io.Serializable`, protobuf messages will always by
+serialized using the protobuf protocol unless specifically overridden. In order
+to disable a default serializer, map its marker type to “none”::
+
+  akka.actor.serialization-bindings {
+    "java.io.Serializable" = none
+  }
 
 Verification
 ------------

akka-remote/src/main/resources/reference.conf

@@ -11,17 +11,18 @@ akka {
 
     # Entries for pluggable serializers and their bindings.
     serializers {
-      java = "akka.serialization.JavaSerializer"
       proto = "akka.serialization.ProtobufSerializer"
     }
 
 
     # Class to Serializer binding. You only need to specify the name of an interface
-    # or abstract base class of the messages. In case of ambiguity it is primarily
-    # using the most specific configured class, and secondly the entry configured first.
+    # or abstract base class of the messages. In case of ambiguity it is 
+    # using the most specific configured class, giving an error if two mappings are found
+    # which cannot be decided by sub-typing relation.
     serialization-bindings {
-      "com.google.protobuf.Message" = proto
-      "java.io.Serializable" = java
+      # Since com.google.protobuf.Message does not extend Serializable but GeneratedMessage
+      # does, need to use the more specific one here in order to avoid ambiguity
+      "com.google.protobuf.GeneratedMessage" = proto
     }
 
     deployment {