Query plugin paramater in source providers (#684)

* power user API when the query plugin instance has custom configuration (or programatic setup)
akka · Sep 9, 2022 · f404ec5 · f404ec5
1 parent a8953b1
commit f404ec5
Show file tree

Hide file tree

Showing 4 changed files with 57 additions and 13 deletions.
diff --git a/...rable-state/src/main/scala/akka/projection/state/javadsl/DurableStateSourceProvider.scala b/...rable-state/src/main/scala/akka/projection/state/javadsl/DurableStateSourceProvider.scala
@@ -36,14 +36,21 @@ import akka.stream.javadsl.Source
  */
 @ApiMayChange
 object DurableStateSourceProvider {
+
   def changesByTag[A](
       system: ActorSystem[_],
       pluginId: String,
       tag: String): SourceProvider[Offset, DurableStateChange[A]] = {
     val durableStateStoreQuery =
       DurableStateStoreRegistry(system)
         .getDurableStateStoreFor[DurableStateStoreQuery[A]](classOf[DurableStateStoreQuery[A]], pluginId)
+    changesByTag(system, durableStateStoreQuery, tag)
+  }
 
+  def changesByTag[A](
+      system: ActorSystem[_],
+      durableStateStoreQuery: DurableStateStoreQuery[A],
+      tag: String): SourceProvider[Offset, DurableStateChange[A]] = {
     new DurableStateStoreQuerySourceProvider(durableStateStoreQuery, tag, system)
   }
 
@@ -83,7 +90,15 @@ object DurableStateSourceProvider {
     val durableStateStoreQuery =
       DurableStateStoreRegistry(system)
         .getDurableStateStoreFor(classOf[DurableStateStoreBySliceQuery[A]], durableStateStoreQueryPluginId)
+    changesBySlices(system, durableStateStoreQuery, entityType, minSlice, maxSlice)
+  }
 
+  def changesBySlices[A](
+      system: ActorSystem[_],
+      durableStateStoreQuery: DurableStateStoreBySliceQuery[A],
+      entityType: String,
+      minSlice: Int,
+      maxSlice: Int): SourceProvider[Offset, DurableStateChange[A]] = {
     new DurableStateBySlicesSourceProvider(durableStateStoreQuery, entityType, minSlice, maxSlice, system)
   }
 

diff --git a/...able-state/src/main/scala/akka/projection/state/scaladsl/DurableStateSourceProvider.scala b/...able-state/src/main/scala/akka/projection/state/scaladsl/DurableStateSourceProvider.scala
@@ -37,7 +37,13 @@ object DurableStateSourceProvider {
 
     val durableStateStoreQuery =
       DurableStateStoreRegistry(system).durableStateStoreFor[DurableStateStoreQuery[A]](pluginId)
+    changesByTag(system, durableStateStoreQuery, tag)
+  }
 
+  def changesByTag[A](
+      system: ActorSystem[_],
+      durableStateStoreQuery: DurableStateStoreQuery[A],
+      tag: String): SourceProvider[Offset, DurableStateChange[A]] = {
     new DurableStateStoreQuerySourceProvider(durableStateStoreQuery, tag, system)
   }
 
@@ -74,7 +80,15 @@ object DurableStateSourceProvider {
     val durableStateStoreQuery =
       DurableStateStoreRegistry(system)
         .durableStateStoreFor[DurableStateStoreBySliceQuery[A]](durableStateStoreQueryPluginId)
+    changesBySlices(system, durableStateStoreQuery, entityType, minSlice, maxSlice)
+  }
 
+  def changesBySlices[A](
+      system: ActorSystem[_],
+      durableStateStoreQuery: DurableStateStoreBySliceQuery[A],
+      entityType: String,
+      minSlice: Int,
+      maxSlice: Int): SourceProvider[Offset, DurableStateChange[A]] = {
     new DurableStateBySlicesSourceProvider(durableStateStoreQuery, entityType, minSlice, maxSlice, system)
   }
 

diff --git a/...entsourced/src/main/scala/akka/projection/eventsourced/javadsl/EventSourcedProvider.scala b/...entsourced/src/main/scala/akka/projection/eventsourced/javadsl/EventSourcedProvider.scala
@@ -39,10 +39,15 @@ object EventSourcedProvider {
       system: ActorSystem[_],
       readJournalPluginId: String,
       tag: String): SourceProvider[Offset, EventEnvelope[Event]] = {
-
     val eventsByTagQuery =
       PersistenceQuery(system).getReadJournalFor(classOf[EventsByTagQuery], readJournalPluginId)
+    eventsByTag(system, eventsByTagQuery, tag)
+  }
 
+  def eventsByTag[Event](
+      system: ActorSystem[_],
+      eventsByTagQuery: EventsByTagQuery,
+      tag: String): SourceProvider[Offset, EventEnvelope[Event]] = {
     new EventsByTagSourceProvider(system, eventsByTagQuery, tag)
   }
 
@@ -78,20 +83,17 @@ object EventSourcedProvider {
       entityType: String,
       minSlice: Int,
       maxSlice: Int): SourceProvider[Offset, akka.persistence.query.typed.EventEnvelope[Event]] = {
-
     val eventsBySlicesQuery =
       PersistenceQuery(system).getReadJournalFor(classOf[EventsBySliceQuery], readJournalPluginId)
+    eventsBySlices(system, eventsBySlicesQuery, entityType, minSlice, maxSlice)
+  }
 
-    if (!eventsBySlicesQuery.isInstanceOf[EventTimestampQuery])
-      throw new IllegalArgumentException(
-        s"[${eventsBySlicesQuery.getClass.getName}] with readJournalPluginId " +
-        s"[$readJournalPluginId] must implement [${classOf[EventTimestampQuery].getName}]")
-
-    if (!eventsBySlicesQuery.isInstanceOf[LoadEventQuery])
-      throw new IllegalArgumentException(
-        s"[${eventsBySlicesQuery.getClass.getName}] with readJournalPluginId " +
-        s"[$readJournalPluginId] must implement [${classOf[LoadEventQuery].getName}]")
-
+  def eventsBySlices[Event](
+      system: ActorSystem[_],
+      eventsBySlicesQuery: EventsBySliceQuery,
+      entityType: String,
+      minSlice: Int,
+      maxSlice: Int): SourceProvider[Offset, akka.persistence.query.typed.EventEnvelope[Event]] = {
     new EventsBySlicesSourceProvider(eventsBySlicesQuery, entityType, minSlice, maxSlice, system)
   }
 

diff --git a/...ntsourced/src/main/scala/akka/projection/eventsourced/scaladsl/EventSourcedProvider.scala b/...ntsourced/src/main/scala/akka/projection/eventsourced/scaladsl/EventSourcedProvider.scala
@@ -32,10 +32,15 @@ object EventSourcedProvider {
       system: ActorSystem[_],
       readJournalPluginId: String,
       tag: String): SourceProvider[Offset, EventEnvelope[Event]] = {
-
     val eventsByTagQuery =
       PersistenceQuery(system).readJournalFor[EventsByTagQuery](readJournalPluginId)
+    eventsByTag(system, eventsByTagQuery, tag)
+  }
 
+  def eventsByTag[Event](
+      system: ActorSystem[_],
+      eventsByTagQuery: EventsByTagQuery,
+      tag: String): SourceProvider[Offset, EventEnvelope[Event]] = {
     new EventsByTagSourceProvider(eventsByTagQuery, tag, system)
   }
 
@@ -67,7 +72,15 @@ object EventSourcedProvider {
       maxSlice: Int): SourceProvider[Offset, akka.persistence.query.typed.EventEnvelope[Event]] = {
     val eventsBySlicesQuery =
       PersistenceQuery(system).readJournalFor[EventsBySliceQuery](readJournalPluginId)
+    eventsBySlices(system, eventsBySlicesQuery, entityType, minSlice, maxSlice)
+  }
 
+  def eventsBySlices[Event](
+      system: ActorSystem[_],
+      eventsBySlicesQuery: EventsBySliceQuery,
+      entityType: String,
+      minSlice: Int,
+      maxSlice: Int): SourceProvider[Offset, akka.persistence.query.typed.EventEnvelope[Event]] = {
     new EventsBySlicesSourceProvider(eventsBySlicesQuery, entityType, minSlice, maxSlice, system)
   }