|
|
@ -7,11 +7,14 @@ |
|
|
|
|
|
|
|
|
using NodaTime; |
|
|
using NodaTime; |
|
|
using Squidex.Infrastructure.EventSourcing; |
|
|
using Squidex.Infrastructure.EventSourcing; |
|
|
|
|
|
using Squidex.Infrastructure.Tasks; |
|
|
|
|
|
using Squidex.Log; |
|
|
|
|
|
|
|
|
namespace Squidex.Infrastructure.States |
|
|
namespace Squidex.Infrastructure.States |
|
|
{ |
|
|
{ |
|
|
public class SimpleState<T> where T : class, new() |
|
|
public class SimpleState<T> where T : class, new() |
|
|
{ |
|
|
{ |
|
|
|
|
|
private readonly AsyncLock? lockObject; |
|
|
private readonly IPersistence<T> persistence; |
|
|
private readonly IPersistence<T> persistence; |
|
|
private bool isLoaded; |
|
|
private bool isLoaded; |
|
|
private Instant lastWrite; |
|
|
private Instant lastWrite; |
|
|
@ -25,12 +28,12 @@ namespace Squidex.Infrastructure.States |
|
|
|
|
|
|
|
|
public IClock Clock { get; set; } = SystemClock.Instance; |
|
|
public IClock Clock { get; set; } = SystemClock.Instance; |
|
|
|
|
|
|
|
|
public SimpleState(IPersistenceFactory<T> persistenceFactory, Type ownerType, string id) |
|
|
public SimpleState(IPersistenceFactory<T> persistenceFactory, Type ownerType, string id, bool lockOperations = false) |
|
|
: this(persistenceFactory, ownerType, DomainId.Create(id)) |
|
|
: this(persistenceFactory, ownerType, DomainId.Create(id), lockOperations) |
|
|
{ |
|
|
{ |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public SimpleState(IPersistenceFactory<T> persistenceFactory, Type ownerType, DomainId id) |
|
|
public SimpleState(IPersistenceFactory<T> persistenceFactory, Type ownerType, DomainId id, bool lockOperations = false) |
|
|
{ |
|
|
{ |
|
|
Guard.NotNull(persistenceFactory); |
|
|
Guard.NotNull(persistenceFactory); |
|
|
|
|
|
|
|
|
@ -38,53 +41,76 @@ namespace Squidex.Infrastructure.States |
|
|
{ |
|
|
{ |
|
|
Value = state; |
|
|
Value = state; |
|
|
}); |
|
|
}); |
|
|
|
|
|
|
|
|
|
|
|
if (lockOperations) |
|
|
|
|
|
{ |
|
|
|
|
|
lockObject = new AsyncLock(); |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public async Task LoadAsync( |
|
|
public async Task LoadAsync( |
|
|
CancellationToken ct = default) |
|
|
CancellationToken ct = default) |
|
|
{ |
|
|
{ |
|
|
await persistence.ReadAsync(ct: ct); |
|
|
using (await LockAsync()) |
|
|
|
|
|
{ |
|
|
isLoaded = true; |
|
|
await LoadInternalAsync(ct); |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public Task ClearAsync( |
|
|
public async Task ClearAsync( |
|
|
CancellationToken ct = default) |
|
|
CancellationToken ct = default) |
|
|
{ |
|
|
{ |
|
|
Value = new T(); |
|
|
using (await LockAsync()) |
|
|
|
|
|
{ |
|
|
|
|
|
// Reset state first, in case the deletion fails.
|
|
|
|
|
|
Value = new T(); |
|
|
|
|
|
|
|
|
return persistence.DeleteAsync(ct); |
|
|
await persistence.DeleteAsync(ct); |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public async Task WriteAsync(int ifNotWrittenWithinMs, |
|
|
public async Task WriteAsync(int ifNotWrittenWithinMs, |
|
|
CancellationToken ct = default) |
|
|
CancellationToken ct = default) |
|
|
{ |
|
|
{ |
|
|
var now = Clock.GetCurrentInstant(); |
|
|
using (await LockAsync()) |
|
|
|
|
|
|
|
|
if (ifNotWrittenWithinMs > 0 && now.Minus(lastWrite).TotalMilliseconds < ifNotWrittenWithinMs) |
|
|
|
|
|
{ |
|
|
{ |
|
|
return; |
|
|
// Calculate the timestamp once.
|
|
|
} |
|
|
var now = Clock.GetCurrentInstant(); |
|
|
|
|
|
|
|
|
await persistence.WriteSnapshotAsync(Value, ct); |
|
|
if (ifNotWrittenWithinMs > 0 && now.Minus(lastWrite).TotalMilliseconds < ifNotWrittenWithinMs) |
|
|
|
|
|
{ |
|
|
|
|
|
return; |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
await persistence.WriteSnapshotAsync(Value, ct); |
|
|
|
|
|
|
|
|
lastWrite = now; |
|
|
// Only update the last write property if it is successful.
|
|
|
|
|
|
lastWrite = now; |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public async Task WriteAsync( |
|
|
public async Task WriteAsync( |
|
|
CancellationToken ct = default) |
|
|
CancellationToken ct = default) |
|
|
{ |
|
|
{ |
|
|
await persistence.WriteSnapshotAsync(Value, ct); |
|
|
using (await LockAsync()) |
|
|
|
|
|
{ |
|
|
|
|
|
await persistence.WriteSnapshotAsync(Value, ct); |
|
|
|
|
|
|
|
|
lastWrite = Clock.GetCurrentInstant(); |
|
|
// Only update the last write property if it is successful.
|
|
|
|
|
|
lastWrite = Clock.GetCurrentInstant(); |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public async Task WriteEventAsync(Envelope<IEvent> envelope, |
|
|
public async Task WriteEventAsync(Envelope<IEvent> envelope, |
|
|
CancellationToken ct = default) |
|
|
CancellationToken ct = default) |
|
|
{ |
|
|
{ |
|
|
await persistence.WriteEventAsync(envelope, ct); |
|
|
using (await LockAsync()) |
|
|
|
|
|
{ |
|
|
|
|
|
await persistence.WriteEventAsync(envelope, ct); |
|
|
|
|
|
|
|
|
lastWrite = Clock.GetCurrentInstant(); |
|
|
// Only update the last write property if it is successful.
|
|
|
|
|
|
lastWrite = Clock.GetCurrentInstant(); |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public Task UpdateAsync(Func<T, bool> updater, int retries = 20, |
|
|
public Task UpdateAsync(Func<T, bool> updater, int retries = 20, |
|
|
@ -99,32 +125,55 @@ namespace Squidex.Infrastructure.States |
|
|
Guard.GreaterEquals(retries, 1); |
|
|
Guard.GreaterEquals(retries, 1); |
|
|
Guard.LessThan(retries, 100); |
|
|
Guard.LessThan(retries, 100); |
|
|
|
|
|
|
|
|
if (!isLoaded) |
|
|
using (await LockAsync()) |
|
|
{ |
|
|
|
|
|
await LoadAsync(ct); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
for (var i = 0; i < retries; i++) |
|
|
|
|
|
{ |
|
|
{ |
|
|
try |
|
|
// Ensure that the state is loaded before we make the update.
|
|
|
|
|
|
if (!isLoaded) |
|
|
{ |
|
|
{ |
|
|
var (isChanged, result) = updater(Value); |
|
|
await LoadInternalAsync(ct); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
if (!isChanged) |
|
|
for (var i = 0; i < retries; i++) |
|
|
|
|
|
{ |
|
|
|
|
|
try |
|
|
{ |
|
|
{ |
|
|
|
|
|
var (isChanged, result) = updater(Value); |
|
|
|
|
|
|
|
|
|
|
|
// If nothing has been changed, we can avoid the call to the database.
|
|
|
|
|
|
if (!isChanged) |
|
|
|
|
|
{ |
|
|
|
|
|
return result; |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
await WriteAsync(ct); |
|
|
return result; |
|
|
return result; |
|
|
} |
|
|
} |
|
|
|
|
|
catch (InconsistentStateException) when (i < retries - 1) |
|
|
await WriteAsync(ct); |
|
|
{ |
|
|
return result; |
|
|
await LoadInternalAsync(ct); |
|
|
} |
|
|
} |
|
|
catch (InconsistentStateException) when (i < retries - 1) |
|
|
|
|
|
{ |
|
|
|
|
|
await LoadAsync(ct); |
|
|
|
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
return default!; |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private async Task LoadInternalAsync( |
|
|
|
|
|
CancellationToken ct = default) |
|
|
|
|
|
{ |
|
|
|
|
|
await persistence.ReadAsync(ct: ct); |
|
|
|
|
|
|
|
|
|
|
|
isLoaded = true; |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private Task<IDisposable> LockAsync() |
|
|
|
|
|
{ |
|
|
|
|
|
if (lockObject != null) |
|
|
|
|
|
{ |
|
|
|
|
|
return lockObject.EnterAsync(); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
return default!; |
|
|
return Task.FromResult<IDisposable>(NoopDisposable.Instance); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|