diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala index ee30945..831ce1d 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventCompatibility.scala @@ -65,14 +65,14 @@ object EventCompatibility { eventCompatibility(event).map(decider).getOrElse(Continue) } - case class BlockOnIncompatibility(compatibility: IncompatibilityReason) extends BlockReason + case class Incompatible(compatibility: IncompatibilityReason) extends BlockReason def stopOnIncompatibility(implicit system: ActorSystem) = eventCompatibilityDecider { - incompatibility => Block(BlockOnIncompatibility(incompatibility)) + incompatibility => Block(Incompatible(incompatibility)) } def stopOnUnserializableKeepOthers(implicit system: ActorSystem) = eventCompatibilityDecider { case _: MinorIncompatibility | _: NoRemotePayloadVersion | _: NoLocalPayloadVersion => Continue - case incompatibility => Block(BlockOnIncompatibility(incompatibility)) + case incompatibility => Block(Incompatible(incompatibility)) } } diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala index cd393a3..b3cfd53 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala @@ -7,24 +7,21 @@ 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.ReplicationBlocker.NoBlocker import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer import scala.collection.immutable.Seq trait EventLogOps { + // --- EventLog --- private var _sequenceNr: Long = 0L private var _versionVector: VectorTime = VectorTime.Zero private var _deletionVector: VectorTime = VectorTime.Zero var eventStore: Vector[EncodedEvent] = Vector.empty - private var progressStore: Map[String, Long] = Map.empty def id: String - def sourceFilter: ReplicationFilter - def inboundReplicationProcessor(sourceLogId: String, currentVersionVector: VectorTime): ReplicationProcessor - def outboundReplicationProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int): ReplicationProcessor - def sequenceNr: Long = _sequenceNr @@ -34,33 +31,6 @@ trait EventLogOps { 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(targetFilter: ReplicationFilter, targetVersionVector: VectorTime): ReplicationFilter = - causalityFilter(targetVersionVector) and targetFilter and sourceFilter - - 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) - - def emissionWrite(events: Seq[EncodedEvent]): Seq[EncodedEvent] = - write(events, (evt, snr) => evt.emitted(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 - private def write(events: Seq[EncodedEvent], prepare: (EncodedEvent, Long) => EncodedEvent): Seq[EncodedEvent] = { var snr = _sequenceNr var cvv = _versionVector @@ -83,6 +53,39 @@ trait EventLogOps { written } + + // --- Eventsourcing --- + + def emissionWrite(events: Seq[EncodedEvent]): Seq[EncodedEvent] = + write(events, (evt, snr) => evt.emitted(id, snr)) + + // --- Replication --- + + private var progressStore: Map[String, Long] = Map.empty + + def replicationWriteProcessor(sourceLogId: String, currentVersionVector: VectorTime): ReplicationProcessor + def replicationReadProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int): ReplicationProcessor + + def replicationRead(fromSequenceNr: Long, num: Int, targetLogId: String, targetVersionVector: VectorTime): ReplicationProcessResult = + replicationReadProcessor(targetLogId, targetVersionVector, num) + .apply(read(fromSequenceNr), fromSequenceNr) + + def causalityFilter(versionVector: VectorTime): ReplicationFilter = new ReplicationFilter { + override def apply(event: EncodedEvent): Boolean = !event.before(versionVector) + } + + def replicationWrite(events: Seq[EncodedEvent], progress: Long, sourceLogId: String): ReplicationProcessResult = { + replicationWriteProcessor(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 + + def progressRead(logId: String): Long = + progressStore.getOrElse(logId, 0L) } trait EventSubscribers { @@ -102,19 +105,24 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto import EventLog._ import context.system - /** Maps target log ids to replication filters */ - private var targetFilters: Map[String, ReplicationFilter] = - Map.empty - - private var eventCompatibilityDeciders: Map[String, ReplicationDecider] = - Map.empty + /* --- Eventsourcing --- */ - override def receive = { + private def eventsourcingReceive: Receive = { case Subscribe(subscriber) => subscribe(subscriber) case Read(from) => val encoded = read(from) sender() ! ReadSuccess(decode(encoded)) + 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) + } + + /* --- Replication --- */ + + private def replicationReceive: Receive = { case ReplicationRead(from, num, tlid, tvv) => replicationRead(from, num, tlid, tvv) match { case Right((processedEvents, progress)) => @@ -122,11 +130,6 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto 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) => replicationWrite(events, progress, sourceLogId) match { case Right((processedEvents, updatedProgress)) => @@ -139,26 +142,73 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto } case GetReplicationProgressAndVersionVector(logId) => sender() ! GetReplicationProgressAndVersionVectorSuccess(progressRead(logId), versionVector) + } + + /* --- Replication Filters --- */ + + /** Maps target log ids to replication filters used for replication reads */ + private var targetFilters: Map[String, ReplicationFilter] = + Map.empty + + private def replicationFilterReceive: Receive = { case AddTargetFilter(logId, filter) => targetFilters = targetFilters.updated(logId, filter) + } + + /* --- RFC --- */ + + /** Maps target log ids to [[RedundantFilterConfig]]s used to build [[RfcBlocker]]s for replication reads */ + private var redundantFilterConfigs: Map[String, RedundantFilterConfig] = + Map.empty + + + private def rfcReceive: Receive = { + case AddRedundantFilterConfig(logId, config) => + redundantFilterConfigs += logId -> config + } + + /* --- Scheme evolution --- */ + + /** Maps source log ids to [[ReplicationDecider]]s used for replication writes */ + private var eventCompatibilityDeciders: Map[String, ReplicationDecider] = + Map.empty + + private def schemaEvolutionReceive: Receive = { case AddEventCompatibilityDecider(sourceLogId, processor) => eventCompatibilityDeciders += sourceLogId -> processor case RemoveEventCompatibilityDecider(sourceLogId) => eventCompatibilityDeciders -= sourceLogId } - override def inboundReplicationProcessor(sourceLogId: String, currentVersionVector: VectorTime) = + override def receive: Receive = + eventsourcingReceive orElse + replicationReceive orElse + replicationFilterReceive orElse + rfcReceive orElse + schemaEvolutionReceive + + /* --- Replication processors --- */ + + override def replicationWriteProcessor(sourceLogId: String, currentVersionVector: VectorTime) = ReplicationProcessor( + // replication ReplicationDecider(causalityFilter(currentVersionVector)) - .andThen(eventCompatibilityDeciders.getOrElse(sourceLogId, stopOnUnserializableKeepOthers))) + // schema evolution + .andThen(eventCompatibilityDeciders.getOrElse(sourceLogId, stopOnUnserializableKeepOthers))) - override def outboundReplicationProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int) = - // TODO RFC processor + override def replicationReadProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int) = { + val targetFilter = targetFilters.getOrElse(targetLogId, NoFilter) + val rfcBlocker = redundantFilterConfigs.get(targetLogId).map(_.rfcBlocker(targetVersionVector)).getOrElse(NoBlocker) ReplicationProcessor( - ReplicationDecider(replicationReadFilter(targetFilter(targetLogId), targetVersionVector), new BlockAfter(num))) - - private def targetFilter(logId: String): ReplicationFilter = - targetFilters.getOrElse(logId, NoFilter) + // replication + ReplicationDecider(causalityFilter(targetVersionVector)) + // replication filters + .andThen(ReplicationDecider(targetFilter and sourceFilter)) + // RFC + .andThen(ReplicationDecider(rfcBlocker)) + // replication + .andThen(ReplicationDecider(BlockAfter(num)))) + } } object EventLog { diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilterConfig.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilterConfig.scala new file mode 100644 index 0000000..435d35e --- /dev/null +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilterConfig.scala @@ -0,0 +1,10 @@ +package com.rbmhtechnology.eventuate.sandbox + +import com.rbmhtechnology.eventuate.sandbox.ReplicationBlocker.NoBlocker +import com.rbmhtechnology.eventuate.sandbox.ReplicationEndpoint.logId + +case class RedundantFilterConfig(logName: String, endpointIds: Set[String] = Set.empty, foreign: Boolean = true) { + def rfcBlocker(targetVersionVector: VectorTime) = + if(endpointIds.isEmpty) NoBlocker + else RfcBlocker(targetVersionVector, endpointIds.map(logId(_, logName)), !foreign) +} diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala index 01e7b4d..3433d8a 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationBlocker.scala @@ -11,18 +11,15 @@ trait ReplicationBlocker { } object ReplicationBlocker { - class SequentialReplicationBlocker(blockers: Seq[ReplicationBlocker]) extends ReplicationBlocker { + case 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 - } - } + def go(blockers: Seq[ReplicationBlocker]): Option[BlockReason] = blockers match { + case Nil => None + case h :: t => + val reason = h(event) + if(reason.isDefined) reason else go(t) // getOrElse violates tailrec + } go(blockers) } } @@ -31,7 +28,7 @@ object ReplicationBlocker { override def apply(event: EncodedEvent) = None } - class BlockAfter(n: Int) extends ReplicationBlocker { + case 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 index cfc195b..d6ea287 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationDecider.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationDecider.scala @@ -1,6 +1,5 @@ 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 @@ -10,9 +9,14 @@ object ReplicationDecider { case class Block(reason: BlockReason) extends ReplicationDecision case object Continue extends ReplicationDecision - def apply(replicationFilter: ReplicationFilter, replicationBlocker: ReplicationBlocker = NoBlocker): ReplicationDecider = new ReplicationDecider { + def apply(filter: ReplicationFilter): ReplicationDecider = new ReplicationDecider { override def apply(event: EncodedEvent) = - if (replicationFilter(event)) replicationBlocker(event).map(Block).getOrElse(Continue) else Filter + if (filter(event)) Continue else Filter + } + + def apply(blocker: ReplicationBlocker): ReplicationDecider = new ReplicationDecider { + override def apply(event: EncodedEvent) = + blocker(event).map(Block).getOrElse(Continue) } } diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala index 48d86a4..06b6136 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala @@ -5,7 +5,6 @@ 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._ @@ -30,7 +29,7 @@ class ReplicationEndpoint( new AtomicReference(Map.empty) val system: ActorSystem = - ActorSystem(s"$id-system", config) + ActorSystem(s"$id-system", config.withFallback(ConfigFactory.load())) val settings: ReplicationSettings = new ReplicationSettings(system.settings.config) @@ -51,6 +50,9 @@ class ReplicationEndpoint( def addTargetFilter(targetEndpointId: String, targetLogName: String, filter: ReplicationFilter): Unit = eventLogs(targetLogName) ! AddTargetFilter(logId(targetEndpointId, targetLogName), filter) + def addRedundantFilterConfig(targetEndpointId: String, config: RedundantFilterConfig): Unit = + eventLogs(config.logName) ! AddRedundantFilterConfig(logId(targetEndpointId, config.logName), config) + def connect(remoteEndpoint: ReplicationEndpoint): Future[String] = connect(remoteEndpoint.connectionAcceptor) @@ -150,6 +152,9 @@ private class Replicator(sourceLogId: String, sourceLog: ActorRef, targetLogId: case ReplicationReadSuccess(events, progress) => context.become(writing) write(events, progress) + case ReplicationReadFailure(cause) => + context.become(idle) + scheduleRead() } val writing: Receive = { diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala index fd18d6b..7eeeb20 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationProtocol.scala @@ -10,6 +10,8 @@ object ReplicationProtocol { case class AddTargetFilter(targetLogId: String, filter: ReplicationFilter) + case class AddRedundantFilterConfig(targetLogId: String, config: RedundantFilterConfig) + case class GetReplicationSourceLogs(logNames: Set[String]) case class GetReplicationSourceLogsSuccess(endpointId: String, sourceLogs: Map[String, ActorRef]) diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/RfcBlocker.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/RfcBlocker.scala new file mode 100644 index 0000000..528186f --- /dev/null +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/RfcBlocker.scala @@ -0,0 +1,15 @@ +package com.rbmhtechnology.eventuate.sandbox + +import com.rbmhtechnology.eventuate.sandbox.RfcBlocker.RfcConditionViolated + +object RfcBlocker { + case class RfcConditionViolated(eventTime: VectorTime, targetVersionVector: VectorTime, projectionProcessIds: Set[String], negateProjection: Boolean) extends BlockReason +} + +case class RfcBlocker(targetVersionVector: VectorTime, processIds: Set[String], negateProjection: Boolean) extends ReplicationBlocker { + def apply(event: EncodedEvent): Option[BlockReason] = + if(event.metadata.vectorTimestamp.projection(processIds, negateProjection) <= targetVersionVector) + None + else + Some(RfcConditionViolated(event.metadata.vectorTimestamp, targetVersionVector, processIds, negateProjection)) +} diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/VectorTime.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/VectorTime.scala index 17d1df3..9e13773 100644 --- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/VectorTime.scala +++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/VectorTime.scala @@ -60,6 +60,9 @@ case class VectorTime(value: Map[String, Long] = Map.empty) { def merge(that: VectorTime): VectorTime = copy(value.unionWith(that.value)(math.max)) + def projection(processIds: Set[String], negate: Boolean = false): VectorTime = + copy(value.filterKeys(p => processIds.contains(p) != negate).view.force) + /** * Returns `true` if this vector time is equivalent (equal) to `that`. */ diff --git a/src/test/resources/application.conf b/src/test/resources/application.conf new file mode 100644 index 0000000..3116700 --- /dev/null +++ b/src/test/resources/application.conf @@ -0,0 +1,2 @@ +akka.actor.warn-about-java-serializer-usage = off +akka.log-dead-letters = off \ No newline at end of file diff --git a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala new file mode 100644 index 0000000..696c3e6 --- /dev/null +++ b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala @@ -0,0 +1,260 @@ +package com.rbmhtechnology.eventuate.sandbox + +import akka.actor.ActorSystem +import akka.pattern.ask +import akka.testkit.TestProbe +import akka.util.Timeout +import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol.Read +import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol.ReadSuccess +import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol.Subscribe +import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol.Write +import com.rbmhtechnology.eventuate.sandbox.ReplicationFilter.NoFilter +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 org.scalatest.concurrent.Eventually +import org.scalatest.time.Millis +import org.scalatest.time.Span + +import scala.concurrent.duration.DurationInt +import scala.concurrent.ExecutionContext.Implicits.global +import scala.collection.immutable.Seq +import scala.concurrent.Await +import scala.util.Random + +object RedundantFilteredConnectionsSpec { + private val settings = + new ReplicationSettings(ConfigFactory.load()) + + implicit val timeout = + Timeout(settings.askTimeout) + + val LogName = "L" + + case class InternalEvent(s: String) + case class ExternalEvent(s: String) + + def replicationFilter(implicit system: ActorSystem): ReplicationFilter = new ReplicationFilter { + override def apply(event: EncodedEvent) = + EventPayloadSerializer.decode(event).get.payload.isInstanceOf[ExternalEvent] + } + + def payloadEquals(payload: AnyRef): PartialFunction[Any, Any] = { + case DecodedEvent(_, actual) if actual == payload => actual + } + + def bidiConnect(location1: Location, location2: Location): Unit = { + location1.endpoint.connect(location2.endpoint) + location2.endpoint.connect(location1.endpoint) + } + + def disconnect(location1: Location, location2: Location): Unit = { + location1.endpoint.disconnect(location2.endpoint.id) + location2.endpoint.disconnect(location1.endpoint.id) + } + + def bidiConnect( + location1: Location, location2: Location, + outboundFilter1: ReplicationFilter = NoFilter, outboundFilter2: ReplicationFilter = NoFilter, + rfcLocations1: Set[Location]= Set.empty, negate1: Boolean = false, + rfcLocations2: Set[Location]= Set.empty, negate2: Boolean = false + ): Unit = { + if(rfcLocations1.nonEmpty) location1.endpoint.addRedundantFilterConfig(location2.endpoint.id, RedundantFilterConfig(LogName, rfcLocations1.map(_.endpoint.id), !negate1)) + if(rfcLocations2.nonEmpty) location2.endpoint.addRedundantFilterConfig(location1.endpoint.id, RedundantFilterConfig(LogName, rfcLocations2.map(_.endpoint.id), !negate2)) + if(outboundFilter1 ne NoFilter) location1.endpoint.addTargetFilter(location2.endpoint.id, LogName, outboundFilter1) + if(outboundFilter2 ne NoFilter) location2.endpoint.addTargetFilter(location1.endpoint.id, LogName, outboundFilter2) + bidiConnect(location1, location2) + } + + def expectPayloads(payloads: Seq[AnyRef], locs: Location*) = + locs.foreach(_.expectPayloads(payloads)) + + def event(payload: AnyRef): DecodedEvent = DecodedEvent("emitter-id", payload) + + class Location(id: String) { + var eventCnt = 0 + var emitted: List[AnyRef] = Nil + val endpoint = new ReplicationEndpoint(s"EP-$id", Set(LogName)) + val probe = TestProbe(s"P-$id")(endpoint.system) + val log = endpoint.eventLogs(LogName) + log ! Subscribe(probe.ref) + + def emit(makePayloads: Function1[String, AnyRef]*): Seq[AnyRef] = { + val payloads = makePayloads.toList.map { makePayload => + eventCnt += 1 + makePayload(s"$id.$eventCnt") + } + log ! Write(payloads.map(DecodedEvent(s"EM-$id", _))) + emitted = payloads.reverse ::: emitted + payloads + } + + def emitN(makePayload: String => AnyRef, n: Int = 1): Seq[AnyRef] = + emit(List.fill(n)(makePayload): _*) + + def emittedInternal = emitted.filter(_.isInstanceOf[InternalEvent]) + def emittedExternal = emitted.filter(_.isInstanceOf[ExternalEvent]) + + def expectPayloads(payloads: Seq[AnyRef]): Unit = + payloads.foreach { payload => + probe.expectMsgPF(hint = s"${probe.ref} expects $payload")(payloadEquals(payload)) + } + + def expectNoMsg(): Unit = + probe.expectNoMsg(200.millis) + + def storedPayloads: Seq[AnyRef] = + Await.result(log.ask(Read(0)).mapTo[ReadSuccess].map(_.events.map(_.payload)), timeout.duration) + + val filter = replicationFilter(endpoint.system) + + override def toString = s"Loc:$id" + } + + def locationMatrix(applicationNames: Seq[String], nReplicas: Int): Seq[Seq[Location]] = { + val applications = applicationNames.map { applicationName => + (1 to nReplicas).map(replica => new Location(applicationName + replica)) + } + // unfiltered connections between replicas of an application + applications.foreach { application => + application.sliding(2).foreach(connected => bidiConnect(connected.head, connected.last)) + } + // filtered connections between applications + var rfcLocations = Set.empty[Location] + for { + Seq(app1, app2) <- applications.sliding(2) + (location1, location2) <- app1 zip app2 + } { + rfcLocations ++= app1 + bidiConnect( + location1 = location1, outboundFilter1 = location1.filter, rfcLocations1 = rfcLocations, negate1 = true, + location2 = location2, outboundFilter2 = location2.filter, rfcLocations2 = rfcLocations) + } + applications + } + + def randomDisconnects(disconnected: Vector[(Location, Location)], applications: Seq[Seq[Location]]): Vector[(Location, Location)] = { + val Seq(loc1, loc2) = Random.shuffle(Random.shuffle(applications).head.sliding(2).toList).head + disconnect(loc1, loc2) + val updated = disconnected.filterNot(_ == (loc1, loc2)) :+ (loc1, loc2) + if(updated.size >= (applications.head.size - 1) * applications.size / 3) { + bidiConnect _ tupled updated.head + updated.tail + } else + updated + } +} + +class RedundantFilteredConnectionsSpec extends WordSpec with Matchers with BeforeAndAfterEach with Eventually { + + import RedundantFilteredConnectionsSpec._ + + implicit override val patienceConfig = + PatienceConfig(timeout = Span(RedundantFilteredConnectionsSpec.timeout.duration.toMillis, Millis), interval = Span(100, Millis)) + + private var systems: Seq[ActorSystem] = Nil + + def newLocations(ids: String*): Seq[Location] = + registerLocations(ids.toList.map(i => new Location(i))) + + def registerLocations(locations: Seq[Location]): Seq[Location] = { + systems = locations.map(_.endpoint.system) + locations + } + + override protected def afterEach(): Unit = + systems.foreach(_.terminate()) + + "ReplicationEndpoint" must { + "stop replication over redundant filtered from A to replicated application B1, B2" in { + // A + // / \ (RFC) + // B1 - B2 + // B1, B2 initially interrupted + val Seq(a, b1, b2)= newLocations("A", "B1", "B2") + val redundantConnectionsA = Set(b1, b2) + bidiConnect(a, b1, outboundFilter2 = b1.filter, rfcLocations1 = redundantConnectionsA) + bidiConnect(a, b2, outboundFilter2 = b2.filter, rfcLocations1 = redundantConnectionsA) + + expectPayloads(a.emit(ExternalEvent), a, b1, b2) + + val fromB1 = b1.emit(InternalEvent, ExternalEvent) + expectPayloads(fromB1.filter(_.isInstanceOf[ExternalEvent]), a) + + val fromA = a.emit(ExternalEvent) + b2.expectNoMsg() + + bidiConnect(b2, b1) + + expectPayloads(fromB1 ++ fromA, b2) + } + "stop replication over redundant filtered from replicated application A1, A2 to replicated application B1, B2 and vice versa" in { + val Seq(a1, a2, b1, b2) = registerLocations(locationMatrix(List("A", "B"), 2).flatten) + disconnect(a1, a2) + + expectPayloads(b1.emit(ExternalEvent), b1, a1, b2, a2) + expectPayloads(b2.emit(ExternalEvent), b2, a2, b1, a1) + + val fromA1 = a1.emit(ExternalEvent) + expectPayloads(fromA1, a1, b1, b2) + val fromB2 = b2.emit(ExternalEvent) + expectPayloads(fromB2, b2, b1, a1) + a2.expectNoMsg() + + disconnect(b1, b2) + bidiConnect(a1, a2) + expectPayloads(fromA1 ++ fromB2, a2) + + val fromB2_2 = b2.emit(ExternalEvent) + expectPayloads(fromB2_2, b2, a2, a1) + b1.expectNoMsg() + + bidiConnect(b1, b2) + expectPayloads(fromB2_2, b1) + } + "replicate events properly over multiple RFC" in { + // RFC RFC + // A1 - B1 - C1 ... + // | | | + // A2 - B2 - ... + // | | + // A3 - ... + // ... + val applications = locationMatrix(List("A", "B", "C", "D"), 4) + val locations = registerLocations(applications.flatten) + + for { + _ <- 1 to 20 + location <- locations + } location.emit(List.fill(6)(List(ExternalEvent, InternalEvent)).flatten: _*) + + awaitEventDistributionWithRandomDisconnects(applications) + + val allExternal = locations.flatMap(_.emittedExternal).toSet + for { + application <- applications + allApplicationInternal = application.flatMap(_.emittedInternal).toSet + replica <- application + } eventually { + replica.storedPayloads.toSet shouldBe allExternal ++ allApplicationInternal + } + } + } + + def awaitEventDistributionWithRandomDisconnects(applications: Seq[Seq[Location]]) = { + val locations = applications.flatten + val lastEmitted = locations.last.emittedExternal.head + var disconnected = Vector.empty[(Location, Location)] + var i = 0 + locations.head.probe.fishForMessage(hint = s"${locations.head.endpoint.id} fish $lastEmitted", max = RedundantFilteredConnectionsSpec.timeout.duration) { + case ev: DecodedEvent if ev.payload == lastEmitted => true + case _ => + i += 1 + if (i % 10 == 0) disconnected = randomDisconnects(disconnected, applications) + false + } + disconnected.foreach(bidiConnect _ tupled _) + } +} diff --git a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala index e304c1d..a5b7785 100644 --- a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala +++ b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/SchemaEvolutionSpec.scala @@ -4,7 +4,7 @@ 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.Incompatible import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.MajorIncompatibility import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.MinorIncompatibility import com.rbmhtechnology.eventuate.sandbox.EventCompatibility.eventCompatibilityDecider @@ -88,7 +88,7 @@ object SchemaEvolutionSpec { eventCompatibilityDecider { case _: MajorIncompatibility => Filter case _: MinorIncompatibility => Continue - case incompatibility => Block(BlockOnIncompatibility(incompatibility)) + case incompatibility => Block(Incompatible(incompatibility)) } def payloadEquals(payload: AnyRef): PartialFunction[Any, Any] = { @@ -108,8 +108,8 @@ class SchemaEvolutionSpec extends WordSpec with Matchers with BeforeAndAfterEach 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])) + endpoint1 = new ReplicationEndpoint(EndpointId1, Set(LogName), config = serializerConfig(classOf[TestSerializer1])) + endpoint2 = new ReplicationEndpoint(EndpointId2, Set(LogName), config = serializerConfig(classOf[TestSerializer2])) probe1 = TestProbe()(endpoint1.system) probe2 = TestProbe()(endpoint2.system)