diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/Event.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/Event.scala index b272365..53b7f21 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/Event.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/Event.scala @@ -1,8 +1,8 @@ package com.rbmhtechnology.eventuate.sandbox case class EventMetadata(emitterId: String, emitterLogId: String, localLogId: String, localSequenceNr: Long, vectorTimestamp: VectorTime) -case class EventVersion(majorVersion: Int, minorVersion: Int) -case class EventManifest(schema: String, isStringManifest: Boolean, eventVersion: Option[EventVersion]) +case class PayloadVersion(majorVersion: Int, minorVersion: Int) +case class EventManifest(schema: String, isStringManifest: Boolean, payloadVersion: Option[PayloadVersion]) case class EventBytes(bytes: Array[Byte], serializerId: Int, manifest: EventManifest) sealed trait DurableEvent { diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala new file mode 100644 index 0000000..ee30945 --- /dev/null +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala @@ -0,0 +1,78 @@ +package com.rbmhtechnology.eventuate.sandbox + +import akka.actor.ActorSystem +import akka.serialization.SerializationExtension +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Block +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Continue +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.ReplicationDecision +import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer +import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer.decode + +import scala.reflect.ClassTag +import scala.reflect._ +import scala.util.Failure +import scala.util.Success +import scala.util.Try + + +object EventCompatibility { + sealed trait IncompatibilityReason + case class MinorIncompatibility(event: EncodedEvent, required: PayloadVersion, supported: PayloadVersion) extends IncompatibilityReason + case class MajorIncompatibility(schema: String, required: PayloadVersion, supported: PayloadVersion) extends IncompatibilityReason + case class FailureOnDeserialization(serializerId: Int, schema: String, cause: Throwable) extends IncompatibilityReason + case class NoSerializer(serializerId: Int) extends IncompatibilityReason + case class NoLocalPayloadVersion(event: EncodedEvent, serializerId: Int) extends IncompatibilityReason + case class NoRemotePayloadVersion(event: EncodedEvent, serializerId: Int) extends IncompatibilityReason + + def eventCompatibility(encoded: EncodedEvent)(implicit system: ActorSystem): Option[IncompatibilityReason]= { + val serializerId = encoded.payload.serializerId + val manifest = encoded.payload.manifest + val compatibility= for { + serializer <- toRight(SerializationExtension(system).serializerByIdentity.get(serializerId), NoSerializer(serializerId)) + event <- toRight(decode(encoded).map(_ => encoded), FailureOnDeserialization(serializerId, manifest.schema, _ : Throwable)) + payloadSerializer <- castOrLeft[EventPayloadSerializer, IncompatibilityReason](serializer, NoLocalPayloadVersion(event, serializerId)) + payloadVersion <- toRight(manifest.payloadVersion, NoRemotePayloadVersion(event, serializerId)) + _ <- compareVersions(event, payloadSerializer.payloadVersion(manifest.schema), payloadVersion) + } yield () + compatibility.left.toOption + } + + private def compareVersions(event: EncodedEvent, supported: PayloadVersion, required: PayloadVersion): Either.RightProjection[IncompatibilityReason, Unit] = { + val res = if(supported.majorVersion < required.majorVersion) + Left(MajorIncompatibility(event.payload.manifest.schema, required, supported)) + else if(supported.majorVersion == required.majorVersion && supported.minorVersion < required.minorVersion) + Left(MinorIncompatibility(event, required, supported)) + else + Right(()) + res.right + } + + private def castOrLeft[A : ClassTag, L](a: AnyRef, left: L): Either.RightProjection[L, A] = + Either.cond(classTag[A].runtimeClass.isAssignableFrom(a.getClass), a.asInstanceOf[A], left).right + + private def toRight[L, R](option: Option[R], left: L): Either.RightProjection[L, R] = + Either.cond(option.isDefined, option.get, left).right + + private def toRight[L, R](t: Try[R], makeLeft: Throwable => L): Either.RightProjection[L, R] = + t match { + case Success(r) => Right(r).right + case Failure(ex) => Left(makeLeft(ex)).right + } + + def eventCompatibilityDecider(decider: IncompatibilityReason => ReplicationDecision)(implicit system: ActorSystem): ReplicationDecider = + new ReplicationDecider { + override def apply(event: EncodedEvent) = + eventCompatibility(event).map(decider).getOrElse(Continue) + } + + case class BlockOnIncompatibility(compatibility: IncompatibilityReason) extends BlockReason + + def stopOnIncompatibility(implicit system: ActorSystem) = eventCompatibilityDecider { + incompatibility => Block(BlockOnIncompatibility(incompatibility)) + } + + def stopOnUnserializableKeepOthers(implicit system: ActorSystem) = eventCompatibilityDecider { + case _: MinorIncompatibility | _: NoRemotePayloadVersion | _: NoLocalPayloadVersion => Continue + case incompatibility => Block(BlockOnIncompatibility(incompatibility)) + } +} diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala index 2283e83..cd393a3 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala @@ -2,8 +2,11 @@ package com.rbmhtechnology.eventuate.sandbox import akka.actor._ import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol._ +import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.stopOnUnserializableKeepOthers import com.rbmhtechnology.eventuate.sandbox.ReplicationFilter.NoFilter +import com.rbmhtechnology.eventuate.sandbox.ReplicationProcessor.ReplicationProcessResult import com.rbmhtechnology.eventuate.sandbox.ReplicationProtocol._ +import com.rbmhtechnology.eventuate.sandbox.ReplicationBlocker.BlockAfter import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer import scala.collection.immutable.Seq @@ -19,7 +22,8 @@ trait EventLogOps { def id: String def sourceFilter: ReplicationFilter - def targetFilter(targetLogId: String): ReplicationFilter + def inboundReplicationProcessor(sourceLogId: String, currentVersionVector: VectorTime): ReplicationProcessor + def outboundReplicationProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int): ReplicationProcessor def sequenceNr: Long = _sequenceNr @@ -27,18 +31,19 @@ trait EventLogOps { def versionVector: VectorTime = _versionVector - def read(fromSquenceNr: Long): Seq[EncodedEvent] = - eventStore.drop(fromSquenceNr.toInt - 1) + def read(fromSequenceNr: Long): Seq[EncodedEvent] = + eventStore.drop(fromSequenceNr.toInt - 1) def causalityFilter(versionVector: VectorTime): ReplicationFilter = new ReplicationFilter { override def apply(event: EncodedEvent): Boolean = !event.before(versionVector) } - def replicationReadFilter(targetLogId: String, targetVersionVector: VectorTime): ReplicationFilter = - causalityFilter(targetVersionVector) and targetFilter(targetLogId) and sourceFilter + def replicationReadFilter(targetFilter: ReplicationFilter, targetVersionVector: VectorTime): ReplicationFilter = + causalityFilter(targetVersionVector) and targetFilter and sourceFilter - def replicationRead(fromSequenceNr: Long, num: Int, targetLogId: String, targetVersionVector: VectorTime): Seq[EncodedEvent] = - read(fromSequenceNr).filter(replicationReadFilter(targetLogId, targetVersionVector).apply).take(num) + def replicationRead(fromSequenceNr: Long, num: Int, targetLogId: String, targetVersionVector: VectorTime): ReplicationProcessResult = + outboundReplicationProcessor(targetLogId, targetVersionVector, num) + .apply(read(fromSequenceNr), fromSequenceNr) def progressRead(logId: String): Long = progressStore.getOrElse(logId, 0L) @@ -46,8 +51,12 @@ trait EventLogOps { def emissionWrite(events: Seq[EncodedEvent]): Seq[EncodedEvent] = write(events, (evt, snr) => evt.emitted(id, snr)) - def replicationWrite(events: Seq[EncodedEvent]): Seq[EncodedEvent] = - write(events.filter(causalityFilter(_versionVector).apply), (evt, snr) => evt.replicated(id, snr)) + def replicationWrite(events: Seq[EncodedEvent], progress: Long, sourceLogId: String): ReplicationProcessResult = { + inboundReplicationProcessor(sourceLogId, versionVector) + .apply(events, progress).right.map { + case (filtered, updatedProgress) => (write(filtered, (evt, snr) => evt.replicated(id, snr)), updatedProgress) + } + } def progressWrite(progresses: Map[String, Long]): Unit = progressStore = progressStore ++ progresses @@ -97,6 +106,9 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto private var targetFilters: Map[String, ReplicationFilter] = Map.empty + private var eventCompatibilityDeciders: Map[String, ReplicationDecider] = + Map.empty + override def receive = { case Subscribe(subscriber) => subscribe(subscriber) @@ -104,25 +116,48 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto val encoded = read(from) sender() ! ReadSuccess(decode(encoded)) case ReplicationRead(from, num, tlid, tvv) => - val encoded = replicationRead(from, num, tlid, tvv) - sender() ! ReplicationReadSuccess(encoded, encoded.lastOption.map(_.metadata.localSequenceNr).getOrElse(from)) + replicationRead(from, num, tlid, tvv) match { + case Right((processedEvents, progress)) => + sender() ! ReplicationReadSuccess(processedEvents, progress) + case Left(reason) => + sender() ! ReplicationReadFailure(new ReplicationStoppedException(reason)) + } case Write(events) => val encoded = emissionWrite(encode(events)) val decoded = encoded.zip(events).map { case (enc, dec) => dec.copy(enc.metadata) } sender() ! WriteSuccess(decoded) publish(decoded) case ReplicationWrite(events, sourceLogId, progress) => - val encoded = replicationWrite(events); progressWrite(Map(sourceLogId -> progress)) - val decoded = decode(encoded) - sender() ! ReplicationWriteSuccess(encoded, sourceLogId, progress, versionVector) - publish(decoded) + replicationWrite(events, progress, sourceLogId) match { + case Right((processedEvents, updatedProgress)) => + progressWrite(Map(sourceLogId -> updatedProgress)) + val decoded = decode(processedEvents) + sender() ! ReplicationWriteSuccess(processedEvents, sourceLogId, progress, versionVector) + publish(decoded) + case Left(reason) => + sender() ! ReplicationWriteFailure(new ReplicationStoppedException(reason)) + } case GetReplicationProgressAndVersionVector(logId) => sender() ! GetReplicationProgressAndVersionVectorSuccess(progressRead(logId), versionVector) case AddTargetFilter(logId, filter) => targetFilters = targetFilters.updated(logId, filter) + case AddEventCompatibilityDecider(sourceLogId, processor) => + eventCompatibilityDeciders += sourceLogId -> processor + case RemoveEventCompatibilityDecider(sourceLogId) => + eventCompatibilityDeciders -= sourceLogId } - def targetFilter(logId: String): ReplicationFilter = + override def inboundReplicationProcessor(sourceLogId: String, currentVersionVector: VectorTime) = + ReplicationProcessor( + ReplicationDecider(causalityFilter(currentVersionVector)) + .andThen(eventCompatibilityDeciders.getOrElse(sourceLogId, stopOnUnserializableKeepOthers))) + + override def outboundReplicationProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int) = + // TODO RFC processor + ReplicationProcessor( + ReplicationDecider(replicationReadFilter(targetFilter(targetLogId), targetVersionVector), new BlockAfter(num))) + + private def targetFilter(logId: String): ReplicationFilter = targetFilters.getOrElse(logId, NoFilter) } @@ -138,4 +173,7 @@ object EventLog { def decode(events: Seq[EncodedEvent])(implicit system: ActorSystem): Seq[DecodedEvent] = events.map(e => EventPayloadSerializer.decode(e).get) + + class ReplicationStoppedException(reason: BlockReason) + extends IllegalStateException(s"Replication stopped: $reason") } diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala new file mode 100644 index 0000000..01e7b4d --- /dev/null +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala @@ -0,0 +1,44 @@ +package com.rbmhtechnology.eventuate.sandbox + +import scala.annotation.tailrec +import scala.collection.immutable.Seq + +trait BlockReason +case class MoreThanN(n: Int) extends BlockReason + +trait ReplicationBlocker { + def apply(event: EncodedEvent): Option[BlockReason] +} + +object ReplicationBlocker { + class SequentialReplicationBlocker(blockers: Seq[ReplicationBlocker]) extends ReplicationBlocker { + override def apply(event: EncodedEvent) = { + @tailrec + def go(blockers: Seq[ReplicationBlocker]): Option[BlockReason] = + blockers match { + case Nil => None + case h :: t => + h(event) match { + case None => go(t) + case reason => reason + } + } + go(blockers) + } + } + + object NoBlocker extends ReplicationBlocker { + override def apply(event: EncodedEvent) = None + } + + class BlockAfter(n: Int) extends ReplicationBlocker { + private var count: Int = 0 + override def apply(event: EncodedEvent) = + if(count > n) + Some(MoreThanN(n)) + else { + count += 1 + None + } + } +} diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationDecider.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationDecider.scala new file mode 100644 index 0000000..cfc195b --- /dev/null +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationDecider.scala @@ -0,0 +1,28 @@ +package com.rbmhtechnology.eventuate.sandbox + +import com.rbmhtechnology.eventuate.sandbox.ReplicationBlocker.NoBlocker +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Continue +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.ReplicationDecision + +object ReplicationDecider { + sealed trait ReplicationDecision + case object Filter extends ReplicationDecision + case class Block(reason: BlockReason) extends ReplicationDecision + case object Continue extends ReplicationDecision + + def apply(replicationFilter: ReplicationFilter, replicationBlocker: ReplicationBlocker = NoBlocker): ReplicationDecider = new ReplicationDecider { + override def apply(event: EncodedEvent) = + if (replicationFilter(event)) replicationBlocker(event).map(Block).getOrElse(Continue) else Filter + } +} + +trait ReplicationDecider { outer => + def apply(event: EncodedEvent): ReplicationDecision + + def andThen(nextDecider: ReplicationDecider) = new ReplicationDecider { + override def apply(event: EncodedEvent) = outer(event) match { + case Continue => nextDecider(event) + case result => result + } + } +} diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala index 686cfc9..48d86a4 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala @@ -5,7 +5,7 @@ import java.util.function.UnaryOperator import akka.actor._ import akka.pattern.{ask, pipe} - +import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.IncompatibilityReason import com.rbmhtechnology.eventuate.sandbox.ReplicationFilter.NoFilter import com.rbmhtechnology.eventuate.sandbox.ReplicationProtocol._ import com.typesafe.config._ @@ -54,8 +54,15 @@ class ReplicationEndpoint( def connect(remoteEndpoint: ReplicationEndpoint): Future[String] = connect(remoteEndpoint.connectionAcceptor) - def connect(remoteAcceptor: ActorRef): Future[String] = + def connect(remoteEndpoint: ReplicationEndpoint, eventCompatibilityDeciders: Map[String, ReplicationDecider]): Future[String] = + connect(remoteEndpoint.connectionAcceptor, eventCompatibilityDeciders) + + def connect(remoteAcceptor: ActorRef, eventCompatibilityDeciders: Map[String, ReplicationDecider] = Map.empty): Future[String] = remoteAcceptor.ask(GetReplicationSourceLogs(logNames))(settings.askTimeout).mapTo[GetReplicationSourceLogsSuccess].map { reply => + eventCompatibilityDeciders.foreach { case (logName, processor) => + eventLogs.get(logName).foreach(_ ! AddEventCompatibilityDecider(logId(reply.endpointId, logName), processor)) + } + //TODO make sure processors are added before replicators are started val replicators = reply.sourceLogs.map { case (logName, sourceLog) => val sourceLogId = logId(reply.endpointId, logName) @@ -66,8 +73,12 @@ class ReplicationEndpoint( reply.endpointId } - def disconnect(remoteEndpointId: String): Unit = + def disconnect(remoteEndpointId: String): Unit = { removeConnection(remoteEndpointId).foreach(system.stop) + eventLogs.foreach { case (logName, eventLog) => + eventLog ! RemoveEventCompatibilityDecider(logId(remoteEndpointId, logName)) + } + } def terminate(): Future[Terminated] = system.terminate() diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProcessor.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProcessor.scala new file mode 100644 index 0000000..623518a --- /dev/null +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProcessor.scala @@ -0,0 +1,35 @@ +package com.rbmhtechnology.eventuate.sandbox + +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Block +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Filter +import com.rbmhtechnology.eventuate.sandbox.ReplicationProcessor.ReplicationProcessResult + +import scala.annotation.tailrec +import scala.collection.immutable.Seq + +object ReplicationProcessor { + type ReplicationProcessResult = Either[BlockReason, (Seq[EncodedEvent], Long)] +} + +case class ReplicationProcessor(replicationDecider: ReplicationDecider) { + + def apply(events: Seq[EncodedEvent], progress: Long): ReplicationProcessResult = { + var lastProgress: Long = 0 + + @tailrec + def go(in: Seq[EncodedEvent], out: Vector[EncodedEvent]): ReplicationProcessResult = in match { + case seq if seq.isEmpty => + Right(out, progress) + case seq => + replicationDecider(seq.head) match { + case Block(reason) => + Either.cond(lastProgress > 0, (out, lastProgress), reason) + case decision => + lastProgress = seq.head.metadata.localSequenceNr + go(seq.tail, if(decision == Filter) out else out :+ seq.head) + } + } + + go(events, Vector.empty) + } +} diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala index 8fd59a8..fd18d6b 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala @@ -5,6 +5,9 @@ import akka.actor.ActorRef import scala.collection.immutable.Seq object ReplicationProtocol { + case class AddEventCompatibilityDecider(sourceLogId: String, decider: ReplicationDecider) + case class RemoveEventCompatibilityDecider(sourceLogId: String) + case class AddTargetFilter(targetLogId: String, filter: ReplicationFilter) case class GetReplicationSourceLogs(logNames: Set[String]) @@ -15,7 +18,9 @@ object ReplicationProtocol { case class ReplicationRead(fromSequenceNr: Long, num: Int, targetLogId: String, targetVersionVector: VectorTime) case class ReplicationReadSuccess(events: Seq[EncodedEvent], progress: Long) + case class ReplicationReadFailure(cause: Throwable) case class ReplicationWrite(events: Seq[EncodedEvent], sourceLogId: String, progress: Long) case class ReplicationWriteSuccess(events: Seq[EncodedEvent], sourceLogId: String, progress: Long, targetVersionVector: VectorTime) + case class ReplicationWriteFailure(cause: Throwable) } diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/serializer/EventPayloadSerializer.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/serializer/EventPayloadSerializer.scala index dab4b6f..a61a175 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/serializer/EventPayloadSerializer.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/serializer/EventPayloadSerializer.scala @@ -8,12 +8,12 @@ import com.rbmhtechnology.eventuate.sandbox.DecodedEvent import com.rbmhtechnology.eventuate.sandbox.EncodedEvent import com.rbmhtechnology.eventuate.sandbox.EventBytes import com.rbmhtechnology.eventuate.sandbox.EventManifest -import com.rbmhtechnology.eventuate.sandbox.EventVersion +import com.rbmhtechnology.eventuate.sandbox.PayloadVersion import scala.util.Try abstract class EventPayloadSerializer extends SerializerWithStringManifest { - def eventVersion(schema: String): EventVersion + def payloadVersion(schema: String): PayloadVersion } object EventPayloadSerializer { @@ -34,7 +34,7 @@ object EventPayloadSerializer { private def eventManifest(serializer: Serializer, payload: AnyRef): EventManifest = { val schema = eventSchema(serializer, payload) - EventManifest(schema, serializer.isInstanceOf[SerializerWithStringManifest], eventVersion(serializer, schema)) + EventManifest(schema, serializer.isInstanceOf[SerializerWithStringManifest], payloadVersion(serializer, schema)) } private def eventSchema(serializer: Serializer, payload: AnyRef): String = @@ -45,9 +45,9 @@ object EventPayloadSerializer { payload.getClass.getName } - private def eventVersion(serializer: Serializer, schema: String): Option[EventVersion] = + private def payloadVersion(serializer: Serializer, schema: String): Option[PayloadVersion] = serializer match { - case payloadSerializer: EventPayloadSerializer => Some(payloadSerializer.eventVersion(schema)) + case payloadSerializer: EventPayloadSerializer => Some(payloadSerializer.payloadVersion(schema)) case _ => None } diff --git a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala new file mode 100644 index 0000000..e304c1d --- /dev/null +++ b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala @@ -0,0 +1,165 @@ +package com.rbmhtechnology.eventuate.sandbox + +import akka.actor.ActorRef +import akka.actor.ActorSystem +import akka.actor.ExtendedActorSystem +import akka.testkit.TestProbe +import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.BlockOnIncompatibility +import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.MajorIncompatibility +import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.MinorIncompatibility +import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.eventCompatibilityDecider +import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol.Subscribe +import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol.Write +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Block +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Continue +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Filter +import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer +import com.typesafe.config.ConfigFactory +import org.scalatest.BeforeAndAfterEach +import org.scalatest.Matchers +import org.scalatest.WordSpec + +import scala.concurrent.duration.DurationInt + +object SchemaEvolutionSpec { + val EmitterId1 = "EM1" + val EmitterId2 = "EM2" + val EndpointId1 = "EP1" + val EndpointId2 = "EP2" + + val LogName = "L" + + def serializerConfig(serializerClass: Class[_]) = + ConfigFactory.parseString( + s""" + |akka.actor { + | serializers { + | test-event = "${serializerClass.getName}" + | } + | serialization-bindings { + | "${classOf[Format].getName}" = test-event + | } + |} + """.stripMargin) + + trait Format extends Serializable + case object CompatibleEvent extends Format + case object MinorIncompatibleEvent extends Format + case object MajorIncompatibleEvent extends Format + case object NoVersionEvent + + + abstract class TestSerializer extends EventPayloadSerializer { + protected val CompatibleEventManifest = CompatibleEvent.toString + protected val MinorIncompatibleEventManifest = MinorIncompatibleEvent.toString + protected val MajorIncompatibleEventManifest = MajorIncompatibleEvent.toString + + override def identifier: Int = 896798 + + override def manifest(o: AnyRef): String = + o.toString + + override def toBinary(o: AnyRef): Array[Byte] = + Array.empty[Byte] + + override def fromBinary(bytes: Array[Byte], manifest: String): AnyRef = + manifest match { + case CompatibleEventManifest => CompatibleEvent + case MinorIncompatibleEventManifest => MinorIncompatibleEvent + case MajorIncompatibleEventManifest => MajorIncompatibleEvent + } + } + + case class TestSerializer1(system: ExtendedActorSystem) extends TestSerializer { + override def payloadVersion(schema: String): PayloadVersion = + schema match { + case CompatibleEventManifest => PayloadVersion(1,1) + case MinorIncompatibleEventManifest => PayloadVersion(1,2) + case MajorIncompatibleEventManifest => PayloadVersion(2,1) + } + } + + case class TestSerializer2(system: ExtendedActorSystem) extends TestSerializer { + override def payloadVersion(schema: String) = + PayloadVersion(1,1) + } + + def stopOnUnexpectedFilterMajorContinueOnMinor(implicit system: ActorSystem): ReplicationDecider = + eventCompatibilityDecider { + case _: MajorIncompatibility => Filter + case _: MinorIncompatibility => Continue + case incompatibility => Block(BlockOnIncompatibility(incompatibility)) + } + + def payloadEquals(payload: AnyRef): PartialFunction[Any, Any] = { + case DecodedEvent(_, actual) if actual == payload => actual + } +} + +class SchemaEvolutionSpec extends WordSpec with Matchers with BeforeAndAfterEach { + + import SchemaEvolutionSpec._ + + private var endpoint1: ReplicationEndpoint = _ + private var endpoint2: ReplicationEndpoint = _ + private var probe1: TestProbe = _ + private var probe2: TestProbe = _ + private var log1: ActorRef = _ + private var log2: ActorRef = _ + + override protected def beforeEach(): Unit = { + endpoint1 = new ReplicationEndpoint(EndpointId1, Set(LogName), Map(), serializerConfig(classOf[TestSerializer1])) + endpoint2 = new ReplicationEndpoint(EndpointId2, Set(LogName), Map(), serializerConfig(classOf[TestSerializer2])) + + probe1 = TestProbe()(endpoint1.system) + probe2 = TestProbe()(endpoint2.system) + + log1 = endpoint1.eventLogs(LogName) + log2 = endpoint2.eventLogs(LogName) + log1 ! Subscribe(probe1.ref) + log2 ! Subscribe(probe2.ref) + + endpoint1.connect(endpoint2, Map(LogName -> stopOnUnexpectedFilterMajorContinueOnMinor(endpoint1.system))) + endpoint2.connect(endpoint1, Map(LogName -> stopOnUnexpectedFilterMajorContinueOnMinor(endpoint2.system))) + } + + override protected def afterEach(): Unit = { + endpoint1.terminate() + endpoint2.terminate() + } + + "ReplicationEndpoint" must { + "replicate event from new to old location based on event compatibility" in { + log1 ! Write(List(DecodedEvent(EmitterId1, MajorIncompatibleEvent))) + // filtered + + log1 ! Write(List(DecodedEvent(EmitterId1, CompatibleEvent))) + probe2.expectMsgPF(hint = CompatibleEvent.toString)(payloadEquals(CompatibleEvent)) + + log1 ! Write(List(DecodedEvent(EmitterId1, MinorIncompatibleEvent))) + probe2.expectMsgPF(hint = MinorIncompatibleEvent.toString)(payloadEquals(MinorIncompatibleEvent)) + + log1 ! Write(List(DecodedEvent(EmitterId1, NoVersionEvent))) + // Blocked + + log1 ! Write(List(DecodedEvent(EmitterId1, CompatibleEvent))) + probe2.expectNoMsg(500.millis) // Still blocked + } + "replicate event from old to new location based on event compatibility" in { + log2 ! Write(List(DecodedEvent(EmitterId2, MajorIncompatibleEvent))) + probe1.expectMsgPF(hint = MajorIncompatibleEvent.toString)(payloadEquals(MajorIncompatibleEvent)) + + log2 ! Write(List(DecodedEvent(EmitterId2, CompatibleEvent))) + probe1.expectMsgPF(hint = CompatibleEvent.toString)(payloadEquals(CompatibleEvent)) + + log2 ! Write(List(DecodedEvent(EmitterId2, MinorIncompatibleEvent))) + probe1.expectMsgPF(hint = MinorIncompatibleEvent.toString)(payloadEquals(MinorIncompatibleEvent)) + + log2 ! Write(List(DecodedEvent(EmitterId2, NoVersionEvent))) + // Blocked + + log2 ! Write(List(DecodedEvent(EmitterId2, CompatibleEvent))) + probe1.expectNoMsg(500.millis) // Still blocked + } + } +}