Fix for snapshot stage double pull race

akka · Oct 19, 2023 · d3bc39b · d3bc39b
1 parent aa314c9
commit d3bc39b
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/core/src/main/scala/akka/persistence/r2dbc/internal/StartingFromSnapshotStage.scala b/core/src/main/scala/akka/persistence/r2dbc/internal/StartingFromSnapshotStage.scala
@@ -66,7 +66,7 @@ import akka.stream.stage.OutHandler
         }
 
         override def onUpstreamFinish(): Unit = {
-          val primaryHandler = new PrimaryHandler
+          val primaryHandler = new PrimaryHandler(isAvailable(out))
           self.setHandler(out, primaryHandler)
 
           subFusingMaterializer.materialize(
@@ -80,9 +80,9 @@ import akka.stream.stage.OutHandler
         }
       }
 
-      class PrimaryHandler extends OutHandler with InHandler {
+      class PrimaryHandler(pullImmediately: Boolean) extends OutHandler with InHandler {
         val subSink = new SubSinkInlet[EventEnvelope[Event]]("snapshots")
-        subSink.pull()
+        if (pullImmediately) subSink.pull()
         subSink.setHandler(this)
 
         override def onPull(): Unit = {