From 361e8f9a4a2bae14ac6206372c564d8412eb1d8a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Tue, 17 Sep 2024 09:03:40 +0200
Subject: [PATCH] fix: Cancel any replays when the filter stage stops

---
 .../scala/akka/projection/grpc/internal/FilterStage.scala    | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilterStage.scala b/akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilterStage.scala
index 8814c2120..7fe21b96f 100644
--- a/akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilterStage.scala
+++ b/akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilterStage.scala
@@ -463,6 +463,11 @@ import org.slf4j.LoggerFactory
           pullInEnvOrReplay()
         }
       })
+
+      override def postStop(): Unit = {
+        replayInProgress.values.foreach(_.queue.cancel())
+        replayInProgress = Map.empty
+      }
     }
 
 }