From e7e1d27151f4ce3374ce43d43bd3d477a588ea10 Mon Sep 17 00:00:00 2001
From: Volker Stampa <Volker.Stampa@delivion.de>
Date: Tue, 13 Dec 2016 09:53:29 +0100
Subject: [PATCH 1/5] Add Rfc logic

- add API for adding redundant filtered connections config
- add RfcBlocker
- rename BlockOnIncompatibility -> Incompatibility
---
 .../sandbox/EventCompatibility.scala          |   6 +-
 .../eventuate/sandbox/EventLog.scala          |  34 ++++--
 .../sandbox/RedundantFilterConfig.scala       |  10 ++
 .../sandbox/ReplicationBlocker.scala          |  19 ++-
 .../sandbox/ReplicationEndpoint.scala         |   4 +-
 .../sandbox/ReplicationProtocol.scala         |   2 +
 .../eventuate/sandbox/RfcBlocker.scala        |  15 +++
 .../eventuate/sandbox/VectorTime.scala        |   3 +
 .../RedundantFilteredConnectionsSpec.scala    | 111 ++++++++++++++++++
 .../sandbox/SchemaEvolutionSpec.scala         |   8 +-
 10 files changed, 180 insertions(+), 32 deletions(-)
 create mode 100644 src/main/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilterConfig.scala
 create mode 100644 src/main/scala/com/rbmhtechnology/eventuate/sandbox/RfcBlocker.scala
 create mode 100644 src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala

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..0afe018 100644
--- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala
+++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala
@@ -7,6 +7,8 @@ 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.ReplicationBlocker.SequentialReplicationBlocker
 import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer
 
 import scala.collection.immutable.Seq
@@ -22,8 +24,8 @@ trait EventLogOps {
   def id: String
 
   def sourceFilter: ReplicationFilter
-  def inboundReplicationProcessor(sourceLogId: String, currentVersionVector: VectorTime): ReplicationProcessor
-  def outboundReplicationProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int): ReplicationProcessor
+  def replicationWriteProcessor(sourceLogId: String, currentVersionVector: VectorTime): ReplicationProcessor
+  def replicationReadProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int): ReplicationProcessor
 
   def sequenceNr: Long =
     _sequenceNr
@@ -42,7 +44,7 @@ trait EventLogOps {
     causalityFilter(targetVersionVector) and targetFilter and sourceFilter
 
   def replicationRead(fromSequenceNr: Long, num: Int, targetLogId: String, targetVersionVector: VectorTime): ReplicationProcessResult =
-    outboundReplicationProcessor(targetLogId, targetVersionVector, num)
+    replicationReadProcessor(targetLogId, targetVersionVector, num)
       .apply(read(fromSequenceNr), fromSequenceNr)
 
   def progressRead(logId: String): Long =
@@ -52,7 +54,7 @@ trait EventLogOps {
     write(events, (evt, snr) => evt.emitted(id, snr))
 
   def replicationWrite(events: Seq[EncodedEvent], progress: Long, sourceLogId: String): ReplicationProcessResult = {
-    inboundReplicationProcessor(sourceLogId, versionVector)
+    replicationWriteProcessor(sourceLogId, versionVector)
       .apply(events, progress).right.map {
         case (filtered, updatedProgress) => (write(filtered, (evt, snr) => evt.replicated(id, snr)), updatedProgress)
       }
@@ -102,13 +104,18 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto
   import EventLog._
   import context.system
 
-  /** Maps target log ids to replication filters */
+  /** Maps target log ids to replication filters used for replication reads */
   private var targetFilters: Map[String, ReplicationFilter] =
     Map.empty
 
+  /** Maps source log ids to [[ReplicationDecider]]s used for replication writes */
   private var eventCompatibilityDeciders: Map[String, ReplicationDecider] =
     Map.empty
 
+  /** Maps target log ids to [[RedundantFilterConfig]]s used to build [[RfcBlocker]]s for replication reads */
+  private var redundantFilterConfigs: Map[String, RedundantFilterConfig] =
+    Map.empty
+
   override def receive = {
     case Subscribe(subscriber) =>
       subscribe(subscriber)
@@ -141,24 +148,25 @@ 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 AddRedundantFilterConfig(logId, config) =>
+      redundantFilterConfigs += logId -> config
     case AddEventCompatibilityDecider(sourceLogId, processor) =>
       eventCompatibilityDeciders += sourceLogId -> processor
     case RemoveEventCompatibilityDecider(sourceLogId) =>
       eventCompatibilityDeciders -= sourceLogId
   }
 
-  override def inboundReplicationProcessor(sourceLogId: String, currentVersionVector: VectorTime) =
+  override def replicationWriteProcessor(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)
+  override def replicationReadProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int) = {
+    val targetFilter = targetFilters.getOrElse(targetLogId, NoFilter)
+    ReplicationProcessor(ReplicationDecider(
+      replicationReadFilter(targetFilter, targetVersionVector),
+      SequentialReplicationBlocker(List(BlockAfter(num), redundantFilterConfigs.get(targetLogId).map(_.rfcBlocker(targetVersionVector)).getOrElse(NoBlocker)))))
+  }
 }
 
 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/ReplicationEndpoint.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
index 48d86a4..54f6094 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._
@@ -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, targetLogName: String, redundantlyConnectedEndpoints: Set[String]): Unit =
+    eventLogs(targetLogName) ! AddRedundantFilterConfig(logId(targetEndpointId, targetLogName), RedundantFilterConfig(targetLogName, redundantlyConnectedEndpoints))
+
   def connect(remoteEndpoint: ReplicationEndpoint): Future[String] =
     connect(remoteEndpoint.connectionAcceptor)
 
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/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
new file mode 100644
index 0000000..d7431cd
--- /dev/null
+++ b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
@@ -0,0 +1,111 @@
+package com.rbmhtechnology.eventuate.sandbox
+
+import akka.actor.ActorRef
+import akka.actor.ActorSystem
+import akka.testkit.TestProbe
+import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol.Subscribe
+import com.rbmhtechnology.eventuate.sandbox.EventsourcingProtocol.Write
+import com.rbmhtechnology.eventuate.sandbox.serializer.EventPayloadSerializer
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.WordSpec
+
+import scala.concurrent.duration.DurationInt
+
+object RedundantFilteredConnectionsSpec {
+  val EmitterIdA = "EM-A"
+  val EmitterIdB1 = "EM-B1"
+  val EmitterIdB2 = "EM-B2"
+  val EndpointIdA = "EP-A"
+  val EndpointIdB1 = "EP-B1"
+  val EndpointIdB2 = "EP-B2"
+
+  val LogName = "L"
+
+  case object InternalEvent
+  case object ExternalEvent
+
+  def replicationFilter(implicit system: ActorSystem): ReplicationFilter = new ReplicationFilter {
+    override def apply(event: EncodedEvent) =
+      EventPayloadSerializer.decode(event).get.payload.isInstanceOf[ExternalEvent.type]
+  }
+
+  def payloadEquals(payload: AnyRef): PartialFunction[Any, Any] = {
+    case DecodedEvent(_, actual) if actual == payload => actual
+  }
+
+  def bidiConnect(endpoint1: ReplicationEndpoint, endpoint2: ReplicationEndpoint): Unit = {
+    endpoint1.connect(endpoint2)
+    endpoint2.connect(endpoint1)
+  }
+
+  def expectPayloads(probe: TestProbe, payloads: AnyRef*): Unit =
+    payloads.foreach { payload =>
+      probe.expectMsgPF(hint = payload.toString)(payloadEquals(payload))
+    }
+}
+
+class RedundantFilteredConnectionsSpec extends WordSpec with BeforeAndAfterEach {
+
+  import RedundantFilteredConnectionsSpec._
+
+  private var endpointA: ReplicationEndpoint = _
+  private var endpointB1: ReplicationEndpoint = _
+  private var endpointB2: ReplicationEndpoint = _
+  private var probeA: TestProbe = _
+  private var probeB1: TestProbe = _
+  private var probeB2: TestProbe = _
+  private var logA: ActorRef = _
+  private var logB1: ActorRef = _
+  private var logB2: ActorRef = _
+
+  override protected def beforeEach(): Unit = {
+    endpointA = new ReplicationEndpoint(EndpointIdA, Set(LogName), Map())
+    endpointB1 = new ReplicationEndpoint(EndpointIdB1, Set(LogName), Map())
+    endpointB2 = new ReplicationEndpoint(EndpointIdB2, Set(LogName), Map())
+
+    probeA = TestProbe()(endpointA.system)
+    probeB1 = TestProbe()(endpointB1.system)
+    probeB2 = TestProbe()(endpointB2.system)
+
+    logA = endpointA.eventLogs(LogName)
+    logB1 = endpointB1.eventLogs(LogName)
+    logB2 = endpointB2.eventLogs(LogName)
+    logA ! Subscribe(probeA.ref)
+    logB1 ! Subscribe(probeB1.ref)
+    logB2 ! Subscribe(probeB2.ref)
+    //     A
+    //   /   \ (RFCs)
+    // B1 --- B2
+    // connection B1 - B2 initially interrupted
+    endpointA.addRedundantFilterConfig(EndpointIdB1, LogName, Set(EndpointIdB1, EndpointIdB2))
+    endpointA.addRedundantFilterConfig(EndpointIdB2, LogName, Set(EndpointIdB1, EndpointIdB2))
+    endpointB1.addTargetFilter(EndpointIdA, LogName, replicationFilter(endpointB1.system))
+    endpointB2.addTargetFilter(EndpointIdA, LogName, replicationFilter(endpointB2.system))
+    bidiConnect(endpointA, endpointB1)
+    bidiConnect(endpointA, endpointB2)
+  }
+
+  override protected def afterEach(): Unit = {
+    endpointA.terminate()
+    endpointB1.terminate()
+    endpointB2.terminate()
+  }
+
+  "ReplicationEndpoint" must {
+    "stop replication over redundant filtered connections on event from replicated application" in {
+      logA ! Write(List(DecodedEvent(EmitterIdA, ExternalEvent)))
+      expectPayloads(probeA, ExternalEvent)
+      expectPayloads(probeB1, ExternalEvent)
+      expectPayloads(probeB2, ExternalEvent)
+
+      logB1 ! Write(List(DecodedEvent(EmitterIdB1, InternalEvent), DecodedEvent(EmitterIdB1, ExternalEvent)))
+      expectPayloads(probeA, ExternalEvent)
+      logA ! Write(List(DecodedEvent(EmitterIdA, ExternalEvent)))
+      probeB2.expectNoMsg(500.millis)
+
+      bidiConnect(endpointB2, endpointB1)
+
+      expectPayloads(probeB2, InternalEvent, ExternalEvent, ExternalEvent)
+    }
+  }
+}
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)

From cb5ae670846ee613e62705301097be015dedcc74 Mon Sep 17 00:00:00 2001
From: Volker Stampa <Volker.Stampa@delivion.de>
Date: Tue, 13 Dec 2016 10:26:14 +0100
Subject: [PATCH 2/5] Reorder definitions in EventLog

Group member definitions by feature. This is basically a first step to
extract separate feature to separate (composable) classes/traits.
---
 .../eventuate/sandbox/EventLog.scala          | 150 +++++++++++-------
 .../sandbox/ReplicationDecider.scala          |  10 +-
 2 files changed, 103 insertions(+), 57 deletions(-)

diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala
index 0afe018..b3cfd53 100644
--- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala
+++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/EventLog.scala
@@ -8,25 +8,20 @@ import com.rbmhtechnology.eventuate.sandbox.ReplicationProcessor.ReplicationProc
 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.ReplicationBlocker.SequentialReplicationBlocker
 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 replicationWriteProcessor(sourceLogId: String, currentVersionVector: VectorTime): ReplicationProcessor
-  def replicationReadProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int): ReplicationProcessor
-
   def sequenceNr: Long =
     _sequenceNr
 
@@ -36,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 =
-    replicationReadProcessor(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 = {
-    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
-
   private def write(events: Seq[EncodedEvent], prepare: (EncodedEvent, Long) => EncodedEvent): Seq[EncodedEvent] = {
     var snr = _sequenceNr
     var cvv = _versionVector
@@ -85,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 {
@@ -104,24 +105,24 @@ class EventLog(val id: String, val sourceFilter: ReplicationFilter) extends Acto
   import EventLog._
   import context.system
 
-  /** Maps target log ids to replication filters used for replication reads */
-  private var targetFilters: Map[String, ReplicationFilter] =
-    Map.empty
-
-  /** Maps source log ids to [[ReplicationDecider]]s used for replication writes */
-  private var eventCompatibilityDeciders: Map[String, ReplicationDecider] =
-    Map.empty
+  /* --- Eventsourcing --- */
 
-  /** Maps target log ids to [[RedundantFilterConfig]]s used to build [[RfcBlocker]]s for replication reads */
-  private var redundantFilterConfigs: Map[String, RedundantFilterConfig] =
-    Map.empty
-
-  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)) =>
@@ -129,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)) =>
@@ -146,26 +142,72 @@ 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 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 replicationReadProcessor(targetLogId: String, targetVersionVector: VectorTime, num: Int) = {
     val targetFilter = targetFilters.getOrElse(targetLogId, NoFilter)
-    ReplicationProcessor(ReplicationDecider(
-      replicationReadFilter(targetFilter, targetVersionVector),
-      SequentialReplicationBlocker(List(BlockAfter(num), redundantFilterConfigs.get(targetLogId).map(_.rfcBlocker(targetVersionVector)).getOrElse(NoBlocker)))))
+    val rfcBlocker = redundantFilterConfigs.get(targetLogId).map(_.rfcBlocker(targetVersionVector)).getOrElse(NoBlocker)
+    ReplicationProcessor(
+      // replication
+      ReplicationDecider(causalityFilter(targetVersionVector))
+      // replication filters
+      .andThen(ReplicationDecider(targetFilter and sourceFilter))
+      // RFC
+      .andThen(ReplicationDecider(rfcBlocker))
+      // replication
+      .andThen(ReplicationDecider(BlockAfter(num))))
   }
 }
 
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)
   }
 }
 

From 82a25cd4f755047962e127c79572c538831225f1 Mon Sep 17 00:00:00 2001
From: Volker Stampa <Volker.Stampa@delivion.de>
Date: Tue, 13 Dec 2016 18:01:32 +0100
Subject: [PATCH 3/5] Add another RFC test

- Make Replicator handle ReplicationReadFailure by scheduling new read
---
 .../sandbox/ReplicationEndpoint.scala         |   3 +
 .../RedundantFilteredConnectionsSpec.scala    | 149 +++++++++++-------
 2 files changed, 94 insertions(+), 58 deletions(-)

diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
index 54f6094..d974dbe 100644
--- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
+++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
@@ -152,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/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
index d7431cd..48eb7c2 100644
--- a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
+++ b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
@@ -5,20 +5,15 @@ import akka.actor.ActorSystem
 import akka.testkit.TestProbe
 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 org.scalatest.BeforeAndAfterEach
 import org.scalatest.WordSpec
 
 import scala.concurrent.duration.DurationInt
+import scala.collection.immutable.Seq
 
 object RedundantFilteredConnectionsSpec {
-  val EmitterIdA = "EM-A"
-  val EmitterIdB1 = "EM-B1"
-  val EmitterIdB2 = "EM-B2"
-  val EndpointIdA = "EP-A"
-  val EndpointIdB1 = "EP-B1"
-  val EndpointIdB2 = "EP-B2"
-
   val LogName = "L"
 
   case object InternalEvent
@@ -38,74 +33,112 @@ object RedundantFilteredConnectionsSpec {
     endpoint2.connect(endpoint1)
   }
 
+  def disconnect(endpoint1: ReplicationEndpoint, endpoint2: ReplicationEndpoint): Unit = {
+    endpoint1.disconnect(endpoint2.id)
+    endpoint2.disconnect(endpoint1.id)
+  }
+
+  def bidiConnect(endpoint1: ReplicationEndpoint, endpoint2: ReplicationEndpoint, outboundFilter1: ReplicationFilter = NoFilter, outboundFilter2: ReplicationFilter = NoFilter, rfcEndpoints1: Set[ReplicationEndpoint]= Set.empty, rfcEndpoints2: Set[ReplicationEndpoint]= Set.empty): Unit = {
+    if(rfcEndpoints1.nonEmpty) endpoint1.addRedundantFilterConfig(endpoint2.id, LogName, rfcEndpoints1.map(_.id))
+    if(rfcEndpoints2.nonEmpty) endpoint2.addRedundantFilterConfig(endpoint1.id, LogName, rfcEndpoints2.map(_.id))
+    if(outboundFilter1 ne NoFilter) endpoint1.addTargetFilter(endpoint2.id, LogName, outboundFilter1)
+    if(outboundFilter2 ne NoFilter) endpoint2.addTargetFilter(endpoint1.id, LogName, outboundFilter2)
+    endpoint1.connect(endpoint2)
+    endpoint2.connect(endpoint1)
+  }
+
   def expectPayloads(probe: TestProbe, payloads: AnyRef*): Unit =
     payloads.foreach { payload =>
-      probe.expectMsgPF(hint = payload.toString)(payloadEquals(payload))
+      probe.expectMsgPF(hint = s"${probe.ref} expects $payload")(payloadEquals(payload))
     }
+
+  def expectPayloads(probes: Seq[TestProbe], payloads: AnyRef*): Unit =
+    probes.foreach(expectPayloads(_, payloads: _*))
+
+  def event(payload: AnyRef): DecodedEvent = DecodedEvent("emitter-id", payload)
 }
 
 class RedundantFilteredConnectionsSpec extends WordSpec with BeforeAndAfterEach {
 
   import RedundantFilteredConnectionsSpec._
 
-  private var endpointA: ReplicationEndpoint = _
-  private var endpointB1: ReplicationEndpoint = _
-  private var endpointB2: ReplicationEndpoint = _
-  private var probeA: TestProbe = _
-  private var probeB1: TestProbe = _
-  private var probeB2: TestProbe = _
-  private var logA: ActorRef = _
-  private var logB1: ActorRef = _
-  private var logB2: ActorRef = _
-
-  override protected def beforeEach(): Unit = {
-    endpointA = new ReplicationEndpoint(EndpointIdA, Set(LogName), Map())
-    endpointB1 = new ReplicationEndpoint(EndpointIdB1, Set(LogName), Map())
-    endpointB2 = new ReplicationEndpoint(EndpointIdB2, Set(LogName), Map())
-
-    probeA = TestProbe()(endpointA.system)
-    probeB1 = TestProbe()(endpointB1.system)
-    probeB2 = TestProbe()(endpointB2.system)
-
-    logA = endpointA.eventLogs(LogName)
-    logB1 = endpointB1.eventLogs(LogName)
-    logB2 = endpointB2.eventLogs(LogName)
-    logA ! Subscribe(probeA.ref)
-    logB1 ! Subscribe(probeB1.ref)
-    logB2 ! Subscribe(probeB2.ref)
-    //     A
-    //   /   \ (RFCs)
-    // B1 --- B2
-    // connection B1 - B2 initially interrupted
-    endpointA.addRedundantFilterConfig(EndpointIdB1, LogName, Set(EndpointIdB1, EndpointIdB2))
-    endpointA.addRedundantFilterConfig(EndpointIdB2, LogName, Set(EndpointIdB1, EndpointIdB2))
-    endpointB1.addTargetFilter(EndpointIdA, LogName, replicationFilter(endpointB1.system))
-    endpointB2.addTargetFilter(EndpointIdA, LogName, replicationFilter(endpointB2.system))
-    bidiConnect(endpointA, endpointB1)
-    bidiConnect(endpointA, endpointB2)
-  }
+  private var systems: List[ActorSystem] = Nil
 
-  override protected def afterEach(): Unit = {
-    endpointA.terminate()
-    endpointB1.terminate()
-    endpointB2.terminate()
+  def newLocations(n: Int): List[(ReplicationEndpoint, TestProbe, ActorRef)] = {
+    val res = (1 to n).toList.map { i =>
+      val endpoint = new ReplicationEndpoint(s"EP-$i", Set(LogName))
+      val probe = TestProbe(s"P-$i")(endpoint.system)
+      val log = endpoint.eventLogs(LogName)
+      log ! Subscribe(probe.ref)
+      (endpoint, probe, log)
+    }
+    systems = res.map(_._1.system)
+    res
   }
 
-  "ReplicationEndpoint" must {
-    "stop replication over redundant filtered connections on event from replicated application" in {
-      logA ! Write(List(DecodedEvent(EmitterIdA, ExternalEvent)))
-      expectPayloads(probeA, ExternalEvent)
-      expectPayloads(probeB1, ExternalEvent)
-      expectPayloads(probeB2, ExternalEvent)
+  override protected def afterEach(): Unit =
+    systems.foreach(_.terminate())
 
-      logB1 ! Write(List(DecodedEvent(EmitterIdB1, InternalEvent), DecodedEvent(EmitterIdB1, ExternalEvent)))
+  "ReplicationEndpoint" must {
+    "stop replication over redundant filtered from A to replicated application B1, B2" in {
+      //    A
+      //   / \ (RFC)
+      // B1 - B2
+      // B1, B2 initially interrupted
+      val (a, probeA, logA) :: (b1, probeB1, logB1) :: (b2, probeB2, logB2) :: Nil = newLocations(3)
+      val redundantConnectionsA = Set(b1, b2)
+      bidiConnect(a, b1, outboundFilter2 = replicationFilter(b1.system), rfcEndpoints1 = redundantConnectionsA)
+      bidiConnect(a, b2, outboundFilter2 = replicationFilter(b2.system), rfcEndpoints1 = redundantConnectionsA)
+
+      logA ! Write(List(event(ExternalEvent)))
+      expectPayloads(List(probeA, probeB1, probeB2), ExternalEvent)
+
+      logB1 ! Write(List(event(InternalEvent), event(ExternalEvent)))
       expectPayloads(probeA, ExternalEvent)
-      logA ! Write(List(DecodedEvent(EmitterIdA, ExternalEvent)))
-      probeB2.expectNoMsg(500.millis)
+      logA ! Write(List(event(ExternalEvent)))
+      probeB2.expectNoMsg(200.millis)
 
-      bidiConnect(endpointB2, endpointB1)
+      bidiConnect(b2, b1)
 
       expectPayloads(probeB2, InternalEvent, ExternalEvent, ExternalEvent)
     }
+    "stop replication over redundant filtered from replicated application A1, A2 to replicated application B1, B2 and vice versa" in {
+      // A1 - A2
+      // | RFC |
+      // B1 - B2
+      // A1, A2 initially interrupted
+      val (a1, probeA1, logA1) :: (a2, probeA2, logA2) :: (b1, probeB1, logB1) :: (b2, probeB2, logB2) :: Nil = newLocations(4)
+      val redundantConnectionsA = Set(b1, b2)
+      val redundantConnectionsB = Set(a1, a2)
+      bidiConnect(
+        endpoint1 = a1, outboundFilter1 = replicationFilter(a1.system), rfcEndpoints1 = redundantConnectionsA,
+        endpoint2 = b1, outboundFilter2 = replicationFilter(b1.system), rfcEndpoints2 = redundantConnectionsB)
+      bidiConnect(
+        endpoint1 = a2, outboundFilter1 = replicationFilter(a2.system), rfcEndpoints1 = redundantConnectionsA,
+        endpoint2 = b2, outboundFilter2 = replicationFilter(b2.system), rfcEndpoints2 = redundantConnectionsB)
+      bidiConnect(b1, b2)
+
+      logB1 ! Write(List(event(ExternalEvent)))
+      expectPayloads(List(probeB1, probeA1, probeB2, probeA2), ExternalEvent)
+      logB2 ! Write(List(event(ExternalEvent)))
+      expectPayloads(List(probeB2, probeA2, probeB1, probeA1), ExternalEvent)
+
+      logA1 ! Write(List(event(ExternalEvent)))
+      expectPayloads(List(probeA1, probeB1, probeB2), ExternalEvent)
+      logB2 ! Write(List(event(ExternalEvent)))
+      expectPayloads(List(probeB2, probeB1, probeA1), ExternalEvent)
+      probeA2.expectNoMsg(200.millis)
+
+      disconnect(b1, b2)
+      bidiConnect(a1, a2)
+      expectPayloads(probeA2, ExternalEvent, ExternalEvent)
+
+      logB2 ! Write(List(event(ExternalEvent)))
+      expectPayloads(List(probeB2, probeA2, probeA1), ExternalEvent)
+      probeB1.expectNoMsg(200.millis)
+
+      bidiConnect(b1, b2)
+      expectPayloads(probeB1, ExternalEvent)
+    }
   }
 }

From 85ca8240fa851db2fdc992862d112e5340b5ec82 Mon Sep 17 00:00:00 2001
From: Volker Stampa <Volker.Stampa@delivion.de>
Date: Wed, 14 Dec 2016 09:30:41 +0100
Subject: [PATCH 4/5] Refactor RFC tests

- introduce Location combining endpoint, probe and log reference
- add id to events
- Locations maintain id-counter for emitted event
---
 .../sandbox/ReplicationEndpoint.scala         |   2 +-
 src/test/resources/application.conf           |   2 +
 .../RedundantFilteredConnectionsSpec.scala    | 136 ++++++++++--------
 3 files changed, 79 insertions(+), 61 deletions(-)
 create mode 100644 src/test/resources/application.conf

diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
index d974dbe..518b0bd 100644
--- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
+++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
@@ -29,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)
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
index 48eb7c2..7752eeb 100644
--- a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
+++ b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
@@ -16,46 +16,70 @@ import scala.collection.immutable.Seq
 object RedundantFilteredConnectionsSpec {
   val LogName = "L"
 
-  case object InternalEvent
-  case object ExternalEvent
+  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.type]
+      EventPayloadSerializer.decode(event).get.payload.isInstanceOf[ExternalEvent]
   }
 
   def payloadEquals(payload: AnyRef): PartialFunction[Any, Any] = {
     case DecodedEvent(_, actual) if actual == payload => actual
   }
 
-  def bidiConnect(endpoint1: ReplicationEndpoint, endpoint2: ReplicationEndpoint): Unit = {
-    endpoint1.connect(endpoint2)
-    endpoint2.connect(endpoint1)
+  def bidiConnect(location1: Location, location2: Location): Unit = {
+    location1.endpoint.connect(location2.endpoint)
+    location2.endpoint.connect(location1.endpoint)
   }
 
-  def disconnect(endpoint1: ReplicationEndpoint, endpoint2: ReplicationEndpoint): Unit = {
-    endpoint1.disconnect(endpoint2.id)
-    endpoint2.disconnect(endpoint1.id)
+  def disconnect(location1: Location, location2: Location): Unit = {
+    location1.endpoint.disconnect(location2.endpoint.id)
+    location2.endpoint.disconnect(location1.endpoint.id)
   }
 
-  def bidiConnect(endpoint1: ReplicationEndpoint, endpoint2: ReplicationEndpoint, outboundFilter1: ReplicationFilter = NoFilter, outboundFilter2: ReplicationFilter = NoFilter, rfcEndpoints1: Set[ReplicationEndpoint]= Set.empty, rfcEndpoints2: Set[ReplicationEndpoint]= Set.empty): Unit = {
-    if(rfcEndpoints1.nonEmpty) endpoint1.addRedundantFilterConfig(endpoint2.id, LogName, rfcEndpoints1.map(_.id))
-    if(rfcEndpoints2.nonEmpty) endpoint2.addRedundantFilterConfig(endpoint1.id, LogName, rfcEndpoints2.map(_.id))
-    if(outboundFilter1 ne NoFilter) endpoint1.addTargetFilter(endpoint2.id, LogName, outboundFilter1)
-    if(outboundFilter2 ne NoFilter) endpoint2.addTargetFilter(endpoint1.id, LogName, outboundFilter2)
-    endpoint1.connect(endpoint2)
-    endpoint2.connect(endpoint1)
+  def bidiConnect(location1: Location, location2: Location, outboundFilter1: ReplicationFilter = NoFilter, outboundFilter2: ReplicationFilter = NoFilter, rfcEndpoints1: Set[Location]= Set.empty, rfcEndpoints2: Set[Location]= Set.empty): Unit = {
+    if(rfcEndpoints1.nonEmpty) location1.endpoint.addRedundantFilterConfig(location2.endpoint.id, LogName, rfcEndpoints1.map(_.endpoint.id))
+    if(rfcEndpoints2.nonEmpty) location2.endpoint.addRedundantFilterConfig(location1.endpoint.id, LogName, rfcEndpoints2.map(_.endpoint.id))
+    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(probe: TestProbe, payloads: AnyRef*): Unit =
-    payloads.foreach { payload =>
-      probe.expectMsgPF(hint = s"${probe.ref} expects $payload")(payloadEquals(payload))
+  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
+    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 {
+        eventCnt += 1
+        _(s"$id.$eventCnt")
+      }
+      log ! Write(payloads.map(DecodedEvent(s"EM-$id", _)))
+      payloads
     }
 
-  def expectPayloads(probes: Seq[TestProbe], payloads: AnyRef*): Unit =
-    probes.foreach(expectPayloads(_, payloads: _*))
+    def emitN(makePayload: String => AnyRef, n: Int = 1): Seq[AnyRef] =
+      emit(List.fill(n)(makePayload): _*)
 
-  def event(payload: AnyRef): DecodedEvent = DecodedEvent("emitter-id", payload)
+    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)
+
+    val filter = replicationFilter(endpoint.system)
+  }
 }
 
 class RedundantFilteredConnectionsSpec extends WordSpec with BeforeAndAfterEach {
@@ -64,15 +88,9 @@ class RedundantFilteredConnectionsSpec extends WordSpec with BeforeAndAfterEach
 
   private var systems: List[ActorSystem] = Nil
 
-  def newLocations(n: Int): List[(ReplicationEndpoint, TestProbe, ActorRef)] = {
-    val res = (1 to n).toList.map { i =>
-      val endpoint = new ReplicationEndpoint(s"EP-$i", Set(LogName))
-      val probe = TestProbe(s"P-$i")(endpoint.system)
-      val log = endpoint.eventLogs(LogName)
-      log ! Subscribe(probe.ref)
-      (endpoint, probe, log)
-    }
-    systems = res.map(_._1.system)
+  def newLocations(ids: String*): List[Location] = {
+    val res = ids.toList.map(i => new Location(i))
+    systems = res.map(_.endpoint.system)
     res
   }
 
@@ -85,60 +103,58 @@ class RedundantFilteredConnectionsSpec extends WordSpec with BeforeAndAfterEach
       //   / \ (RFC)
       // B1 - B2
       // B1, B2 initially interrupted
-      val (a, probeA, logA) :: (b1, probeB1, logB1) :: (b2, probeB2, logB2) :: Nil = newLocations(3)
+      val a :: b1 :: b2 :: Nil = newLocations("A", "B1", "B2")
       val redundantConnectionsA = Set(b1, b2)
-      bidiConnect(a, b1, outboundFilter2 = replicationFilter(b1.system), rfcEndpoints1 = redundantConnectionsA)
-      bidiConnect(a, b2, outboundFilter2 = replicationFilter(b2.system), rfcEndpoints1 = redundantConnectionsA)
+      bidiConnect(a, b1, outboundFilter2 = b1.filter, rfcEndpoints1 = redundantConnectionsA)
+      bidiConnect(a, b2, outboundFilter2 = b2.filter, rfcEndpoints1 = redundantConnectionsA)
+
+      expectPayloads(a.emit(ExternalEvent), a, b1, b2)
 
-      logA ! Write(List(event(ExternalEvent)))
-      expectPayloads(List(probeA, probeB1, probeB2), ExternalEvent)
+      val fromB1 = b1.emit(InternalEvent, ExternalEvent)
+      expectPayloads(fromB1.filter(_.isInstanceOf[ExternalEvent]), a)
 
-      logB1 ! Write(List(event(InternalEvent), event(ExternalEvent)))
-      expectPayloads(probeA, ExternalEvent)
-      logA ! Write(List(event(ExternalEvent)))
-      probeB2.expectNoMsg(200.millis)
+      val fromA = a.emit(ExternalEvent)
+      b2.expectNoMsg()
 
       bidiConnect(b2, b1)
 
-      expectPayloads(probeB2, InternalEvent, ExternalEvent, ExternalEvent)
+      expectPayloads(fromB1 ++ fromA, b2)
     }
     "stop replication over redundant filtered from replicated application A1, A2 to replicated application B1, B2 and vice versa" in {
       // A1 - A2
       // | RFC |
       // B1 - B2
       // A1, A2 initially interrupted
-      val (a1, probeA1, logA1) :: (a2, probeA2, logA2) :: (b1, probeB1, logB1) :: (b2, probeB2, logB2) :: Nil = newLocations(4)
+      val a1 :: a2 :: b1 :: b2 :: Nil = newLocations("A1", "A2", "B1", "B2")
       val redundantConnectionsA = Set(b1, b2)
       val redundantConnectionsB = Set(a1, a2)
       bidiConnect(
-        endpoint1 = a1, outboundFilter1 = replicationFilter(a1.system), rfcEndpoints1 = redundantConnectionsA,
-        endpoint2 = b1, outboundFilter2 = replicationFilter(b1.system), rfcEndpoints2 = redundantConnectionsB)
+        location1 = a1, outboundFilter1 = a1.filter, rfcEndpoints1 = redundantConnectionsA,
+        location2 = b1, outboundFilter2 = b1.filter, rfcEndpoints2 = redundantConnectionsB)
       bidiConnect(
-        endpoint1 = a2, outboundFilter1 = replicationFilter(a2.system), rfcEndpoints1 = redundantConnectionsA,
-        endpoint2 = b2, outboundFilter2 = replicationFilter(b2.system), rfcEndpoints2 = redundantConnectionsB)
+        location1 = a2, outboundFilter1 = a2.filter, rfcEndpoints1 = redundantConnectionsA,
+        location2 = b2, outboundFilter2 = b2.filter, rfcEndpoints2 = redundantConnectionsB)
       bidiConnect(b1, b2)
 
-      logB1 ! Write(List(event(ExternalEvent)))
-      expectPayloads(List(probeB1, probeA1, probeB2, probeA2), ExternalEvent)
-      logB2 ! Write(List(event(ExternalEvent)))
-      expectPayloads(List(probeB2, probeA2, probeB1, probeA1), ExternalEvent)
+      expectPayloads(b1.emit(ExternalEvent), b1, a1, b2, a2)
+      expectPayloads(b2.emit(ExternalEvent), b2, a2, b1, a1)
 
-      logA1 ! Write(List(event(ExternalEvent)))
-      expectPayloads(List(probeA1, probeB1, probeB2), ExternalEvent)
-      logB2 ! Write(List(event(ExternalEvent)))
-      expectPayloads(List(probeB2, probeB1, probeA1), ExternalEvent)
-      probeA2.expectNoMsg(200.millis)
+      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(probeA2, ExternalEvent, ExternalEvent)
+      expectPayloads(fromA1 ++ fromB2, a2)
 
-      logB2 ! Write(List(event(ExternalEvent)))
-      expectPayloads(List(probeB2, probeA2, probeA1), ExternalEvent)
-      probeB1.expectNoMsg(200.millis)
+      val fromB2_2 = b2.emit(ExternalEvent)
+      expectPayloads(fromB2_2, b2, a2, a1)
+      b1.expectNoMsg()
 
       bidiConnect(b1, b2)
-      expectPayloads(probeB1, ExternalEvent)
+      expectPayloads(fromB2_2, b1)
     }
   }
 }

From dc805b825ada3f0c98d2d31211d3d5ef1761525a Mon Sep 17 00:00:00 2001
From: Volker Stampa <Volker.Stampa@delivion.de>
Date: Wed, 14 Dec 2016 16:42:34 +0100
Subject: [PATCH 5/5] Add another RFC test

Complex replication network (matrix) with random disconnects
---
 .../sandbox/ReplicationEndpoint.scala         |   4 +-
 .../RedundantFilteredConnectionsSpec.scala    | 158 ++++++++++++++----
 2 files changed, 131 insertions(+), 31 deletions(-)

diff --git a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
index 518b0bd..06b6136 100644
--- a/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
+++ b/src/main/scala/com/rbmhtechnology/eventuate/sandbox/ReplicationEndpoint.scala
@@ -50,8 +50,8 @@ class ReplicationEndpoint(
   def addTargetFilter(targetEndpointId: String, targetLogName: String, filter: ReplicationFilter): Unit =
     eventLogs(targetLogName) ! AddTargetFilter(logId(targetEndpointId, targetLogName), filter)
 
-  def addRedundantFilterConfig(targetEndpointId: String, targetLogName: String, redundantlyConnectedEndpoints: Set[String]): Unit =
-    eventLogs(targetLogName) ! AddRedundantFilterConfig(logId(targetEndpointId, targetLogName), RedundantFilterConfig(targetLogName, redundantlyConnectedEndpoints))
+  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)
diff --git a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
index 7752eeb..696c3e6 100644
--- a/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
+++ b/src/test/scala/com/rbmhtechnology/eventuate/sandbox/RedundantFilteredConnectionsSpec.scala
@@ -1,19 +1,36 @@
 package com.rbmhtechnology.eventuate.sandbox
 
-import akka.actor.ActorRef
 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)
@@ -38,9 +55,14 @@ object RedundantFilteredConnectionsSpec {
     location2.endpoint.disconnect(location1.endpoint.id)
   }
 
-  def bidiConnect(location1: Location, location2: Location, outboundFilter1: ReplicationFilter = NoFilter, outboundFilter2: ReplicationFilter = NoFilter, rfcEndpoints1: Set[Location]= Set.empty, rfcEndpoints2: Set[Location]= Set.empty): Unit = {
-    if(rfcEndpoints1.nonEmpty) location1.endpoint.addRedundantFilterConfig(location2.endpoint.id, LogName, rfcEndpoints1.map(_.endpoint.id))
-    if(rfcEndpoints2.nonEmpty) location2.endpoint.addRedundantFilterConfig(location1.endpoint.id, LogName, rfcEndpoints2.map(_.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)
@@ -53,23 +75,28 @@ object RedundantFilteredConnectionsSpec {
 
   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 {
+      val payloads = makePayloads.toList.map { makePayload =>
         eventCnt += 1
-        _(s"$id.$eventCnt")
+        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))
@@ -78,20 +105,63 @@ object RedundantFilteredConnectionsSpec {
     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 BeforeAndAfterEach {
+class RedundantFilteredConnectionsSpec extends WordSpec with Matchers with BeforeAndAfterEach with Eventually {
 
   import RedundantFilteredConnectionsSpec._
 
-  private var systems: List[ActorSystem] = Nil
+  implicit override val patienceConfig =
+    PatienceConfig(timeout = Span(RedundantFilteredConnectionsSpec.timeout.duration.toMillis, Millis), interval = Span(100, Millis))
 
-  def newLocations(ids: String*): List[Location] = {
-    val res = ids.toList.map(i => new Location(i))
-    systems = res.map(_.endpoint.system)
-    res
+  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 =
@@ -103,10 +173,10 @@ class RedundantFilteredConnectionsSpec extends WordSpec with BeforeAndAfterEach
       //   / \ (RFC)
       // B1 - B2
       // B1, B2 initially interrupted
-      val a :: b1 :: b2 :: Nil = newLocations("A", "B1", "B2")
+      val Seq(a, b1, b2)= newLocations("A", "B1", "B2")
       val redundantConnectionsA = Set(b1, b2)
-      bidiConnect(a, b1, outboundFilter2 = b1.filter, rfcEndpoints1 = redundantConnectionsA)
-      bidiConnect(a, b2, outboundFilter2 = b2.filter, rfcEndpoints1 = redundantConnectionsA)
+      bidiConnect(a, b1, outboundFilter2 = b1.filter, rfcLocations1 = redundantConnectionsA)
+      bidiConnect(a, b2, outboundFilter2 = b2.filter, rfcLocations1 = redundantConnectionsA)
 
       expectPayloads(a.emit(ExternalEvent), a, b1, b2)
 
@@ -121,20 +191,8 @@ class RedundantFilteredConnectionsSpec extends WordSpec with BeforeAndAfterEach
       expectPayloads(fromB1 ++ fromA, b2)
     }
     "stop replication over redundant filtered from replicated application A1, A2 to replicated application B1, B2 and vice versa" in {
-      // A1 - A2
-      // | RFC |
-      // B1 - B2
-      // A1, A2 initially interrupted
-      val a1 :: a2 :: b1 :: b2 :: Nil = newLocations("A1", "A2", "B1", "B2")
-      val redundantConnectionsA = Set(b1, b2)
-      val redundantConnectionsB = Set(a1, a2)
-      bidiConnect(
-        location1 = a1, outboundFilter1 = a1.filter, rfcEndpoints1 = redundantConnectionsA,
-        location2 = b1, outboundFilter2 = b1.filter, rfcEndpoints2 = redundantConnectionsB)
-      bidiConnect(
-        location1 = a2, outboundFilter1 = a2.filter, rfcEndpoints1 = redundantConnectionsA,
-        location2 = b2, outboundFilter2 = b2.filter, rfcEndpoints2 = redundantConnectionsB)
-      bidiConnect(b1, b2)
+      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)
@@ -156,5 +214,47 @@ class RedundantFilteredConnectionsSpec extends WordSpec with BeforeAndAfterEach
       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 _)
   }
 }