json4s serializer for [EventStore.Akka.Persistence](https://github.com/EventStore/EventStore.Akka.Persistence)

JsonSerializer.scala

package com.company.serialization

import java.nio.ByteBuffer
import java.nio.charset.Charset

import akka.actor.{ActorRef, ExtendedActorSystem}
import akka.persistence.PersistentRepr
import akka.persistence.eventstore.EventStoreSerializer
import akka.persistence.eventstore.snapshot.EventStoreSnapshotStore.SnapshotEvent
import akka.persistence.journal.Tagged
import akka.util.ByteString
import eventstore.{Content, ContentType, Event, EventData}
import org.json4s.Extraction.decompose
import org.json4s.JsonAST.JString
import org.json4s._
import org.json4s.native.Serialization.{read, write}

class JsonSerializer(val system: ExtendedActorSystem)
  extends EventStoreSerializer {

  import JsonSerializer._

  //JsonFormats.Domain contains my customized serializers and such.
  implicit val formats = JsonFormats.Domain +
    new PersistentReprSerializer(system) +
    ActorRefSerializer

  def identifier = Identifier

  def includeManifest = true

  def fromBinary(bytes: Array[Byte], manifestOpt: Option[Class[_]]) = {
    implicit val manifest = manifestOpt match {
      case Some(x) => Manifest.classType(x)
      case None => Manifest.AnyRef
    }
    read(new String(bytes, UTF8))
  }

  def toBinary(o: AnyRef) = write(o).getBytes(UTF8)

  def toEvent(x: AnyRef) = x match {
    case x: PersistentRepr =>
      val (repr: PersistentRepr, metadata) = x.payload match {
        case Tagged(event, tags) => x.withPayload(event) -> meta(x, Map("tag" -> JString(tags.mkString(","))))
        case event => x -> meta(x)
      }
      EventData(
        eventType = repr.payload.getClass.getSimpleName,
        data = Content(ByteString(toBinary(repr)), ContentType.Json),
        metadata = metadata
      )

    case x: SnapshotEvent => EventData(
      eventType = classFor(x).getName,
      data = Content(ByteString(toBinary(x)), ContentType.Json),
      metadata = meta(x))

    case _ => sys.error(s"Cannot serialize $x, PersistentRepr or SnapshotEvent expected")
  }

  def fromEvent(event: Event, manifest: Class[_]) = {
    val clazz = getContainerClass(event)
    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")
  }

  def meta(x: AnyRef, data: Map[String, JValue] = Map.empty) = Content(ByteString(toBinary(Map(
    "container" -> classFor(x).getName
  ) ++ data)), ContentType.Json)

  def getContainerClass(event: Event) = {
    val bytes: Array[Byte] = event.data.metadata.value.toArray
    val meta: Map[String, String] = read[Map[String, String]](new String(bytes, UTF8))
    Class.forName(
      meta.get("container") match {
        case Some(x) => x
        case _ => classOf[PersistentRepr].getName
      }
    )
  }

  def classFor(x: AnyRef) = x match {
    case x: PersistentRepr => classOf[PersistentRepr]
    case _ => x.getClass
  }

  object ActorRefSerializer extends Serializer[ActorRef] {
    val Clazz = classOf[ActorRef]

    def deserialize(implicit format: Formats) = {
      case (TypeInfo(Clazz, _), JString(x)) => system.provider.resolveActorRef(x)
    }

    def serialize(implicit format: Formats) = {
      case x: ActorRef => JString(x.path.toSerializationFormat)
    }
  }
}

object JsonSerializer {
  val UTF8: Charset = Charset.forName("UTF-8")
  val Identifier: Int = ByteBuffer.wrap("json4s".getBytes(UTF8)).getInt

  class PersistentReprSerializer(system: ExtendedActorSystem) extends Serializer[PersistentRepr] {
    val Clazz = classOf[PersistentRepr]

    def deserialize(implicit format: Formats) = {
      case (TypeInfo(Clazz, _), json) =>
        val x = json.extract[PersistentRepMapping]
        PersistentRepr(
          payload = x.payload,
          sequenceNr = x.sequenceNr,
          persistenceId = x.persistenceId,
          manifest = x.manifest,
          sender = system.provider.resolveActorRef(x.sender),
          writerUuid = x.writerUuid)
    }
    def serialize(implicit format: Formats) = {
      case x: PersistentRepr =>
        val mapping = PersistentRepMapping(
          payload = x.payload,
          sequenceNr = x.sequenceNr,
          persistenceId = x.persistenceId,
          manifest = x.manifest,
          sender = Option(x.sender) match {
            case Some(sender) => sender.path.toSerializationFormat
            case None => ""
          },
          writerUuid = x.writerUuid)
        decompose(mapping)
    }

  }
}

case class PersistentRepMapping(
                                 payload: Any,
                                 sequenceNr: Long,
                                 persistenceId: String,
                                 manifest: String,
                                 sender: String,
                                 writerUuid: String)