Make possible to add own serializer for Event => T and `T => EventD…

…ata`

README.md

@@ -13,7 +13,7 @@
   </tr>
   <tr>
     <td><a href="https://github.com/EventStore/EventStore.JVM">eventstore-client</a> </td>
-    <td>1.0.0</td>
+    <td>1.0.1</td>
   </tr>
 </table>
 
@@ -32,11 +32,13 @@ To configure EventStore.JVM client, see it's [reference.conf](https://github.com
 
 Akka serializes your messages into binary data by default.
 However you can [add your own serializer](http://doc.akka.io/docs/akka/2.3.6/scala/serialization.html#Customization) to serialize as JSON,
-But make sure to extend `akka.persistence.eventstore.EventStoreSerializer` rather then `akka.serialization.Serializer`
+But make sure you extend `akka.persistence.eventstore.EventStoreSerializer` rather then `akka.serialization.Serializer`. 
+And in case you are really going to serialize as json, please specify `ContentType.Json`, it will allow you to use projections.
 
 ```scala
-class JsonSerializer extends EventStoreSerializer {
-  def contentType = ContentType.Json // This will tell Event Store to handle your data as JSON
+trait EventStoreSerializer extends Serializer {
+  def toEvent(o: AnyRef): EventData
+  def fromEvent(event: Event, manifest: Class[_]): AnyRef
 }
 ```
 

src/main/scala/akka/persistence/eventstore/EventStorePlugin.scala

@@ -3,38 +3,38 @@ package akka.persistence.eventstore
 import akka.actor.{ ActorLogging, Actor }
 import akka.serialization.{ SerializationExtension, Serialization }
 import scala.concurrent.Future
-import scala.util.{ Failure, Success, Try }
-import eventstore.{ ContentType, Content, EsConnection, EventStoreExtension }
-import akka.util.ByteString
+import scala.util.control.NonFatal
+import eventstore._
 
 trait EventStorePlugin extends ActorLogging { self: Actor =>
   val connection: EsConnection = EventStoreExtension(context.system).connection
-  private val serialization: Serialization = SerializationExtension(context.system)
+  val serialization: Serialization = SerializationExtension(context.system)
   import context.dispatcher
 
-  def deserialize[T](content: Content, clazz: Class[T]): T = {
-    serialization.deserialize(content.value.toArray, clazz).get
+  def deserialize[T](event: Event, clazz: Class[T]): T = {
+    val ser = serialization.serializerFor(clazz)
+    val res = ser match {
+      case ser: EventStoreSerializer => ser.fromEvent(event, clazz)
+      case _                         => ser.fromBinary(event.data.data.value.toArray, clazz)
+    }
+    res.asInstanceOf[T]
   }
 
-  def serialize(x: AnyRef): Content = {
-    val serializer = serialization.findSerializerFor(x)
-    val contentType = serializer match {
-      case x: EventStoreSerializer => x.contentType
-      case _                       => ContentType.Binary
+  def serialize(data: AnyRef, eventType: => Option[Any] = None): EventData = {
+    val ser = serialization.findSerializerFor(data)
+    ser match {
+      case ser: EventStoreSerializer => ser.toEvent(data)
+      case _ => EventData(
+        eventType = (eventType getOrElse data).getClass.getName,
+        data = Content(ser.toBinary(data)))
     }
-    Content(ByteString(serializer.toBinary(x)), contentType)
   }
 
   def asyncUnit(x: => Future[_]): Future[Unit] = async(x).map(_ => Unit)
 
-  def async[T](x: => Future[T]): Future[T] = Try(x) match {
-    case Success(s) => s
-    case Failure(f) => Future.failed(f)
+  def async[T](x: => Future[T]): Future[T] = try x catch {
+    case NonFatal(f) => Future.failed(f)
   }
 
   def asyncSeq[A](x: => Iterable[Future[A]]): Future[Unit] = asyncUnit(Future.sequence(x))
-
-  def logNoClassFoundFor(eventType: String): Unit = {
-    log.warning("Can't find class to deserialize eventType {}", eventType)
-  }
 }

src/main/scala/akka/persistence/eventstore/EventStoreSerializer.scala

@@ -1,8 +1,9 @@
 package akka.persistence.eventstore
 
 import akka.serialization.Serializer
-import eventstore.ContentType
+import eventstore.{ Event, EventData }
 
 trait EventStoreSerializer extends Serializer {
-  def contentType: ContentType
+  def toEvent(o: AnyRef): EventData
+  def fromEvent(event: Event, manifest: Class[_]): AnyRef
 }

src/main/scala/akka/persistence/eventstore/Helpers.scala

@@ -40,7 +40,7 @@ object Helpers {
       if (self == Long.MaxValue) Int.MaxValue
       else {
         if (self.isValidInt) self.toInt
-        else sys.error(s"Can't convert $self to Int")
+        else sys.error(s"Cannot convert $self to Int")
       }
   }
 

src/main/scala/akka/persistence/eventstore/journal/EventStoreJournal.scala

@@ -19,7 +19,7 @@ class EventStoreJournal extends AsyncWriteJournal with EventStorePlugin {
   def asyncWriteMessages(messages: Seq[PersistentRepr]) = asyncSeq {
     messages.groupBy(_.persistenceId).map {
       case (persistenceId, msgs) =>
-        val events = msgs.map(eventData)
+        val events = msgs.map(x => serialize(x, Some(x.payload)))
         val expVer = msgs.head.sequenceNr - 1 match {
           case 0L => ExpectedVersion.NoStream
           case x  => ExpectedVersion.Exact(eventNumber(x))
@@ -57,7 +57,7 @@ class EventStoreJournal extends AsyncWriteJournal with EventStorePlugin {
         connection.foldLeft(req, max) {
           case (left, event) if event.number <= to && left > 0 =>
             val seqNr = sequenceNumber(event.number)
-            val repr = persistentRepr(event.data).update(deleted = seqNr <= deletedTo)
+            val repr = deserialize(event, classOf[PersistentRepr]).update(deleted = seqNr <= deletedTo)
             replayCallback(repr)
             left - 1
         }
@@ -74,16 +74,9 @@ class EventStoreJournal extends AsyncWriteJournal with EventStorePlugin {
 
   def eventStream(x: PersistenceId): EventStream.Plain = EventStream(UrlEncoder(x)) match {
     case plain: EventStream.Plain => plain
-    case other                    => sys.error(s"can't create plain event stream for $x")
+    case other                    => sys.error(s"Cannot create plain event stream for $x")
   }
 
-  def eventData(x: PersistentRepr): EventData = EventData(
-    eventType = x.payload.getClass.getSimpleName,
-    data = serialize(x))
-
-  def persistentRepr(x: EventData): PersistentRepr =
-    deserialize[PersistentRepr](x.data, classOf[PersistentRepr])
-
   def deletedTo(persistenceId: PersistenceId): Future[SequenceNr] = {
     deleteToCache.get(persistenceId) match {
       case Some(x) => Future.successful(x)

src/main/scala/akka/persistence/eventstore/snapshot/EventStoreSnapshotStore.scala

@@ -23,15 +23,14 @@ class EventStoreSnapshotStore extends SnapshotStore with EventStorePlugin {
   def loadAsync(persistenceId: PersistenceId, criteria: SnapshotSelectionCriteria) = async {
     import Selection._
     def fold(deletes: Deletes, event: Event): Selection = {
-      val eventType = event.data.eventType
-      ClassMap.get(eventType) match {
-        case None =>
-          logNoClassFoundFor(eventType)
-          deletes
+      deserialize(event, classOf[SnapshotEvent]) match {
+        case Delete(seqNr, _) => deletes.copy(deleted = deletes.deleted + seqNr)
 
-        case Some(SnapshotClass) =>
-          val metadata = deserialize(event.data.metadata, classOf[SnapshotMetadata])
+        case DeleteCriteria(maxSeqNr, maxTimestamp) => deletes.copy(
+          minSequenceNr = math.max(deletes.minSequenceNr, maxSeqNr),
+          minTimestamp = math.max(deletes.minTimestamp, maxTimestamp))
 
+        case Snapshot(snapshot, metadata) =>
           val seqNr = metadata.sequenceNr
           val timestamp = metadata.timestamp
 
@@ -42,31 +41,18 @@ class EventStoreSnapshotStore extends SnapshotStore with EventStorePlugin {
           val acceptable = seqNr <= criteria.maxSequenceNr && timestamp <= criteria.maxTimestamp
 
           if (deleted || !acceptable) deletes
-          else {
-            val snapshot = deserialize(event.data.data, SnapshotClass)
-            Selected(SelectedSnapshot(metadata, snapshot.data))
-          }
-
-        case Some(clazz) => deserialize(event.data.data, clazz) match {
-          case Snapshot(_)      => deletes // should not happen
-          case Delete(seqNr, _) => deletes.copy(deleted = deletes.deleted + seqNr)
-          case DeleteCriteria(maxSeqNr, maxTimestamp) => deletes.copy(
-            minSequenceNr = math.max(deletes.minSequenceNr, maxSeqNr),
-            minTimestamp = math.max(deletes.minTimestamp, maxTimestamp))
-        }
+          else Selected(SelectedSnapshot(metadata, snapshot))
       }
     }
 
     val streamId = eventStream(persistenceId)
     val req = ReadStreamEvents(streamId, EventNumber.Last, maxCount = readBatchSize, direction = Backward)
-    connection.foldLeft(req, Empty) {
-      case (deletes: Deletes, event) => fold(deletes, event)
-    }.map(_.selected)
+    connection.foldLeft(req, Empty) { case (deletes: Deletes, event) => fold(deletes, event) }.map(_.selected)
   }
 
   def saveAsync(metadata: SnapshotMetadata, snapshot: Any) = asyncUnit {
     val streamId = eventStream(metadata.persistenceId)
-    connection.future(WriteEvents(streamId, List(eventData(metadata, snapshot))))
+    connection.future(WriteEvents(streamId, List(serialize(Snapshot(snapshot, metadata), Some(snapshot)))))
   }
 
   def saved(metadata: SnapshotMetadata) = {}
@@ -81,38 +67,21 @@ class EventStoreSnapshotStore extends SnapshotStore with EventStorePlugin {
       maxTimestamp = criteria.maxTimestamp))
   }
 
-  def eventData(metadata: SnapshotMetadata, snapshot: Any): EventData = EventData(
-    eventType = EventTypeMap(SnapshotClass),
-    data = serialize(Snapshot(snapshot)),
-    metadata = serialize(metadata))
-
-  def eventData(x: SnapshotEvent): EventData = EventData(
-    eventType = EventTypeMap(x.getClass),
-    data = serialize(x))
-
   def eventStream(x: PersistenceId): EventStream.Id = EventStream.Id(UrlEncoder(x) + "-snapshots")
 
   def delete(persistenceId: PersistenceId, se: DeleteEvent): Unit = {
     val streamId = eventStream(persistenceId)
-    val future = connection.future(WriteEvents(streamId, List(eventData(se))))
+    val future = connection.future(WriteEvents(streamId, List(serialize(se, None))))
     Await.result(future, deleteAwait)
   }
 }
 
 object EventStoreSnapshotStore {
-  sealed trait SnapshotEvent
+  sealed trait SnapshotEvent extends Serializable
 
   object SnapshotEvent {
-    val SnapshotClass: Class[Snapshot] = classOf[Snapshot]
-    val ClassMap: Map[String, Class[_ <: SnapshotEvent]] = Map(
-      "snapshot" -> SnapshotClass,
-      "delete" -> classOf[Delete],
-      "deleteCriteria" -> classOf[DeleteCriteria])
-
-    val EventTypeMap: Map[Class[_ <: SnapshotEvent], String] = ClassMap.map(_.swap)
-
-    @SerialVersionUID(0)
-    case class Snapshot(data: Any) extends SnapshotEvent
+    @SerialVersionUID(1)
+    case class Snapshot(data: Any, metadata: SnapshotMetadata) extends SnapshotEvent
 
     sealed trait DeleteEvent extends SnapshotEvent
 

src/test/resources/json4s.conf

@@ -3,10 +3,6 @@ include "application"
 akka.actor {
   serializers.json4s = "akka.persistence.eventstore.Json4sSerializer"
   serialization-bindings {
-    "akka.persistence.eventstore.snapshot.EventStoreSnapshotStore$SnapshotEvent$Snapshot" = json4s
-    "akka.persistence.eventstore.snapshot.EventStoreSnapshotStore$SnapshotEvent$Delete" = json4s
-    "akka.persistence.eventstore.snapshot.EventStoreSnapshotStore$SnapshotEvent$DeleteCriteria" = json4s
-
-    "akka.persistence.SnapshotMetadata" = json4s
+    "akka.persistence.eventstore.snapshot.EventStoreSnapshotStore$SnapshotEvent" = json4s
   }
 }

src/test/resources/spray-json.conf

@@ -3,10 +3,6 @@ include "application"
 akka.actor {
   serializers.spray-json = "akka.persistence.eventstore.SprayJsonSerializer"
   serialization-bindings {
-    "akka.persistence.eventstore.snapshot.EventStoreSnapshotStore$SnapshotEvent$Snapshot" = spray-json
-    "akka.persistence.eventstore.snapshot.EventStoreSnapshotStore$SnapshotEvent$Delete" = spray-json
-    "akka.persistence.eventstore.snapshot.EventStoreSnapshotStore$SnapshotEvent$DeleteCriteria" = spray-json
-
-    "akka.persistence.SnapshotMetadata" = spray-json
+    "akka.persistence.eventstore.snapshot.EventStoreSnapshotStore$SnapshotEvent" = spray-json
   }
 }

src/test/scala/akka/persistence/eventstore/Json4sSerializer.scala

@@ -1,16 +1,22 @@
 package akka.persistence.eventstore
 
+import akka.actor.ExtendedActorSystem
+import akka.persistence.SnapshotMetadata
+import akka.persistence.eventstore.snapshot.EventStoreSnapshotStore.SnapshotEvent
 import akka.persistence.eventstore.snapshot.EventStoreSnapshotStore.SnapshotEvent.Snapshot
+import akka.util.ByteString
 import org.json4s._
+import org.json4s.Extraction.decompose
 import org.json4s.native.Serialization.{ read, write }
 import java.nio.charset.Charset
 import java.nio.ByteBuffer
-import snapshot.EventStoreSnapshotStore.SnapshotEvent.SnapshotClass
-import eventstore.ContentType
+import eventstore.{ Content, EventData, Event, ContentType }
 
-class Json4sSerializer extends EventStoreSerializer {
+class Json4sSerializer(val system: ExtendedActorSystem) extends EventStoreSerializer {
   import Json4sSerializer._
 
+  implicit val formats: Formats = DefaultFormats + SnapshotSerializer
+
   def identifier = Identifier
 
   def includeManifest = true
@@ -25,23 +31,37 @@ class Json4sSerializer extends EventStoreSerializer {
 
   def toBinary(o: AnyRef) = write(o).getBytes(UTF8)
 
-  def contentType = ContentType.Json
+  def toEvent(x: AnyRef) = x match {
+    case x: SnapshotEvent => EventData(
+      eventType = x.getClass.getName,
+      data = Content(ByteString(toBinary(x)), ContentType.Json))
+
+    case _ => sys.error(s"Cannot serialize $x, SnapshotEvent expected")
+  }
+
+  def fromEvent(event: Event, manifest: Class[_]) = {
+    val clazz = Class.forName(event.data.eventType)
+    val result = fromBinary(event.data.data.value.toArray, clazz)
+    if (manifest.isInstance(result)) result
+    else sys.error(s"Cannot deserialize event as $manifest, event: $event")
+  }
 }
 
 object Json4sSerializer {
-
   val UTF8 = Charset.forName("UTF-8")
   val Identifier: Int = ByteBuffer.wrap("json4s".getBytes(UTF8)).getInt
-  implicit val formats: Formats = DefaultFormats + SnapshotSerializer
 
   object SnapshotSerializer extends Serializer[Snapshot] {
+    val Clazz = classOf[Snapshot]
 
     def deserialize(implicit format: Formats) = {
-      case (TypeInfo(SnapshotClass, _), JObject(List(JField("data", JString(x))))) => Snapshot(x)
+      case (TypeInfo(Clazz, _), JObject(List(
+        JField("data", JString(x)),
+        JField("metadata", metadata)))) => Snapshot(x, metadata.extract[SnapshotMetadata])
     }
 
     def serialize(implicit format: Formats) = {
-      case Snapshot(x: String) => JObject("data" -> JString(x))
+      case Snapshot(data, metadata) => JObject("data" -> JString(data.toString), "metadata" -> decompose(metadata))
     }
   }
 }