From 7609c88f828aaf096beee73291903bfb0a3faf37 Mon Sep 17 00:00:00 2001 From: Volker Stampa Date: Fri, 9 Dec 2016 13:25:13 +0100 Subject: [PATCH 1/3] Introduce ReplicationProcessor/Stopper ReplicationProcessor combines a ReplicationFilter and a ReplicationStopper. A ReplicationStopper given an event to be replicated decides if the replication shall stop on this event. This decision can be taken on source (e.g. RFC) or target side (e.g. because of schema evolution). The ReplicationProcessor applies filtering and stop-decisions on a sequence of events and returns the remaining events to be replicated (and a modified progress, if required) or a reason why the replication was stopped. --- .../eventuate/sandbox/EventLog.scala | 37 +++++++++++----- .../sandbox/ReplicationProcessor.scala | 38 ++++++++++++++++ .../sandbox/ReplicationProtocol.scala | 2 + .../sandbox/ReplicationStopper.scala | 44 +++++++++++++++++++ 4 files changed, 111 insertions(+), 10 deletions(-) create mode 100644 src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProcessor.scala create mode 100644 src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationStopper.scala diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala index 2283e83..9323491 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala @@ -3,7 +3,9 @@ package com.rbmhtechnology.eventuate.sandbox import akka.actor._ import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol._ 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.ReplicationStopper.StopAfter import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer import scala.collection.immutable.Seq @@ -37,8 +39,8 @@ trait EventLogOps { def replicationReadFilter(targetLogId: String, targetVersionVector: VectorTime): ReplicationFilter = causalityFilter(targetVersionVector) and targetFilter(targetLogId) 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 = + new ReplicationProcessor(replicationReadFilter(targetLogId, targetVersionVector), new StopAfter(num)).apply(read(fromSequenceNr)) def progressRead(logId: String): Long = progressStore.getOrElse(logId, 0L) @@ -46,8 +48,11 @@ 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]): ReplicationProcessResult = { + new ReplicationProcessor(causalityFilter(_versionVector)).apply(events).right.map { + case (filtered, progress) => (write(filtered, (evt, snr) => evt.replicated(id, snr)), progress) + } + } def progressWrite(progresses: Map[String, Long]): Unit = progressStore = progressStore ++ progresses @@ -104,18 +109,27 @@ 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, _)) => + sender() ! ReplicationReadSuccess(processedEvents, processedEvents.lastOption.map(_.metadata.localSequenceNr).getOrElse(from)) + 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) match { + case Right((processedEvents, updatedProgress)) => + progressWrite(Map(sourceLogId -> updatedProgress.getOrElse(progress))) + 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) => @@ -138,4 +152,7 @@ object EventLog { def decode(events: Seq[EncodedEvent])(implicit system: ActorSystem): Seq[DecodedEvent] = events.map(e => EventPayloadSerializer.decode(e).get) + + class ReplicationStoppedException(reason: StopReason) + extends IllegalStateException(s"Replication stopped: $reason") } 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..a095bb6 --- /dev/null +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProcessor.scala @@ -0,0 +1,38 @@ +package com.rbmhtechnology.eventuate.sandbox + +import com.rbmhtechnology.eventuate.sandbox.ReplicationFilter.NoFilter +import com.rbmhtechnology.eventuate.sandbox.ReplicationProcessor.ReplicationProcessResult +import com.rbmhtechnology.eventuate.sandbox.ReplicationStopper.NoStopper + +import scala.collection.immutable.Seq + +object ReplicationProcessor { + type ReplicationProcessResult = Either[StopReason, (Seq[EncodedEvent], Option[Long])] +} + +class ReplicationProcessor(replicationFilter: ReplicationFilter = NoFilter, replicationStopper: ReplicationStopper = NoStopper) { + + def apply(events: Seq[EncodedEvent]): ReplicationProcessResult = { + var stopReason: Option[StopReason] = None + var noEventsFiltered = true + var lastProcessed: Long = 0 + // TODO replace filter by tail recursive function + val filteredEvents = events.filter { event => + stopReason.isEmpty && + { + val passed = replicationFilter(event) + if(!passed) noEventsFiltered = false + lastProcessed = event.metadata.localSequenceNr + passed + } && { + stopReason = replicationStopper(event) + stopReason.isEmpty + } + } + stopReason match { + case None => Right(filteredEvents, None) + case Some(reason) if noEventsFiltered && filteredEvents == Nil => Left(reason) + case _ => Right(filteredEvents, Some(lastProcessed)) + } + } +} diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala index 8fd59a8..f1b27e3 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala @@ -15,7 +15,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/ReplicationStopper.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationStopper.scala new file mode 100644 index 0000000..a7ea06c --- /dev/null +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationStopper.scala @@ -0,0 +1,44 @@ +package com.rbmhtechnology.eventuate.sandbox + +import scala.annotation.tailrec +import scala.collection.immutable.Seq + +sealed trait StopReason +case class MoreThanN(n: Int) extends StopReason + +trait ReplicationStopper { + def apply(event: EncodedEvent): Option[StopReason] +} + +object ReplicationStopper { + class SequentialReplicationStopper(stoppers: Seq[ReplicationStopper]) extends ReplicationStopper { + override def apply(event: EncodedEvent) = { + @tailrec + def go(stoppers: Seq[ReplicationStopper]): Option[StopReason] = + stoppers match { + case Nil => None + case h :: t => + h(event) match { + case None => go(t) + case reason => reason + } + } + go(stoppers) + } + } + + object NoStopper extends ReplicationStopper { + override def apply(event: EncodedEvent) = None + } + + class StopAfter(n: Int) extends ReplicationStopper { + private var count: Int = 0 + override def apply(event: EncodedEvent) = + if(count > n) + Some(MoreThanN(n)) + else { + count += 1 + None + } + } +} From a534910fd8d76cdcc9d2ef771e93daa8dbd0c1f2 Mon Sep 17 00:00:00 2001 From: Volker Stampa Date: Mon, 12 Dec 2016 10:41:51 +0100 Subject: [PATCH 2/3] Add EventCompatibility - this includes some helpers to create ReplicationProcessors working on EventCompatibility - rename ReplicationStopper -> ReplicationBlocker - add inbound/outbound replication processors to EventLog - support adding/removing ReplicationProcessors to EventLog by source log - instead of using ReplicationFilter and ReplicationBlocker one can provide a ReplicationDecider - ReplicationDeciders are compoable with andThen --- .../sandbox/EventCompatibility.scala | 78 +++++++++ .../eventuate/sandbox/EventLog.scala | 55 ++++-- ...Stopper.scala => ReplicationBlocker.scala} | 22 +-- .../sandbox/ReplicationDecider.scala | 28 +++ .../sandbox/ReplicationEndpoint.scala | 17 +- .../sandbox/ReplicationProcessor.scala | 47 +++-- .../sandbox/ReplicationProtocol.scala | 3 + .../sandbox/SchemaEvolutionSpec.scala | 165 ++++++++++++++++++ 8 files changed, 359 insertions(+), 56 deletions(-) create mode 100644 src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala rename src/main/scala/com/rbmhtechnology/eventuate/sandbox/{ReplicationStopper.scala => ReplicationBlocker.scala} (54%) create mode 100644 src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationDecider.scala create mode 100644 src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala 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..dfa0f90 --- /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: EventVersion, supported: EventVersion) extends IncompatibilityReason + case class MajorIncompatibility(schema: String, required: EventVersion, supported: EventVersion) extends IncompatibilityReason + case class FailureOnDeserialization(serializerId: Int, schema: String, cause: Throwable) extends IncompatibilityReason + case class NoSerializer(serializerId: Int) extends IncompatibilityReason + case class NoLocalEventVersion(event: EncodedEvent, serializerId: Int) extends IncompatibilityReason + case class NoRemoteEventVersion(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, NoLocalEventVersion(event, serializerId)) + eventVersion <- toRight(manifest.eventVersion, NoRemoteEventVersion(event, serializerId)) + _ <- compareVersions(event, payloadSerializer.eventVersion(manifest.schema), eventVersion) + } yield () + compatibility.left.toOption + } + + private def compareVersions(event: EncodedEvent, supported: EventVersion, required: EventVersion): 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 | _: NoRemoteEventVersion | _: NoLocalEventVersion => 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 9323491..cd393a3 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala @@ -2,10 +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.ReplicationStopper.StopAfter +import com.rbmhtechnology.eventuate.sandbox.ReplicationBlocker.BlockAfter import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer import scala.collection.immutable.Seq @@ -21,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 @@ -29,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): ReplicationProcessResult = - new ReplicationProcessor(replicationReadFilter(targetLogId, targetVersionVector), new StopAfter(num)).apply(read(fromSequenceNr)) + outboundReplicationProcessor(targetLogId, targetVersionVector, num) + .apply(read(fromSequenceNr), fromSequenceNr) def progressRead(logId: String): Long = progressStore.getOrElse(logId, 0L) @@ -48,10 +51,11 @@ trait EventLogOps { def emissionWrite(events: Seq[EncodedEvent]): Seq[EncodedEvent] = write(events, (evt, snr) => evt.emitted(id, snr)) - def replicationWrite(events: Seq[EncodedEvent]): ReplicationProcessResult = { - new ReplicationProcessor(causalityFilter(_versionVector)).apply(events).right.map { - case (filtered, progress) => (write(filtered, (evt, snr) => evt.replicated(id, snr)), progress) - } + 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 = @@ -102,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) @@ -110,8 +117,8 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto sender() ! ReadSuccess(decode(encoded)) case ReplicationRead(from, num, tlid, tvv) => replicationRead(from, num, tlid, tvv) match { - case Right((processedEvents, _)) => - sender() ! ReplicationReadSuccess(processedEvents, processedEvents.lastOption.map(_.metadata.localSequenceNr).getOrElse(from)) + case Right((processedEvents, progress)) => + sender() ! ReplicationReadSuccess(processedEvents, progress) case Left(reason) => sender() ! ReplicationReadFailure(new ReplicationStoppedException(reason)) } @@ -121,9 +128,9 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto sender() ! WriteSuccess(decoded) publish(decoded) case ReplicationWrite(events, sourceLogId, progress) => - replicationWrite(events) match { + replicationWrite(events, progress, sourceLogId) match { case Right((processedEvents, updatedProgress)) => - progressWrite(Map(sourceLogId -> updatedProgress.getOrElse(progress))) + progressWrite(Map(sourceLogId -> updatedProgress)) val decoded = decode(processedEvents) sender() ! ReplicationWriteSuccess(processedEvents, sourceLogId, progress, versionVector) publish(decoded) @@ -134,9 +141,23 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto 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) } @@ -153,6 +174,6 @@ object EventLog { def decode(events: Seq[EncodedEvent])(implicit system: ActorSystem): Seq[DecodedEvent] = events.map(e => EventPayloadSerializer.decode(e).get) - class ReplicationStoppedException(reason: StopReason) + class ReplicationStoppedException(reason: BlockReason) extends IllegalStateException(s"Replication stopped: $reason") } diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationStopper.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala similarity index 54% rename from src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationStopper.scala rename to src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala index a7ea06c..01e7b4d 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationStopper.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala @@ -3,19 +3,19 @@ package com.rbmhtechnology.eventuate.sandbox import scala.annotation.tailrec import scala.collection.immutable.Seq -sealed trait StopReason -case class MoreThanN(n: Int) extends StopReason +trait BlockReason +case class MoreThanN(n: Int) extends BlockReason -trait ReplicationStopper { - def apply(event: EncodedEvent): Option[StopReason] +trait ReplicationBlocker { + def apply(event: EncodedEvent): Option[BlockReason] } -object ReplicationStopper { - class SequentialReplicationStopper(stoppers: Seq[ReplicationStopper]) extends ReplicationStopper { +object ReplicationBlocker { + class SequentialReplicationBlocker(blockers: Seq[ReplicationBlocker]) extends ReplicationBlocker { override def apply(event: EncodedEvent) = { @tailrec - def go(stoppers: Seq[ReplicationStopper]): Option[StopReason] = - stoppers match { + def go(blockers: Seq[ReplicationBlocker]): Option[BlockReason] = + blockers match { case Nil => None case h :: t => h(event) match { @@ -23,15 +23,15 @@ object ReplicationStopper { case reason => reason } } - go(stoppers) + go(blockers) } } - object NoStopper extends ReplicationStopper { + object NoBlocker extends ReplicationBlocker { override def apply(event: EncodedEvent) = None } - class StopAfter(n: Int) extends ReplicationStopper { + class BlockAfter(n: Int) extends ReplicationBlocker { private var count: Int = 0 override def apply(event: EncodedEvent) = if(count > n) 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 index a095bb6..623518a 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProcessor.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProcessor.scala @@ -1,38 +1,35 @@ package com.rbmhtechnology.eventuate.sandbox -import com.rbmhtechnology.eventuate.sandbox.ReplicationFilter.NoFilter +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Block +import com.rbmhtechnology.eventuate.sandbox.ReplicationDecider.Filter import com.rbmhtechnology.eventuate.sandbox.ReplicationProcessor.ReplicationProcessResult -import com.rbmhtechnology.eventuate.sandbox.ReplicationStopper.NoStopper +import scala.annotation.tailrec import scala.collection.immutable.Seq object ReplicationProcessor { - type ReplicationProcessResult = Either[StopReason, (Seq[EncodedEvent], Option[Long])] + type ReplicationProcessResult = Either[BlockReason, (Seq[EncodedEvent], Long)] } -class ReplicationProcessor(replicationFilter: ReplicationFilter = NoFilter, replicationStopper: ReplicationStopper = NoStopper) { +case class ReplicationProcessor(replicationDecider: ReplicationDecider) { - def apply(events: Seq[EncodedEvent]): ReplicationProcessResult = { - var stopReason: Option[StopReason] = None - var noEventsFiltered = true - var lastProcessed: Long = 0 - // TODO replace filter by tail recursive function - val filteredEvents = events.filter { event => - stopReason.isEmpty && - { - val passed = replicationFilter(event) - if(!passed) noEventsFiltered = false - lastProcessed = event.metadata.localSequenceNr - passed - } && { - stopReason = replicationStopper(event) - stopReason.isEmpty - } - } - stopReason match { - case None => Right(filteredEvents, None) - case Some(reason) if noEventsFiltered && filteredEvents == Nil => Left(reason) - case _ => Right(filteredEvents, Some(lastProcessed)) + 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 f1b27e3..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]) 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..7e2dfc2 --- /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 eventVersion(schema: String): EventVersion = + schema match { + case CompatibleEventManifest => EventVersion(1,1) + case MinorIncompatibleEventManifest => EventVersion(1,2) + case MajorIncompatibleEventManifest => EventVersion(2,1) + } + } + + case class TestSerializer2(system: ExtendedActorSystem) extends TestSerializer { + override def eventVersion(schema: String) = + EventVersion(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 + } + } +} From 8b48b5e106f8f44eb0e4f84bef5e724100fa0bd6 Mon Sep 17 00:00:00 2001 From: Volker Stampa Date: Mon, 12 Dec 2016 14:33:22 +0100 Subject: [PATCH 3/3] Rename EventVersion -> PayloadVersion --- .../eventuate/sandbox/Event.scala | 4 ++-- .../eventuate/sandbox/EventCompatibility.scala | 18 +++++++++--------- .../serializer/EventPayloadSerializer.scala | 10 +++++----- .../sandbox/SchemaEvolutionSpec.scala | 12 ++++++------ 4 files changed, 22 insertions(+), 22 deletions(-) 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 index dfa0f90..ee30945 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala @@ -17,12 +17,12 @@ import scala.util.Try object EventCompatibility { sealed trait IncompatibilityReason - case class MinorIncompatibility(event: EncodedEvent, required: EventVersion, supported: EventVersion) extends IncompatibilityReason - case class MajorIncompatibility(schema: String, required: EventVersion, supported: EventVersion) extends 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 NoLocalEventVersion(event: EncodedEvent, serializerId: Int) extends IncompatibilityReason - case class NoRemoteEventVersion(event: EncodedEvent, 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 @@ -30,14 +30,14 @@ object EventCompatibility { 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, NoLocalEventVersion(event, serializerId)) - eventVersion <- toRight(manifest.eventVersion, NoRemoteEventVersion(event, serializerId)) - _ <- compareVersions(event, payloadSerializer.eventVersion(manifest.schema), eventVersion) + 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: EventVersion, required: EventVersion): Either.RightProjection[IncompatibilityReason, Unit] = { + 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) @@ -72,7 +72,7 @@ object EventCompatibility { } def stopOnUnserializableKeepOthers(implicit system: ActorSystem) = eventCompatibilityDecider { - case _: MinorIncompatibility | _: NoRemoteEventVersion | _: NoLocalEventVersion => Continue + case _: MinorIncompatibility | _: NoRemotePayloadVersion | _: NoLocalPayloadVersion => Continue case incompatibility => Block(BlockOnIncompatibility(incompatibility)) } } 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 index 7e2dfc2..e304c1d 100644 --- a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala +++ b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala @@ -71,17 +71,17 @@ object SchemaEvolutionSpec { } case class TestSerializer1(system: ExtendedActorSystem) extends TestSerializer { - override def eventVersion(schema: String): EventVersion = + override def payloadVersion(schema: String): PayloadVersion = schema match { - case CompatibleEventManifest => EventVersion(1,1) - case MinorIncompatibleEventManifest => EventVersion(1,2) - case MajorIncompatibleEventManifest => EventVersion(2,1) + case CompatibleEventManifest => PayloadVersion(1,1) + case MinorIncompatibleEventManifest => PayloadVersion(1,2) + case MajorIncompatibleEventManifest => PayloadVersion(2,1) } } case class TestSerializer2(system: ExtendedActorSystem) extends TestSerializer { - override def eventVersion(schema: String) = - EventVersion(1,1) + override def payloadVersion(schema: String) = + PayloadVersion(1,1) } def stopOnUnexpectedFilterMajorContinueOnMinor(implicit system: ActorSystem): ReplicationDecider =