Use plugin-dispatcher all the way, #866

akka · Apr 21, 2021 · 5bf89b4 · 5bf89b4
1 parent b7dd238
commit 5bf89b4
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 8 deletions.
diff --git a/core/src/main/scala/akka/persistence/cassandra/journal/CassandraRecovery.scala b/core/src/main/scala/akka/persistence/cassandra/journal/CassandraRecovery.scala
@@ -14,9 +14,10 @@ import akka.persistence.cassandra.journal.TagWriters.TagWrite
 import akka.persistence.cassandra.query.EventsByPersistenceIdStage.{ Extractors, TaggedPersistentRepr }
 import akka.stream.scaladsl.{ Sink, Source }
 import akka.util.OptionVal
-
 import scala.concurrent._
 
+import akka.stream.ActorAttributes
+
 trait CassandraRecovery extends CassandraTagRecovery with TaggedPreparedStatements {
   this: CassandraJournal =>
 
@@ -74,9 +75,17 @@ trait CassandraRecovery extends CassandraTagRecovery with TaggedPreparedStatemen
               dispatcher = sessionSettings.pluginDispatcher)
             .mapAsync(1)(sendMissingTagWrite(tp, tagWrites.get))
         }))
-        .map(te => queries.mapEvent(te.pr))
-        .runForeach(replayCallback)
-        .map(_ => ())
+        .map { te =>
+          println(s"# asyncReplayMessages mapEvent ${Thread.currentThread().getName}") // FIXME
+          queries.mapEvent(te.pr)
+        }
+        .map { p =>
+          println(s"# asyncReplayMessages replayCallback ${Thread.currentThread().getName}") // FIXME
+          replayCallback(p)
+        }
+        .withAttributes(ActorAttributes.dispatcher(sessionSettings.pluginDispatcher))
+        .runWith(Sink.ignore)
+        .map(_ => ())(akka.dispatch.ExecutionContexts.sameThreadExecutionContext)
 
     } else {
       queries
@@ -93,9 +102,17 @@ trait CassandraRecovery extends CassandraTagRecovery with TaggedPreparedStatemen
           extractor = Extractors.persistentRepr(eventDeserializer, serialization),
           // run the query on the journal dispatcher (not the queries dispatcher)
           dispatcher = sessionSettings.pluginDispatcher)
-        .map(p => queries.mapEvent(p.persistentRepr))
-        .runForeach(replayCallback)
-        .map(_ => ())
+        .map { p =>
+          println(s"# asyncReplayMessages mapEvent ${Thread.currentThread().getName}") // FIXME
+          queries.mapEvent(p.persistentRepr)
+        }
+        .map { p =>
+          println(s"# asyncReplayMessages replayCallback ${Thread.currentThread().getName}") // FIXME
+          replayCallback(p)
+        }
+        .withAttributes(ActorAttributes.dispatcher(sessionSettings.pluginDispatcher))
+        .runWith(Sink.ignore)
+        .map(_ => ())(akka.dispatch.ExecutionContexts.sameThreadExecutionContext)
     }
   }
 

diff --git a/core/src/main/scala/akka/persistence/cassandra/query/EventsByPersistenceIdStage.scala b/core/src/main/scala/akka/persistence/cassandra/query/EventsByPersistenceIdStage.scala
@@ -86,8 +86,10 @@ import akka.util.OptionVal
       executeStatement(selectDeletedToQuery.bind(persistenceId)).map(r =>
         Option(r.one()).map(_.getLong("deleted_to")).getOrElse(0))
 
-    private def executeStatement(statement: Statement)(implicit ec: ExecutionContext): Future[ResultSet] =
+    private def executeStatement(statement: Statement)(implicit ec: ExecutionContext): Future[ResultSet] = {
+      println(s"# EventsByPersistenceIdStage: ${Thread.currentThread().getName}") // FIXME
       session.executeAsync(withCustom(statement)).asScala
+    }
 
     private def withCustom(statement: Statement): Statement = {
       customConsistencyLevel.foreach(statement.setConsistencyLevel)

diff --git a/core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala b/core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala
@@ -619,6 +619,8 @@ class CassandraReadJournal(system: ExtendedActorSystem, cfg: Config)
       fastForwardEnabled: Boolean = false,
       dispatcher: String): Source[T, Future[EventsByPersistenceIdStage.Control]] = {
 
+    println(s"# eventsByPersistenceId call on ${Thread.currentThread().getName}") // FIXME
+
     val deserializeEventAsync = queryPluginConfig.deserializationParallelism > 1
 
     createFutureSource(combinedEventsByPersistenceIdStmts) { (s, c) =>
@@ -642,6 +644,7 @@ class CassandraReadJournal(system: ExtendedActorSystem, cfg: Config)
             fastForwardEnabled))
         .named(name)
     }.mapAsync(queryPluginConfig.deserializationParallelism) { row =>
+        println(s"# eventsByPersistenceId deserialization mapAsync ${Thread.currentThread().getName}") // FIXME
         extractor.extract(row, deserializeEventAsync)
       }
       .withAttributes(ActorAttributes.dispatcher(dispatcher))