Fix CurrentEventsByTag never completes (#379)

* Fix CurrentEventsByTag never completes * Fix off by one and empty case
akkadotnet · Jul 12, 2024 · d6083bb · d6083bb
1 parent 2f423b6
commit d6083bb
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/src/Akka.Persistence.MongoDb/Query/EventsByTagPublisher.cs b/src/Akka.Persistence.MongoDb/Query/EventsByTagPublisher.cs
@@ -216,7 +216,7 @@ protected override void ReceiveInitialRequest()
         protected override void ReceiveIdleRequest()
         {
             Buffer.DeliverBuffer(TotalDemand);
-            if (Buffer.IsEmpty && CurrentOffset > ToOffset)
+            if (Buffer.IsEmpty && CurrentOffset >= ToOffset)
                 OnCompleteThenStop();
             else
                 Self.Tell(EventsByTagPublisher.Continue.Instance);
@@ -228,7 +228,7 @@ protected override void ReceiveRecoverySuccess(long highestSequenceNr)
             if (highestSequenceNr > 0 && highestSequenceNr < ToOffset)
                 _toOffset = highestSequenceNr;
 
-            if (Buffer.IsEmpty && CurrentOffset > ToOffset)
+            if (Buffer.IsEmpty && (CurrentOffset >= ToOffset || CurrentOffset == FromOffset))
                 OnCompleteThenStop();
             else
                 Self.Tell(EventsByTagPublisher.Continue.Instance);