#43: added support for overlapping persisters using a marker interface

README.md

@@ -73,15 +73,15 @@ val v1CartCreatedPersister = persister[CartCreatedV1]("cart-created")
 
 // spray-json persister for V2 but with support for migration
 // of data writen in the V1 format.
-val v2CartCreatedPersister = persister[CartCreatedV2, V2](
+val v2CartCreatedPersister = persisterVn[CartCreatedV2, V2](
   "cart-created",
   from[V1]
     .to[V2](_.update('cart / 'items / * / 'price ! set[Int](1000)))
 )
 
 // spray-json persister for V3 but with support for migration
 // of data writen in the V1 and V2 formats.
-val v3CartCreatedPersister = persister[CartCreatedV3, V3](
+val v3CartCreatedPersister = persisterVn[CartCreatedV3, V3](
   "cart-created",
   from[V1]
     .to[V2](_.update('cart / 'items / * / 'price ! set[Int](1000)))

project/Build.scala

@@ -41,7 +41,8 @@ object Build extends Build {
     .settings(libSettings: _*)
     .settings(libraryDependencies ++=
       compile(
-        akkaActor
+        akkaActor,
+	scalaReflect
       ) ++
       test(
         scalatest

project/Dependencies.scala

@@ -7,6 +7,8 @@ object Dependencies {
   val jsonLenses    = "net.virtual-void"   %% "json-lenses"      % "0.6.1"
   val scalatest     = "org.scalatest"      %% "scalatest"        % "2.2.6"
   val base64        = "me.lessis"          %% "base64"           % "0.2.0"
+  val scalaReflect =  "org.scala-lang" % "scala-reflect" % "2.11.7"
+
 
   // Only used by the tests
   val sprayJsonShapeless = "com.github.fommil" %% "spray-json-shapeless" % "1.1.0"

stamina-core/src/main/scala/stamina/Persister.scala

@@ -2,13 +2,14 @@ package stamina
 
 import scala.reflect._
 import scala.util._
+import scala.reflect.runtime.universe.{Try ⇒ uTry, _}
 
 /**
  * A Persister[T, V] provides a type-safe API for persisting instances of T
  * at version V and unpersisting persisted instances of T for all versions up
  * to and including version V.
  */
-abstract class Persister[T: ClassTag, V <: Version: VersionInfo](val key: String) {
+abstract class Persister[T: ClassTag, V <: Version: VersionInfo](val key: String)(typeTagOption: Option[TypeTag[T]] = None) {
   lazy val currentVersion = Version.numberFor[V]
 
   def persist(t: T): Persisted
@@ -18,8 +19,17 @@ abstract class Persister[T: ClassTag, V <: Version: VersionInfo](val key: String
   def canUnpersist(p: Persisted): Boolean = p.key == key && p.version <= currentVersion
 
   private[stamina] def convertToT(any: AnyRef): Option[T] = any match {
-    case t: T ⇒ Some(t)
-    case _    ⇒ None
+    case t: T ⇒ t match {
+      case tagged: TypeTagged[_] if typeTagOption.isDefined ⇒
+        val typeTag = tagged.tag
+        val currentTypeTag = typeTagOption.get.tpe.toString
+        typeTag.tpe.toString match {
+          case `currentTypeTag` ⇒ Some(t)
+          case _                ⇒ None
+        }
+      case _ ⇒ Some(t)
+    }
+    case _ ⇒ None
   }
 
   private[stamina] def persistAny(any: AnyRef): Persisted = {
@@ -37,5 +47,9 @@ abstract class Persister[T: ClassTag, V <: Version: VersionInfo](val key: String
     }
   }
 
-  private[stamina] val tag = classTag[T]
+  private[stamina] val tag = typeTagOption.map(_.tpe).getOrElse(classTag[T].runtimeClass)
+}
+
+object Persister {
+  implicit def optionTypeTag[E](implicit typeTag: TypeTag[E]) = Some(typeTag)
 }

stamina-core/src/main/scala/stamina/Persisters.scala

@@ -33,7 +33,7 @@ case class Persisters(persisters: List[Persister[_, _]]) {
 
   private def requireNoOverlappingTags() = {
     val overlappingTags = persisters.groupBy(_.tag).filter(_._2.length > 1).mapValues(_.map(_.key))
-    val warnings = overlappingTags.map { case (tag, keys) ⇒ s"""Persisters with keys ${keys.mkString("'", "', '", "'")} all persist ${tag.runtimeClass}.""" }
+    val warnings = overlappingTags.map { case (tag, keys) ⇒ s"""Persisters with keys ${keys.mkString("'", "', '", "'")} all persist ${tag}.""" }
 
     require(overlappingTags.isEmpty, s"""Overlapping persisters: ${warnings.mkString(" ")}""")
   }

stamina-core/src/main/scala/stamina/TypeTagged.scala

@@ -0,0 +1,22 @@
+package stamina
+
+import scala.reflect.runtime.universe._
+
+/**
+ * This marker interface can be used to solve the problem of nested json formats of the same
+ * root format.
+ * By example:
+ * trait Event[E] {
+ * }
+ *
+ * case class Payload1()
+ * case class Payload2()
+ *
+ * The Persister cannot distinguish Event[Payload1] from Event[Payload2] due to type erasure within
+ * Akka serialization to AnyRef. Therefore you can mark your Event envelop using a TypeTagged marker
+ * interface which whould allow stamina to choose the correct persister for the kind of event payload
+ * which should get serialized.
+ */
+class TypeTagged[X: TypeTag] extends AnyRef {
+  @transient val tag = typeTag[X]
+}

stamina-core/src/test/scala/stamina/PersistersSpec.scala

@@ -1,4 +1,10 @@
 package stamina
+import scala.reflect.runtime.universe._
+
+case class PayloadEvent[E: TypeTag](payload: E) extends TypeTagged[PayloadEvent[E]]
+case class Payload1(txt: String)
+case class Payload2(value: Int)
+case object Payload3
 
 class PersistersSpec extends StaminaSpec {
   import TestDomain._
@@ -57,4 +63,27 @@ class PersistersSpec extends StaminaSpec {
         be thrownBy unpersist(Persisted("item", 1, ByteString("not an item")))
     }
   }
+
+  "Persist overlapping events using the TypeTagged marker interface" should {
+    val persister1 = persister[PayloadEvent[Payload1]]("payload1")
+    val persister2 = persister[PayloadEvent[Payload2]]("payload2")
+
+    val event1 = PayloadEvent(Payload1("test"))
+    val event2 = PayloadEvent(Payload2(123))
+    val event3 = PayloadEvent(Payload3)
+
+    val nestedPersisters = Persisters(persister1, persister2)
+    import nestedPersisters._
+
+    "Persist nested events correctly" in {
+      canPersist(event1) should be(true)
+      canPersist(event2) should be(true)
+      canPersist(event3) should be(false)
+    }
+
+    "correctly implement canUnpersist()" in {
+      canUnpersist(persister1.persist(event1)) should be(true)
+      canUnpersist(persister2.persist(event2)) should be(true)
+    }
+  }
 }

stamina-core/src/test/scala/stamina/TestOnlyPersister.scala

@@ -1,17 +1,18 @@
 package stamina
 
-import scala.reflect._
 import akka.actor._
 import akka.serialization._
+import scala.reflect.ClassTag
+import scala.reflect.runtime.universe._
 
 object TestOnlyPersister {
   private val system = ActorSystem("TestOnlyPersister")
   private val javaSerializer = new JavaSerializer(system.asInstanceOf[ExtendedActorSystem])
   import javaSerializer._
 
-  def persister[T <: AnyRef: ClassTag](key: String): Persister[T, V1] = new JavaPersister[T](key)
+  def persister[T <: AnyRef: ClassTag](key: String)(implicit typeTag: TypeTag[T] = null): Persister[T, V1] = new JavaPersister[T](key)(Option(typeTag))
 
-  private class JavaPersister[T <: AnyRef: ClassTag](key: String) extends Persister[T, V1](key) {
+  private class JavaPersister[T <: AnyRef: ClassTag](key: String)(typeTagOption: Option[TypeTag[T]] = None) extends Persister[T, V1](key)(typeTagOption) {
     def persist(t: T): Persisted = Persisted(key, currentVersion, toBinary(t))
     def unpersist(p: Persisted): T = {
       if (canUnpersist(p)) fromBinary(p.bytes.toArray).asInstanceOf[T]

stamina-json/src/main/scala/stamina/json/json.scala

@@ -2,6 +2,7 @@ package stamina
 
 import scala.reflect.ClassTag
 import spray.json._
+import scala.reflect.runtime.universe._
 
 import migrations._
 
@@ -45,15 +46,15 @@ package object json {
    * and unpersist version 1. Use this function to produce the initial persister
    * for a new domain class/event/entity.
    */
-  def persister[T: RootJsonFormat: ClassTag](key: String): JsonPersister[T, V1] = new V1JsonPersister[T](key)
+  def persister[T: RootJsonFormat: ClassTag](key: String)(implicit typeTag: TypeTag[T] = null): JsonPersister[T, V1] = new V1JsonPersister[T](key)(Option(typeTag))
 
   /**
    * Creates a JsonPersister[T, V] where V is a version greater than V1.
    * It will always persist instances of T to version V but it will use the specified
    * JsonMigrator[V] to migrate any values older than version V to version V before
    * unpersisting them.
    */
-  def persister[T: RootJsonFormat: ClassTag, V <: Version: VersionInfo: MigratableVersion](key: String, migrator: JsonMigrator[V]): JsonPersister[T, V] = new VnJsonPersister[T, V](key, migrator)
+  def persisterVn[T: RootJsonFormat: ClassTag, V <: Version: VersionInfo: MigratableVersion](key: String, migrator: JsonMigrator[V])(implicit typeTag: TypeTag[T] = null): JsonPersister[T, V] = new VnJsonPersister[T, V](key, migrator)(Option(typeTag))
 
   private[json] def toJsonBytes[T](t: T)(implicit writer: RootJsonWriter[T]): ByteString = ByteString(writer.write(t).compactPrint)
   private[json] def fromJsonBytes[T](bytes: ByteString)(implicit reader: RootJsonReader[T]): T = reader.read(parseJson(bytes))
@@ -64,20 +65,20 @@ package json {
   /**
    * Simple abstract marker superclass to unify (and hide) the two internal Persister implementations.
    */
-  sealed abstract class JsonPersister[T: RootJsonFormat: ClassTag, V <: Version: VersionInfo](key: String) extends Persister[T, V](key) {
+  sealed abstract class JsonPersister[T: RootJsonFormat: ClassTag, V <: Version: VersionInfo](key: String)(typeTagOption: Option[TypeTag[T]]) extends Persister[T, V](key)(typeTagOption) {
     private[json] def cannotUnpersist(p: Persisted) =
       s"""JsonPersister[${implicitly[ClassTag[T]].runtimeClass.getSimpleName}, V${currentVersion}](key = "${key}") cannot unpersist data with key "${p.key}" and version ${p.version}."""
   }
 
-  private[json] class V1JsonPersister[T: RootJsonFormat: ClassTag](key: String) extends JsonPersister[T, V1](key) {
+  private[json] class V1JsonPersister[T: RootJsonFormat: ClassTag](key: String)(typeTagOption: Option[TypeTag[T]]) extends JsonPersister[T, V1](key)(typeTagOption) {
     def persist(t: T): Persisted = Persisted(key, currentVersion, toJsonBytes(t))
     def unpersist(p: Persisted): T = {
       if (canUnpersist(p)) fromJsonBytes[T](p.bytes)
       else throw new IllegalArgumentException(cannotUnpersist(p))
     }
   }
 
-  private[json] class VnJsonPersister[T: RootJsonFormat: ClassTag, V <: Version: VersionInfo: MigratableVersion](key: String, migrator: JsonMigrator[V]) extends JsonPersister[T, V](key) {
+  private[json] class VnJsonPersister[T: RootJsonFormat: ClassTag, V <: Version: VersionInfo: MigratableVersion](key: String, migrator: JsonMigrator[V])(typeTagOption: Option[TypeTag[T]]) extends JsonPersister[T, V](key)(typeTagOption) {
     override def canUnpersist(p: Persisted): Boolean = p.key == key && migrator.canMigrate(p.version)
 
     def persist(t: T): Persisted = Persisted(key, currentVersion, toJsonBytes(t))

stamina-json/src/test/scala/stamina/json/JsonPersisterSpec.scala

@@ -8,12 +8,12 @@ class JsonPersisterSpec extends StaminaJsonSpec {
 
   val v1CartCreatedPersister = persister[CartCreatedV1]("cart-created")
 
-  val v2CartCreatedPersister = persister[CartCreatedV2, V2](
+  val v2CartCreatedPersister = persisterVn[CartCreatedV2, V2](
     "cart-created",
     from[V1].to[V2](_.update('cart / 'items / * / 'price ! set[Int](1000)))
   )
 
-  val v3CartCreatedPersister = persister[CartCreatedV3, V3](
+  val v3CartCreatedPersister = persisterVn[CartCreatedV3, V3](
     "cart-created",
     from[V1]
       .to[V2](_.update('cart / 'items / * / 'price ! set[Int](1000)))

stamina-json/src/test/scala/stamina/json/OverlappingPersistersSpec.scala

@@ -3,6 +3,7 @@ package json
 
 import spray.json._
 import DefaultJsonProtocol._
+import scala.reflect.runtime.universe._
 
 class OverlappingPersisterSpec extends StaminaJsonSpec {
   import OverlappingPersisterSpecDomain._
@@ -17,28 +18,21 @@ class OverlappingPersisterSpec extends StaminaJsonSpec {
 
     /** #43 In the future we might want to support this situation instead of failing at initialization time */
     "correctly handle overlapping persisters" in {
-      val e = intercept[IllegalArgumentException] {
-        Persisters(
-          persister[Event[Payload1]]("payload1"),
-          persister[Event[Payload2]]("payload2")
-        )
-      }
-      e.getMessage() should be("requirement failed: Overlapping persisters: Persisters with keys 'payload1', 'payload2' all persist class stamina.json.OverlappingPersisterSpecDomain$Event.")
-
-      /**
-       * When we actually want to support this situation, then this should work:
-       *
-       * val event1 = Event(Payload1("abcd"))
-       * persisters.unpersist(persisters.persist(event1)) should equal(event1)
-       * val event2 = Event(Payload2(42))
-       * persisters.unpersist(persisters.persist(event2)) should equal(event2)
-       */
+      val persisters = Persisters(
+        persister[Event[Payload1]]("payload1"),
+        persister[Event[Payload2]]("payload2")
+      )
+
+      val event1 = Event(Payload1("abcd"))
+      persisters.unpersist(persisters.persist(event1)) should equal(event1)
+      val event2 = Event(Payload2(42))
+      persisters.unpersist(persisters.persist(event2)) should equal(event2)
     }
   }
 }
 
 object OverlappingPersisterSpecDomain {
-  case class Event[P](payload: P)
+  case class Event[P: TypeTag](payload: P) extends TypeTagged[Event[P]]
   case class Payload1(msg: String)
   case class Payload2(value: Int)
 }

stamina-testkit/src/test/scala/stamina/testkit/ScalatestTestGenerationSpec.scala

@@ -11,7 +11,7 @@ class ScalatestTestGenerationSpec extends StaminaTestKitSpec {
 
   import TestDomain._
 
-  case class ItemPersister(override val key: String) extends Persister[Item, V1](key) {
+  case class ItemPersister(override val key: String) extends Persister[Item, V1](key)(None) {
     def persist(t: Item): Persisted = Persisted(key, currentVersion, ByteString())
     def unpersist(p: Persisted): Item = item1
   }