Skip to content

Commit 3d1f43c

Browse files
AarononthewebAaronontheweb
authored
Merge branch 'v1.6' into AOT-canary
2 parents 4aff06e + 447b550 commit 3d1f43c

File tree

14 files changed

+201
-133
lines changed

14 files changed

+201
-133
lines changed

‎src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs‎

Lines changed: 70 additions & 36 deletions
Large diffs are not rendered by default.

‎src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/SqlJournal.cs‎

Lines changed: 61 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -76,11 +76,13 @@ protected override bool ReceivePluginInternal(object message)
7676
{
7777
case SelectCurrentPersistenceIds msg:
7878
SelectAllPersistenceIdsAsync(msg.Offset)
79-
.PipeTo(msg.ReplyTo, success: h => new CurrentPersistenceIds(h.Ids, h.LastOrdering), failure: e => new Status.Failure(e));
79+
.PipeTo(msg.ReplyTo, success: h => new CurrentPersistenceIds(h.Ids, h.LastOrdering),
80+
failure: e => new Status.Failure(e));
8081
return true;
8182
case ReplayTaggedMessages replay:
8283
ReplayTaggedMessagesAsync(replay)
83-
.PipeTo(replay.ReplyTo, success: h => new RecoverySuccess(h), failure: e => new ReplayMessagesFailure(e));
84+
.PipeTo(replay.ReplyTo, success: h => new RecoverySuccess(h),
85+
failure: e => new ReplayMessagesFailure(e));
8486
return true;
8587
case ReplayAllEvents replay:
8688
ReplayAllEventsAsync(replay)
@@ -122,11 +124,13 @@ protected override async Task<IImmutableList<Exception>> WriteMessagesAsync(IEnu
122124
else eventToTags.Add(p, ImmutableHashSet<string>.Empty);
123125

124126
if (IsTagId(p.PersistenceId))
125-
throw new InvalidOperationException($"Persistence Id {p.PersistenceId} must not start with {QueryExecutor.Configuration.TagsColumnName}");
127+
throw new InvalidOperationException(
128+
$"Persistence Id {p.PersistenceId} must not start with {QueryExecutor.Configuration.TagsColumnName}");
126129
}
127130

128131
var batch = new WriteJournalBatch(eventToTags);
129-
using(var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
132+
using (var cancellationToken =
133+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
130134
await QueryExecutor.InsertBatchAsync(connection, cancellationToken.Token, batch);
131135
}
132136
}).ToArray();
@@ -149,15 +153,20 @@ protected virtual async Task<long> ReplayTaggedMessagesAsync(ReplayTaggedMessage
149153
using (var connection = CreateDbConnection())
150154
{
151155
await connection.OpenAsync();
152-
using(var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
156+
using (var cancellationToken =
157+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
153158
{
154159
return await QueryExecutor
155-
.SelectByTagAsync(connection, cancellationToken.Token, replay.Tag, replay.FromOffset, replay.ToOffset, replay.Max, replayedTagged => {
156-
foreach(var adapted in AdaptFromJournal(replayedTagged.Persistent))
157-
{
158-
replay.ReplyTo.Tell(new ReplayedTaggedMessage(adapted, replayedTagged.Tag, replayedTagged.Offset), ActorRefs.NoSender);
159-
}
160-
});
160+
.SelectByTagAsync(connection, cancellationToken.Token, replay.Tag, replay.FromOffset,
161+
replay.ToOffset, replay.Max, replayedTagged =>
162+
{
163+
foreach (var adapted in AdaptFromJournal(replayedTagged.Persistent))
164+
{
165+
replay.ReplyTo.Tell(
166+
new ReplayedTaggedMessage(adapted, replayedTagged.Tag, replayedTagged.Offset),
167+
ActorRefs.NoSender);
168+
}
169+
});
161170
}
162171
}
163172
}
@@ -167,34 +176,41 @@ protected virtual async Task<long> ReplayAllEventsAsync(ReplayAllEvents replay)
167176
using (var connection = CreateDbConnection())
168177
{
169178
await connection.OpenAsync();
170-
using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
179+
using (var cancellationToken =
180+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
171181
{
172182
return await QueryExecutor
173183
.SelectAllEventsAsync(
174184
connection,
175-
cancellationToken.Token,
176-
replay.FromOffset,
185+
cancellationToken.Token,
186+
replay.FromOffset,
177187
replay.ToOffset,
178-
replay.Max,
179-
replayedEvent => {
188+
replay.Max,
189+
replayedEvent =>
190+
{
180191
foreach (var adapted in AdaptFromJournal(replayedEvent.Persistent))
181192
{
182-
replay.ReplyTo.Tell(new ReplayedEvent(adapted, replayedEvent.Offset), ActorRefs.NoSender);
193+
replay.ReplyTo.Tell(new ReplayedEvent(adapted, replayedEvent.Offset),
194+
ActorRefs.NoSender);
183195
}
184196
});
185197
}
186198
}
187199
}
188200

189-
protected virtual async Task<(IEnumerable<string> Ids, long LastOrdering)> SelectAllPersistenceIdsAsync(long offset)
201+
protected virtual async Task<(IEnumerable<string> Ids, long LastOrdering)> SelectAllPersistenceIdsAsync(
202+
long offset)
190203
{
191204
using (var connection = CreateDbConnection())
192205
{
193206
await connection.OpenAsync();
194-
using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
207+
using (var cancellationToken =
208+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
195209
{
196-
var lastOrdering = await QueryExecutor.SelectHighestSequenceNrAsync(connection, cancellationToken.Token);
197-
var ids = await QueryExecutor.SelectAllPersistenceIdsAsync(connection, cancellationToken.Token, offset);
210+
var lastOrdering =
211+
await QueryExecutor.SelectHighestSequenceNrAsync(connection, cancellationToken.Token);
212+
var ids = await QueryExecutor.SelectAllPersistenceIdsAsync(connection, cancellationToken.Token,
213+
offset);
198214
return (ids, lastOrdering);
199215
}
200216
}
@@ -210,15 +226,18 @@ protected virtual async Task<long> ReplayAllEventsAsync(ReplayAllEvents replay)
210226
/// <param name="max">TBD</param>
211227
/// <param name="recoveryCallback">TBD</param>
212228
/// <returns>TBD</returns>
213-
public override async Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr, long toSequenceNr, long max,
229+
public override async Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr,
230+
long toSequenceNr, long max,
214231
Action<IPersistentRepresentation> recoveryCallback)
215232
{
216233
using (var connection = CreateDbConnection())
217234
{
218235
await connection.OpenAsync();
219-
using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
236+
using (var cancellationToken =
237+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
220238
{
221-
await QueryExecutor.SelectByPersistenceIdAsync(connection, cancellationToken.Token, persistenceId, fromSequenceNr, toSequenceNr, max, recoveryCallback);
239+
await QueryExecutor.SelectByPersistenceIdAsync(connection, cancellationToken.Token, persistenceId,
240+
fromSequenceNr, toSequenceNr, max, recoveryCallback);
222241
}
223242
}
224243
}
@@ -257,7 +276,7 @@ protected bool WaitingForInitialization(object message)
257276
return true;
258277
case Status.Failure fail:
259278
Log.Error(fail.Cause, "Failure during {0} initialization.", Self);
260-
279+
261280
// trigger a restart so we have some hope of succeeding in the future even if initialization failed
262281
throw new ApplicationException("Failed to initialize SQL Journal.", fail.Cause);
263282
default:
@@ -268,15 +287,16 @@ protected bool WaitingForInitialization(object message)
268287

269288
private async Task<object> Initialize()
270289
{
271-
if (!Settings.AutoInitialize)
290+
if (!Settings.AutoInitialize)
272291
return new Status.Success(NotUsed.Instance);
273292

274293
try
275294
{
276295
using (var connection = CreateDbConnection())
277296
{
278297
await connection.OpenAsync();
279-
using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
298+
using (var cancellationToken =
299+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
280300
{
281301
await QueryExecutor.CreateTablesAsync(connection, cancellationToken.Token);
282302
}
@@ -286,6 +306,7 @@ private async Task<object> Initialize()
286306
{
287307
return new Status.Failure(e);
288308
}
309+
289310
return new Status.Success(NotUsed.Instance);
290311
}
291312

@@ -328,9 +349,11 @@ protected override async Task DeleteMessagesToAsync(string persistenceId, long t
328349
using (var connection = CreateDbConnection())
329350
{
330351
await connection.OpenAsync();
331-
using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
352+
using (var cancellationToken =
353+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
332354
{
333-
await QueryExecutor.DeleteBatchAsync(connection, cancellationToken.Token, persistenceId, toSequenceNr);
355+
await QueryExecutor.DeleteBatchAsync(connection, cancellationToken.Token, persistenceId,
356+
toSequenceNr);
334357
}
335358
}
336359
}
@@ -346,9 +369,11 @@ public override async Task<long> ReadHighestSequenceNrAsync(string persistenceId
346369
using (var connection = CreateDbConnection())
347370
{
348371
await connection.OpenAsync();
349-
using (var cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
372+
using (var cancellationToken =
373+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
350374
{
351-
return await QueryExecutor.SelectHighestSequenceNrAsync(connection, cancellationToken.Token, persistenceId);
375+
return await QueryExecutor.SelectHighestSequenceNrAsync(connection, cancellationToken.Token,
376+
persistenceId);
352377
}
353378
}
354379
}
@@ -361,15 +386,18 @@ protected virtual string GetConnectionString()
361386
{
362387
var connectionString = Settings.ConnectionString;
363388

389+
#if NETSTANDARD
364390
if (string.IsNullOrEmpty(connectionString))
365391
{
366-
connectionString = System.Configuration.ConfigurationManager.ConnectionStrings[Settings.ConnectionStringName].ConnectionString;
392+
connectionString =
393+
System.Configuration.ConfigurationManager.ConnectionStrings[Settings.ConnectionStringName].ConnectionString;
367394
}
395+
#endif
368396

369397
return connectionString;
370398
}
371399

372400
protected ITimestampProvider GetTimestampProvider(string typeName) =>
373401
TimestampProviderProvider.GetTimestampProvider(typeName, Context);
374402
}
375-
}
403+
}

‎src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs‎

Lines changed: 32 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,16 @@ namespace Akka.Persistence.Sql.Common.Snapshot
2424
public abstract class SqlSnapshotStore : SnapshotStore, IWithUnboundedStash
2525
{
2626
#region messages
27-
27+
2828
private sealed class Initialized
2929
{
3030
public static readonly Initialized Instance = new Initialized();
31-
private Initialized() { }
31+
32+
private Initialized()
33+
{
34+
}
3235
}
33-
36+
3437
#endregion
3538

3639
/// <summary>
@@ -57,6 +60,7 @@ protected SqlSnapshotStore(Config config)
5760
/// TBD
5861
/// </summary>
5962
protected ILoggingAdapter Log => _log ?? (_log ?? Context.GetLogger());
63+
6064
private ILoggingAdapter _log;
6165

6266
/// <summary>
@@ -114,7 +118,8 @@ private async Task<object> Initialize()
114118
try
115119
{
116120
using (var connection = CreateDbConnection())
117-
using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
121+
using (var nestedCancellationTokenSource =
122+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
118123
{
119124
await connection.OpenAsync(nestedCancellationTokenSource.Token);
120125
await QueryExecutor.CreateTableAsync(connection, nestedCancellationTokenSource.Token);
@@ -129,15 +134,15 @@ private async Task<object> Initialize()
129134

130135
private bool WaitingForInitialization(object message)
131136
{
132-
switch(message)
137+
switch (message)
133138
{
134139
case Initialized _:
135140
UnbecomeStacked();
136141
Stash.UnstashAll();
137142
return true;
138143
case Status.Failure msg:
139144
Log.Error(msg.Cause, "Error during snapshot store initialization");
140-
145+
141146
// trigger a restart so we have some hope of succeeding in the future even if initialization failed
142147
throw new ApplicationException("Failed to initialize SQL SnapshotStore.", msg.Cause);
143148
default:
@@ -154,10 +159,13 @@ protected virtual string GetConnectionString()
154159
{
155160
var connectionString = Settings.ConnectionString;
156161

162+
#if NETSTANDARD
157163
if (string.IsNullOrEmpty(connectionString))
158164
{
159-
connectionString = System.Configuration.ConfigurationManager.ConnectionStrings[Settings.ConnectionStringName].ConnectionString;
165+
connectionString =
166+
System.Configuration.ConfigurationManager.ConnectionStrings[Settings.ConnectionStringName].ConnectionString;
160167
}
168+
#endif
161169

162170
return connectionString;
163171
}
@@ -168,13 +176,16 @@ protected virtual string GetConnectionString()
168176
/// <param name="persistenceId">TBD</param>
169177
/// <param name="criteria">TBD</param>
170178
/// <returns>TBD</returns>
171-
protected override async Task<SelectedSnapshot> LoadAsync(string persistenceId, SnapshotSelectionCriteria criteria)
179+
protected override async Task<SelectedSnapshot> LoadAsync(string persistenceId,
180+
SnapshotSelectionCriteria criteria)
172181
{
173182
using (var connection = CreateDbConnection())
174-
using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
183+
using (var nestedCancellationTokenSource =
184+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
175185
{
176186
await connection.OpenAsync(nestedCancellationTokenSource.Token);
177-
return await QueryExecutor.SelectSnapshotAsync(connection, nestedCancellationTokenSource.Token, persistenceId, criteria.MaxSequenceNr, criteria.MaxTimeStamp);
187+
return await QueryExecutor.SelectSnapshotAsync(connection, nestedCancellationTokenSource.Token,
188+
persistenceId, criteria.MaxSequenceNr, criteria.MaxTimeStamp);
178189
}
179190
}
180191

@@ -187,7 +198,8 @@ protected override async Task<SelectedSnapshot> LoadAsync(string persistenceId,
187198
protected override async Task SaveAsync(SnapshotMetadata metadata, object snapshot)
188199
{
189200
using (var connection = CreateDbConnection())
190-
using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
201+
using (var nestedCancellationTokenSource =
202+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
191203
{
192204
await connection.OpenAsync(nestedCancellationTokenSource.Token);
193205
await QueryExecutor.InsertAsync(connection, nestedCancellationTokenSource.Token, snapshot, metadata);
@@ -202,11 +214,13 @@ protected override async Task SaveAsync(SnapshotMetadata metadata, object snapsh
202214
protected override async Task DeleteAsync(SnapshotMetadata metadata)
203215
{
204216
using (var connection = CreateDbConnection())
205-
using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
217+
using (var nestedCancellationTokenSource =
218+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
206219
{
207220
await connection.OpenAsync(nestedCancellationTokenSource.Token);
208221
DateTime? timestamp = metadata.Timestamp != DateTime.MinValue ? metadata.Timestamp : default(DateTime?);
209-
await QueryExecutor.DeleteAsync(connection, nestedCancellationTokenSource.Token, metadata.PersistenceId, metadata.SequenceNr, timestamp);
222+
await QueryExecutor.DeleteAsync(connection, nestedCancellationTokenSource.Token, metadata.PersistenceId,
223+
metadata.SequenceNr, timestamp);
210224
}
211225
}
212226

@@ -219,11 +233,13 @@ protected override async Task DeleteAsync(SnapshotMetadata metadata)
219233
protected override async Task DeleteAsync(string persistenceId, SnapshotSelectionCriteria criteria)
220234
{
221235
using (var connection = CreateDbConnection())
222-
using (var nestedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
236+
using (var nestedCancellationTokenSource =
237+
CancellationTokenSource.CreateLinkedTokenSource(_pendingRequestsCancellation.Token))
223238
{
224239
await connection.OpenAsync(nestedCancellationTokenSource.Token);
225-
await QueryExecutor.DeleteBatchAsync(connection, nestedCancellationTokenSource.Token, persistenceId, criteria.MaxSequenceNr, criteria.MaxTimeStamp);
240+
await QueryExecutor.DeleteBatchAsync(connection, nestedCancellationTokenSource.Token, persistenceId,
241+
criteria.MaxSequenceNr, criteria.MaxTimeStamp);
226242
}
227243
}
228244
}
229-
}
245+
}

0 commit comments

Comments
 (0)