Initial consumer filter for edge replication

akka · Dec 8, 2023 · 6e552a5 · 6e552a5
1 parent e358cac
commit 6e552a5
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/...ction-grpc/src/main/scala/akka/projection/grpc/replication/internal/ReplicationImpl.scala b/...ction-grpc/src/main/scala/akka/projection/grpc/replication/internal/ReplicationImpl.scala
@@ -285,6 +285,10 @@ private[akka] object ReplicationImpl {
     val sharding = ClusterSharding(system)
     sharding.init(replicatedEntity.entity)
 
+    if (settings.initialConsumerFilter.nonEmpty) {
+      ConsumerFilter(system).ref ! ConsumerFilter.UpdateFilter(settings.streamId, settings.initialConsumerFilter)
+    }
+
     // sharded daemon process for consuming event stream from the other replicas
     val shardingEntityRefFactory: String => EntityRef[Command] =
       sharding.entityRefFor(replicatedEntity.entity.typeKey, _)