Update to latest releases

src/Akka.Persistence.MongoDb.Tests/MongoDbAllEventsSpec.cs

@@ -13,6 +13,7 @@ public class MongoDbAllEventsSpec: AllEventsSpec, IClassFixture<DatabaseFixture>
     {
         private static Config CreateSpecConfig(DatabaseFixture databaseFixture, int id)
         {
+            // akka.test.single-expect-default = 10s
             var specString = @"
                 akka.test.single-expect-default = 10s
                 akka.persistence {
@@ -37,13 +38,9 @@ class = ""Akka.Persistence.MongoDb.Query.MongoDbReadJournalProvider, Akka.Persis
             return ConfigurationFactory.ParseString(specString);
         }
         public static readonly AtomicCounter Counter = new AtomicCounter(0);
-        private readonly ITestOutputHelper _output;
 
-        public MongoDbAllEventsSpec(ITestOutputHelper output, DatabaseFixture databaseFixture)
-            : base(CreateSpecConfig(databaseFixture, Counter.GetAndIncrement()), "MongoDbAllEventsSpec", output)
+        public MongoDbAllEventsSpec(ITestOutputHelper output, DatabaseFixture databaseFixture) : base(CreateSpecConfig(databaseFixture, Counter.GetAndIncrement()), "MongoDbAllEventsSpec", output)
         {
-            _output = output;
-            output.WriteLine(databaseFixture.ConnectionString + Counter.Current);
             ReadJournal = Sys.ReadJournalFor<MongoDbReadJournal>(MongoDbReadJournal.Identifier);
         }
     }

src/Akka.Persistence.MongoDb.Tests/MongoDbCurrentAllEventsSpec.cs

@@ -0,0 +1,49 @@
+using Akka.Configuration;
+using Akka.Persistence.MongoDb.Query;
+using Akka.Persistence.Query;
+using Akka.Persistence.TCK.Query;
+using Akka.Util.Internal;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Persistence.MongoDb.Tests
+{
+    public class MongoDbCurrentAllEventsSpec : CurrentAllEventsSpec, IClassFixture<DatabaseFixture>
+    {
+        private static Config CreateSpecConfig(DatabaseFixture databaseFixture, int id)
+        {
+            // akka.test.single-expect-default = 10s
+            var specString = @"
+                akka.test.single-expect-default = 10s
+                akka.persistence {
+                    publish-plugin-commands = on
+                    journal {
+                        plugin = ""akka.persistence.journal.mongodb""
+                        mongodb {
+                            class = ""Akka.Persistence.MongoDb.Journal.MongoDbJournal, Akka.Persistence.MongoDb""
+                            connection-string = """ + databaseFixture.ConnectionString + id + @"""
+                            auto-initialize = on
+                            collection = ""EventJournal""
+                        }
+                    }
+                    query {
+                        mongodb {
+                            class = ""Akka.Persistence.MongoDb.Query.MongoDbReadJournalProvider, Akka.Persistence.MongoDb""
+                            refresh-interval = 1s
+                        }
+                    }
+                }";
+
+            return ConfigurationFactory.ParseString(specString);
+        }
+        public static readonly AtomicCounter Counter = new AtomicCounter(0);
+
+        public MongoDbCurrentAllEventsSpec(ITestOutputHelper output, DatabaseFixture databaseFixture) : base(CreateSpecConfig(databaseFixture, Counter.GetAndIncrement()), "MongoDbAllEventsSpec", output)
+        {
+            ReadJournal = Sys.ReadJournalFor<MongoDbReadJournal>(MongoDbReadJournal.Identifier);
+        }
+    }
+}

src/Akka.Persistence.MongoDb/Auto/AutoSequence.cs

@@ -0,0 +1,17 @@
+using MongoDB.Bson;
+using MongoDB.Bson.Serialization.Attributes;
+
+namespace Akka.Persistence.MongoDb.Auto
+{
+    public class AutoSequence
+    {
+        [BsonId]
+        [BsonRepresentation(BsonType.ObjectId)]
+        [BsonElement("_id")]
+        public string Id { get; set; }
+
+        public string SequenceName { get; set; }
+
+        public int SequenceValue { get; set; }
+    }
+}

src/Akka.Persistence.MongoDb/Auto/SequenceRepository.cs

@@ -0,0 +1,26 @@
+using MongoDB.Driver;
+
+namespace Akka.Persistence.MongoDb.Auto
+{
+    public class SequenceRepository
+    {
+        protected readonly IMongoDatabase _database;
+        protected readonly IMongoCollection<AutoSequence> _collection;
+
+        public SequenceRepository(IMongoDatabase database)
+        {
+            _database = database;
+            _collection = _database.GetCollection<AutoSequence>(typeof(AutoSequence).Name);
+        }
+
+        public long GetSequenceValue(string sequenceName)
+        {
+            var filter = Builders<AutoSequence>.Filter.Eq(s => s.SequenceName, sequenceName);
+            var update = Builders<AutoSequence>.Update.Inc(s => s.SequenceValue, 1);
+
+            var result = _collection.FindOneAndUpdate(filter, update, new FindOneAndUpdateOptions<AutoSequence, AutoSequence> { IsUpsert = true, ReturnDocument = ReturnDocument.After });
+
+            return result.SequenceValue;
+        }
+    }
+}

src/Akka.Persistence.MongoDb/Journal/JournalEntry.cs

@@ -36,7 +36,7 @@ public class JournalEntry
         public string Manifest { get; set; }
 
         [BsonElement("Ordering")]
-        public BsonTimestamp Ordering { get; set; }
+        public long Ordering { get; set; }
 
         [BsonElement("Tags")]
         public ICollection<string> Tags { get; set; } = new HashSet<string>();

src/Akka.Persistence.MongoDb/Journal/MongoDbJournal.cs

@@ -7,10 +7,10 @@
 
 using Akka.Actor;
 using Akka.Persistence.Journal;
+using Akka.Persistence.MongoDb.Auto;
 using Akka.Persistence.MongoDb.Query;
 using Akka.Streams.Dsl;
 using Akka.Util;
-using MongoDB.Bson;
 using MongoDB.Driver;
 using System;
 using System.Collections.Generic;
@@ -31,6 +31,7 @@ public class MongoDbJournal : AsyncWriteJournal
         private Lazy<IMongoDatabase> _mongoDatabase;
         private Lazy<IMongoCollection<JournalEntry>> _journalCollection;
         private Lazy<IMongoCollection<MetadataEntry>> _metadataCollection;
+        private SequenceRepository _sequenceRepository;
 
         private readonly HashSet<string> _allPersistenceIds = new HashSet<string>();
         private readonly HashSet<IActorRef> _allPersistenceIdSubscribers = new HashSet<IActorRef>();
@@ -61,7 +62,6 @@ protected override void PreStart()
 
                 return client.GetDatabase(connectionString.DatabaseName);
             });
-
             _journalCollection = new Lazy<IMongoCollection<JournalEntry>>(() =>
             {
                 var collection = _mongoDatabase.Value.GetCollection<JournalEntry>(_settings.Collection);
@@ -107,6 +107,7 @@ protected override void PreStart()
 
                 return collection;
             });
+            _sequenceRepository = new SequenceRepository(_mongoDatabase.Value);
         }
 
         public override async Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr, long toSequenceNr, long max, Action<IPersistentRepresentation> recoveryCallback)
@@ -160,9 +161,9 @@ private async Task<long> ReplayTaggedMessagesAsync(ReplayTaggedMessages replay)
             var builder = Builders<JournalEntry>.Filter;
             var seqNoFilter = builder.AnyEq(x => x.Tags, tag);
             if (fromSequenceNr > 0)
-                seqNoFilter &= builder.Gt(x => x.Ordering, new BsonTimestamp(fromSequenceNr));
+                seqNoFilter &= builder.Gt(x => x.Ordering, fromSequenceNr);
             if (toSequenceNr != long.MaxValue)
-                seqNoFilter &= builder.Lte(x => x.Ordering, new BsonTimestamp(toSequenceNr));
+                seqNoFilter &= builder.Lte(x => x.Ordering, toSequenceNr);
 
 
             // Need to know what the highest seqNo of this query will be
@@ -175,14 +176,14 @@ private async Task<long> ReplayTaggedMessagesAsync(ReplayTaggedMessages replay)
             if (maxSeqNoEntry == null)
                 return 0L; // recovered nothing
 
-            var maxOrderingId = maxSeqNoEntry.Ordering.Value;
+            var maxOrderingId = maxSeqNoEntry.Ordering;
             var toSeqNo = Math.Min(toSequenceNr, maxOrderingId);
 
             var readFilter = builder.AnyEq(x => x.Tags, tag);
             if (fromSequenceNr > 0)
-                readFilter &= builder.Gt(x => x.Ordering, new BsonTimestamp(fromSequenceNr));
+                readFilter &= builder.Gt(x => x.Ordering, fromSequenceNr);
             if (toSequenceNr != long.MaxValue)
-                readFilter &= builder.Lte(x => x.Ordering, new BsonTimestamp(toSeqNo));
+                readFilter &= builder.Lte(x => x.Ordering, toSeqNo);
             var sort = Builders<JournalEntry>.Sort.Ascending(x => x.Ordering);
 
             await _journalCollection.Value
@@ -193,7 +194,7 @@ await _journalCollection.Value
                 {
                     var persistent = ToPersistenceRepresentation(entry, ActorRefs.NoSender);
                     foreach (var adapted in AdaptFromJournal(persistent))
-                        replay.ReplyTo.Tell(new ReplayedTaggedMessage(adapted, tag, entry.Ordering.Value),
+                        replay.ReplyTo.Tell(new ReplayedTaggedMessage(adapted, tag, entry.Ordering),
                             ActorRefs.NoSender);
                 });
 
@@ -319,7 +320,7 @@ private JournalEntry ToJournalEntry(IPersistentRepresentation message)
                 return new JournalEntry
                 {
                     Id = message.PersistenceId + "_" + message.SequenceNr,
-                    Ordering = new BsonTimestamp(0), // Auto-populates with timestamp
+                    Ordering = _sequenceRepository.GetSequenceValue("journalentry"), // Auto-populates with timestamp
                     IsDeleted = message.IsDeleted,
                     Payload = payload,
                     PersistenceId = message.PersistenceId,
@@ -338,7 +339,7 @@ private JournalEntry ToJournalEntry(IPersistentRepresentation message)
             return new JournalEntry
             {
                 Id = message.PersistenceId + "_" + message.SequenceNr,
-                Ordering = new BsonTimestamp(0), // Auto-populates with timestamp
+                Ordering = _sequenceRepository.GetSequenceValue("journalentry"), // Auto-populates with timestamp
                 IsDeleted = message.IsDeleted,
                 Payload = binary,
                 PersistenceId = message.PersistenceId,
@@ -475,27 +476,32 @@ protected virtual async Task<long> ReplayAllEventsAsync(ReplayAllEvents replay)
         {
             // Limit allows only integer
             var limitValue = replay.Max >= int.MaxValue ? int.MaxValue : (int)replay.Max;
-
-            var maxOrdering = 0L;
+            var take = Math.Min(replay.ToOffset - replay.FromOffset, replay.Max);
+            var maxOrdering = await GetHighestOrdering();
             var builder = Builders<JournalEntry>.Filter;
-            var filter = builder.Gte(x => x.SequenceNr, replay.FromOffset);
-            filter &= builder.Lte(x => x.SequenceNr, replay.ToOffset);
-            var sort = Builders<JournalEntry>.Sort.Ascending(x => x.SequenceNr);
-            await _journalCollection.Value
+            var filter = builder.Gt(x => x.Ordering, replay.FromOffset);
+            var sort = Builders<JournalEntry>.Sort.Ascending(x => x.Ordering);
+
+            var j = await _journalCollection.Value
                 .Find(filter)
                 .Sort(sort)
-                .Limit(limitValue)
+                .Limit(limitValue).ToListAsync();/*
                 .ForEachAsync(entry =>
                 {
                     var persistent = ToPersistenceRepresentation(entry, ActorRefs.NoSender);
                     foreach (var adapted in AdaptFromJournal(persistent))
                     {
-                        var currentSequence = persistent.SequenceNr;
-                        replay.ReplyTo.Tell(new ReplayedEvent(adapted, currentSequence), ActorRefs.NoSender);
-                        if(currentSequence > maxOrdering)
-                            maxOrdering = currentSequence;
+                        replay.ReplyTo.Tell(new ReplayedEvent(adapted, entry.Ordering), ActorRefs.NoSender);                        
                     }
-                });
+                });*/
+            foreach(var t in j)
+            {
+                var persistent = ToPersistenceRepresentation(t, ActorRefs.NoSender);
+                foreach (var adapted in AdaptFromJournal(persistent))
+                {
+                    replay.ReplyTo.Tell(new ReplayedEvent(adapted, t.Ordering), ActorRefs.NoSender);
+                }
+            }
             return maxOrdering;
         }
 
@@ -524,12 +530,20 @@ private async Task<long> GetHighestOrdering()
             return await Task.Run(() =>
             {
                 return _journalCollection.Value.AsQueryable()
-                    .Select(je => je.SequenceNr)
+                    .Select(je => je.Ordering)
                     .Distinct()
                     .Max();
             });
         }
-
+        private async Task<long> CountTotalRows()
+        {
+            return await Task.Run(() =>
+            {
+                return _journalCollection.Value.AsQueryable()
+                    .Select(je => je.SequenceNr)
+                    .Count();
+            });
+        }
         private void AddPersistenceIdSubscriber(IActorRef subscriber, string persistenceId)
         {
             if (!_persistenceIdSubscribers.TryGetValue(persistenceId, out var subscriptions))

src/Akka.Persistence.MongoDb/Query/DeliveryBuffer.cs

@@ -35,14 +35,17 @@ public void AddRange(IEnumerable<T> elements)
 
         public void DeliverBuffer(long demand)
         {
-            if (!Buffer.IsEmpty && demand > 0) {
+            if (!Buffer.IsEmpty && demand > 0) 
+            {
                 var totalDemand = Math.Min((int)demand, Buffer.Length);
-                if (Buffer.Length == 1) {
+                if (Buffer.Length == 1) 
+                {
                     // optimize for this common case
                     _onNext(Buffer[0]);
                     Buffer = ImmutableArray<T>.Empty;
                 }
-                else if (demand <= int.MaxValue) {
+                else if (demand <= int.MaxValue) 
+                {
                     for (var i = 0; i < totalDemand; i++)
                         _onNext(Buffer[i]);
 
@@ -55,6 +58,7 @@ public void DeliverBuffer(long demand)
                     Buffer = ImmutableArray<T>.Empty;
                 }
             }
+
         }
 
     }