Merge pull request #147 from SwissBorg/import-upstream-changes

Import upstream changes
SwissBorg · Feb 24, 2021 · 621d3c9 · 621d3c9
2 parents a4565cc + cd48ce7
commit 621d3c9
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 6 deletions.
diff --git a/core/src/main/scala/akka/persistence/postgres/query/scaladsl/PostgresReadJournal.scala b/core/src/main/scala/akka/persistence/postgres/query/scaladsl/PostgresReadJournal.scala
@@ -50,7 +50,9 @@ class PostgresReadJournal(config: Config, configPath: String)(implicit val syste
   val readJournalConfig = new ReadJournalConfig(config)
 
   private val writePluginId = config.getString("write-plugin")
-  private val eventAdapters = Persistence(system).adaptersFor(writePluginId)
+  // If 'config' is empty, or if the plugin reference is not found, then the write plugin will be resolved from the
+  // ActorSystem configuration. Otherwise, it will be resolved from the provided 'config'.
+  private val eventAdapters = Persistence(system).adaptersFor(writePluginId, config)
 
   val readJournalDao: ReadJournalDao = {
     val slickDb = SlickExtension(system).database(config)
@@ -294,10 +296,13 @@ class PostgresReadJournal(config: Config, configPath: String)(implicit val syste
       .mapConcat(identity)
   }
 
-  def currentEventsByTag(tag: String, offset: Long): Source[EventEnvelope, NotUsed] =
-    Source.future(readJournalDao.maxJournalSequence()).flatMapConcat { maxOrderingInDb =>
-      eventsByTag(tag, offset, terminateAfterOffset = Some(maxOrderingInDb))
-    }
+  def currentEventsByTag(tag: String, offset: Long): Source[EventEnvelope, NotUsed] = {
+    Source
+      .futureSource(readJournalDao.maxJournalSequence().map { maxOrderingInDb =>
+        eventsByTag(tag, offset, terminateAfterOffset = Some(maxOrderingInDb))
+      })
+      .mapMaterializedValue(_ => NotUsed)
+  }
 
   /**
    * Query events that have a specific tag.

diff --git a/project/ProjectAutoPlugin.scala b/project/ProjectAutoPlugin.scala
@@ -27,7 +27,7 @@ object ProjectAutoPlugin extends AutoPlugin {
   override val projectSettings: Seq[Setting[_]] = Seq(
     crossVersion := CrossVersion.binary,
     crossScalaVersions := Dependencies.ScalaVersions,
-    scalaVersion := Dependencies.Scala212,
+    scalaVersion := Dependencies.Scala213,
     Test / fork := true,
     Test / parallelExecution := false,
     Test / logBuffered := true,