Add CancellationToken support to all driver calls
#328
Conversation
close akkadotnet#191 - so far, only implemented journal calls
| using (var session = await _journalCollection.Value.Database.Client.StartSessionAsync(sessionOptions/*, cancellationToken*/)) | ||
| using (var session = | ||
| await _journalCollection.Value.Database.Client.StartSessionAsync( | ||
| sessionOptions /*, cancellationToken*/)) |
There was a problem hiding this comment.
Should we pass in a cancellation token here?
| ```hocon | ||
| akka.persistence { | ||
| journal { | ||
| plugin = "akka.persistence.journal.mongodb" |
There was a problem hiding this comment.
Updated README to include call-timeout settings.
| mongoPersistence.JournalSettings.Collection.Should().Be("EventJournal"); | ||
| mongoPersistence.JournalSettings.MetadataCollection.Should().Be("Metadata"); | ||
| mongoPersistence.JournalSettings.LegacySerialization.Should().BeFalse(); | ||
| mongoPersistence.JournalSettings.CallTimeout.Should().Be(10.Seconds()); |
There was a problem hiding this comment.
Validate new settings as part of our normal MongoDbSettingsSpecs.
|
|
||
| <PropertyGroup> | ||
| <TargetFrameworks>$(NetStandardLibVersion)</TargetFrameworks> | ||
| <LangVersion>latest</LangVersion> |
There was a problem hiding this comment.
Added this in order to support using statements, which I ended up not even needing in most cases.
| /// <summary> | ||
| /// Used to cancel all outstanding commands when the actor is stopped. | ||
| /// </summary> | ||
| private readonly CancellationTokenSource _pendingCommandsCancellation = new(); |
There was a problem hiding this comment.
Similar to what we do in Akka.Persistence.Sql.Common - have a global CTS for aborting all in-process operations when the Journal shuts down. This same architecture is mirrored in the SnapshotStore as well.
| .IndexKeys | ||
| .Ascending(entry => entry.PersistenceId) | ||
| .Descending(entry => entry.SequenceNr)); | ||
| var (perCallCts, unitedCts) = CreatePerCalLCts(); |
There was a problem hiding this comment.
Helper method to create a per-call CTS that is also linked to the global "operation CTS" is created here. We wrap both CTS in a using statement so they're safely disposed of afterwards.
| /// <summary> | ||
| /// Used to cancel all outstanding commands when the actor is stopped. | ||
| /// </summary> | ||
| private readonly CancellationTokenSource _pendingCommandsCancellation = new(); |
There was a problem hiding this comment.
Identical to what we did in journal
| _serialization = Context.System.Serialization; | ||
| } | ||
|
|
||
| private (CancellationTokenSource perCallCts, CancellationTokenSource untiedCts) CreatePerCalLCts() |
There was a problem hiding this comment.
Identical to what we did in journal
Changes
Fixes #191
Checklist
For significant changes, please ensure that the following have been completed (delete if not relevant):