Update to latest releases #159
Conversation
src/Akka.Persistence.MongoDb.Tests/Akka.Persistence.MongoDb.Tests.csproj
Show resolved
Hide resolved
src/Akka.Persistence.MongoDb.Tests/Akka.Persistence.MongoDb.Tests.csproj
Outdated
Show resolved
Hide resolved
…h missing Tests not in the latest version of 1.4.11
…n_the_last_subscriber_left()
|
All in all, the codes looks good, just need some minor tweaks. |
|
Ok @Arkatufus. Can you direct me to what blocks of codes need tweaking? |
|
@eaba One last changes from my side: |
|
|
||
| namespace Akka.Persistence.MongoDb.Tests | ||
| { | ||
| class MongoDbAllCurrentEventsSpec |
There was a problem hiding this comment.
Is this a spec you're still working on?
|
The current test failures with the "AllEventsSpec" - is this because the events are being returned out of order? I only briefly checked the error messages but haven't looked at the detailed logs. |
|
It is being caused by Message mismatch |
| @@ -113,8 +110,6 @@ protected override void PreStart() | |||
|
|
|||
| public override async Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr, long toSequenceNr, long max, Action<IPersistentRepresentation> recoveryCallback) | |||
| { | |||
| NotifyNewPersistenceIdAdded(persistenceId); | |||
| public override async Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr) | ||
| { | ||
| NotifyNewPersistenceIdAdded(persistenceId); |
| } | ||
| } | ||
|
|
||
| internal sealed class LivePersistenceIdsPublisher : ActorPublisher<string>, IWithUnboundedStash |
| } | ||
| private async Task<long> CountTotalRows() | ||
| { | ||
| return await Task.Run(() => |
There was a problem hiding this comment.
Rather than do this, why now just use CountAsync on the MongoDb driver itself?
There was a problem hiding this comment.
I have now deleted that...I only created it while I was trying to understand the issue with ReplayAllEventsAsyn
| } | ||
|
|
||
| private async Task<long> GetHighestOrdering() | ||
| { | ||
| return await Task.Run(() => |
There was a problem hiding this comment.
Same comment as below - just use the MaxAsync call
There was a problem hiding this comment.
Also, this query can be rewritten in a much less expensive way:
We have a sorted index built on top of the Ordering column. Just select the maximum value from it. Don't do an entire table scan. That turns this into a O(1) query rather than an O(n) query.
| @@ -400,7 +423,7 @@ private async Task SetHighSequenceId(IList<AtomicWrite> messages) | |||
| SequenceNr = highSequenceId | |||
| }; | |||
|
|
|||
| await _metadataCollection.Value.ReplaceOneAsync(filter, metadataEntry, new UpdateOptions() { IsUpsert = true }); | |||
| await _metadataCollection.Value.ReplaceOneAsync(filter, metadataEntry, new ReplaceOptions() { IsUpsert = true }); | |||
There was a problem hiding this comment.
LGTM, since we are replacing the metadataentry outright
| { | ||
| _allPersistenceIdSubscribers.Add(subscriber); | ||
| } | ||
| subscriber.Tell(new CurrentPersistenceIds(GetAllPersistenceIds())); |
| readFilter &= builder.Lte(x => x.Ordering, new BsonTimestamp(toSeqNo)); | ||
| var sort = Builders<JournalEntry>.Sort.Ascending(x => x.Ordering); | ||
|
|
||
| await _journalCollection.Value.Find(readFilter) |
| } | ||
|
|
||
| subscriptions.Add(subscriber); | ||
| } | ||
|
|
||
| private IEnumerable<string> GetAllPersistenceIds() | ||
| { | ||
| return _journalCollection.Value.AsQueryable() |
There was a problem hiding this comment.
Change this to query the _metadataCollection instead - that will be a tremendously less expensive query than scanning the entire journal.
|
nice work @eaba |
We have Akka.Persistence.MongoDB plugin stuck on older releases of Akka.NET - this is because of some changes we made to the Akka.Persistence.TCK in Akka.NET v1.4.10.