Re-implement PersistenceIds persitence query to match scala implementation behavior

src/contrib/persistence/Akka.Persistence.Query.Sql/AllEventsPublisher.cs

@@ -141,7 +141,7 @@ internal sealed class LiveAllEventsPublisher : AbstractAllEventsPublisher
         public LiveAllEventsPublisher(long fromOffset, TimeSpan refreshInterval, int maxBufferSize, string writeJournalPluginId)
             : base(fromOffset, maxBufferSize, writeJournalPluginId)
         {
-            _tickCancelable = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(refreshInterval, refreshInterval, Self, EventsByTagPublisher.Continue.Instance, Self);
+            _tickCancelable = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(refreshInterval, refreshInterval, Self, AllEventsPublisher.Continue.Instance, Self);
         }
 
         protected override long ToOffset => long.MaxValue;

src/contrib/persistence/Akka.Persistence.Query.Sql/AllPersistenceIdsPublisher.cs

@@ -5,64 +5,193 @@
 // </copyright>
 //-----------------------------------------------------------------------
 
+using System;
 using Akka.Actor;
 using Akka.Persistence.Sql.Common.Journal;
 using Akka.Streams.Actors;
 
 namespace Akka.Persistence.Query.Sql
 {
-    internal sealed class AllPersistenceIdsPublisher : ActorPublisher<string>
+    internal sealed class CurrentPersistenceIdsPublisher : ActorPublisher<string>, IWithUnboundedStash
     {
-        public static Props Props(bool liveQuery, string writeJournalPluginId)
+        public static Props Props(string writeJournalPluginId)
         {
-            return Actor.Props.Create(() => new AllPersistenceIdsPublisher(liveQuery, writeJournalPluginId));
+            return Actor.Props.Create(() => new CurrentPersistenceIdsPublisher(writeJournalPluginId));
         }
 
-        private readonly bool _liveQuery;
         private readonly IActorRef _journalRef;
 
         private readonly DeliveryBuffer<string> _buffer;
 
-        public AllPersistenceIdsPublisher(bool liveQuery, string writeJournalPluginId)
+        public IStash Stash { get; set; }
+
+        public CurrentPersistenceIdsPublisher(string writeJournalPluginId)
         {
-            _liveQuery = liveQuery;
             _buffer = new DeliveryBuffer<string>(OnNext);
             _journalRef = Persistence.Instance.Apply(Context.System).JournalFor(writeJournalPluginId);
         }
 
-        protected override bool Receive(object message) => message.Match()
-            .With<Request>(_ =>
+        protected override bool Receive(object message)
+        {
+            switch (message)
             {
-                _journalRef.Tell(SubscribeAllPersistenceIds.Instance);
-                Become(Active);
-            })
-            .With<Cancel>(_ => Context.Stop(Self))
-            .WasHandled;
-
-        private bool Active(object message) => message.Match()
-            .With<CurrentPersistenceIds>(current =>
+                case Request _:
+                    _journalRef.Tell(new SelectCurrentPersistenceIds(0, Self));
+                    Become(Initializing);
+                    return true;
+                case Cancel _:
+                    Context.Stop(Self);
+                    return true;
+                default:
+                    return false;
+            }
+        }
+
+        private bool Initializing(object message)
+        {
+            switch (message)
             {
-                _buffer.AddRange(current.AllPersistenceIds);
-                _buffer.DeliverBuffer(TotalDemand);
+                case CurrentPersistenceIds current:
+                    _buffer.AddRange(current.AllPersistenceIds);
+                    _buffer.DeliverBuffer(TotalDemand);
 
-                if (!_liveQuery && _buffer.IsEmpty)
-                    OnCompleteThenStop();
-            })
-            .With<PersistenceIdAdded>(added =>
+                    if (_buffer.IsEmpty)
+                    {
+                        OnCompleteThenStop();
+                        return true;
+                    }
+
+                    Become(Active);
+                    Stash.UnstashAll();
+                    return true;
+                case Cancel _:
+                    Context.Stop(Self);
+                    return true;
+                default:
+                    Stash.Stash();
+                    return true;
+            }
+        }
+
+        private bool Active(object message)
+        {
+            switch (message)
             {
-                if (_liveQuery)
-                {
-                    _buffer.Add(added.PersistenceId);
+                case Request _:
                     _buffer.DeliverBuffer(TotalDemand);
-                }
-            })
-            .With<Request>(_ =>
+                    if (_buffer.IsEmpty)
+                        OnCompleteThenStop();
+                    return true;
+                case Cancel _:
+                    Context.Stop(Self);
+                    return true;
+                default:
+                    return false;
+            }
+        }
+    }
+
+    internal sealed class LivePersistenceIdsPublisher : ActorPublisher<string>, IWithUnboundedStash
+    {
+        private class Continue
+        {
+            public static readonly Continue Instance = new Continue();
+
+            private Continue() { }
+        }
+
+        public static Props Props(TimeSpan refreshInterval, string writeJournalPluginId)
+        {
+            return Actor.Props.Create(() => new LivePersistenceIdsPublisher(refreshInterval, writeJournalPluginId));
+        }
+
+        private long _lastOrderingOffset;
+        private readonly ICancelable _tickCancelable;
+        private readonly IActorRef _journalRef;
+        private readonly DeliveryBuffer<string> _buffer;
+
+        public IStash Stash { get; set; }
+
+        public LivePersistenceIdsPublisher(TimeSpan refreshInterval, string writeJournalPluginId)
+        {
+            _tickCancelable = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
+                refreshInterval, 
+                refreshInterval, 
+                Self, 
+                Continue.Instance, 
+                Self);
+            _buffer = new DeliveryBuffer<string>(OnNext);
+            _journalRef = Persistence.Instance.Apply(Context.System).JournalFor(writeJournalPluginId);
+        }
+
+        protected override void PostStop()
+        {
+            _tickCancelable.Cancel();
+            base.PostStop();
+        }
+
+        protected override bool Receive(object message)
+        {
+            switch (message)
             {
-                _buffer.DeliverBuffer(TotalDemand);
-                if (!_liveQuery && _buffer.IsEmpty)
-                    OnCompleteThenStop();
-            })
-            .With<Cancel>(_ => Context.Stop(Self))
-            .WasHandled;
+                case Request _:
+                    _journalRef.Tell(new SelectCurrentPersistenceIds(0, Self));
+                    Become(Initializing);
+                    return true;
+                case Continue _:
+                    return true;
+                case Cancel _:
+                    Context.Stop(Self);
+                    return true;
+                default:
+                    return false;
+            }
+        }
+
+        private bool Initializing(object message)
+        {
+            switch (message)
+            {
+                case CurrentPersistenceIds current:
+                    _lastOrderingOffset = current.HighestOrderingNumber;
+                    _buffer.AddRange(current.AllPersistenceIds);
+                    _buffer.DeliverBuffer(TotalDemand);
+
+                    Become(Active);
+                    Stash.UnstashAll();
+                    return true;
+                case Continue _:
+                    return true;
+                case Cancel _:
+                    Context.Stop(Self);
+                    return true;
+                default:
+                    Stash.Stash();
+                    return true;
+            }
+        }
+
+        private bool Active(object message)
+        {
+            switch (message)
+            {
+                case CurrentPersistenceIds added:
+                    _lastOrderingOffset = added.HighestOrderingNumber;
+                    _buffer.AddRange(added.AllPersistenceIds);
+                    _buffer.DeliverBuffer(TotalDemand);
+                    return true;
+                case Request _:
+                    _buffer.DeliverBuffer(TotalDemand);
+                    return true;
+                case Continue _:
+                    _journalRef.Tell(new SelectCurrentPersistenceIds(_lastOrderingOffset, Self));
+                    return true;
+                case Cancel _:
+                    Context.Stop(Self);
+                    return true;
+                default:
+                    return false;
+            }
+        }
     }
 }

src/contrib/persistence/Akka.Persistence.Query.Sql/SqlReadJournal.cs

@@ -11,6 +11,7 @@
 using Akka.Configuration;
 using Akka.Persistence.Journal;
 using Akka.Streams.Dsl;
+using Akka.Streams;
 using Akka.Util.Internal;
 
 namespace Akka.Persistence.Query.Sql
@@ -40,12 +41,16 @@ public static Config DefaultConfiguration()
         private readonly TimeSpan _refreshInterval;
         private readonly string _writeJournalPluginId;
         private readonly int _maxBufferSize;
+        private readonly ExtendedActorSystem _system;
+
+        private IPublisher<string> _persistenceIdsPublisher = null;
 
         public SqlReadJournal(ExtendedActorSystem system, Config config)
         {
             _refreshInterval = config.GetTimeSpan("refresh-interval", null);
             _writeJournalPluginId = config.GetString("write-plugin", null);
             _maxBufferSize = config.GetInt("max-buffer-size", 0);
+            _system = system;
         }
 
         /// <summary>
@@ -68,18 +73,36 @@ public SqlReadJournal(ExtendedActorSystem system, Config config)
         /// backend journal.
         /// </para>
         /// </summary>
-        public Source<string, NotUsed> PersistenceIds() => 
-            Source.ActorPublisher<string>(AllPersistenceIdsPublisher.Props(true, _writeJournalPluginId))
-            .MapMaterializedValue(_ => NotUsed.Instance)
-            .Named("AllPersistenceIds") as Source<string, NotUsed>;
+        public Source<string, NotUsed> PersistenceIds()
+        {
+            if (_persistenceIdsPublisher is null)
+            {
+                var graph =
+                    Source.ActorPublisher<string>(
+                            LivePersistenceIdsPublisher.Props(
+                                _refreshInterval, 
+                                _writeJournalPluginId))
+                        .ToMaterialized(Sink.DistinctRetainingFanOutPublisher<string>(PersistenceIdsShutdownCallback), Keep.Right);
+                _persistenceIdsPublisher = graph.Run(_system.Materializer());
+            }
+
+            return Source.FromPublisher(_persistenceIdsPublisher)
+                .MapMaterializedValue(_ => NotUsed.Instance)
+                .Named("AllPersistenceIds") as Source<string, NotUsed>;
+        }
+
+        private void PersistenceIdsShutdownCallback()
+        {
+            _persistenceIdsPublisher = null;
+        }
 
         /// <summary>
         /// Same type of query as <see cref="PersistenceIds"/> but the stream
         /// is completed immediately when it reaches the end of the "result set". Persistent
         /// actors that are created after the query is completed are not included in the stream.
         /// </summary>
-        public Source<string, NotUsed> CurrentPersistenceIds() => 
-            Source.ActorPublisher<string>(AllPersistenceIdsPublisher.Props(false, _writeJournalPluginId))
+        public Source<string, NotUsed> CurrentPersistenceIds()
+            => Source.ActorPublisher<string>(CurrentPersistenceIdsPublisher.Props(_writeJournalPluginId))
             .MapMaterializedValue(_ => NotUsed.Instance)
             .Named("CurrentPersistenceIds") as Source<string, NotUsed>;