mirror of https://github.com/Squidex/squidex.git
Browse Source
* Code cleanup * Batch store. * Fixes. * More performance improvements. * Temp * Fix parallelization. * Ignore restored events for some consumers. * Tests fixed.pull/680/head
Sebastian Stehle
5 years ago
committed by
GitHub
GitHub
83 changed files with 1466 additions and 697 deletions
@ -0,0 +1,76 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Generic; |
|||
using Squidex.Infrastructure; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Backup |
|||
{ |
|||
public sealed class StreamMapper |
|||
{ |
|||
private readonly Dictionary<string, long> streams = new Dictionary<string, long>(1000); |
|||
private readonly RestoreContext context; |
|||
private readonly DomainId brokenAppId; |
|||
|
|||
public StreamMapper(RestoreContext context) |
|||
{ |
|||
Guard.NotNull(context, nameof(context)); |
|||
|
|||
this.context = context; |
|||
|
|||
brokenAppId = DomainId.Combine(context.PreviousAppId, context.PreviousAppId); |
|||
} |
|||
|
|||
public (string Stream, DomainId) Map(string stream) |
|||
{ |
|||
Guard.NotNullOrEmpty(stream, nameof(stream)); |
|||
|
|||
var typeIndex = stream.IndexOf("-", StringComparison.Ordinal); |
|||
var typeName = stream.Substring(0, typeIndex); |
|||
|
|||
var id = DomainId.Create(stream[(typeIndex + 1)..]); |
|||
|
|||
if (id.Equals(context.PreviousAppId) || id.Equals(brokenAppId)) |
|||
{ |
|||
id = context.AppId; |
|||
} |
|||
else |
|||
{ |
|||
var separator = DomainId.IdSeparator; |
|||
|
|||
var secondId = id.ToString().AsSpan(); |
|||
|
|||
var indexOfSecondPart = secondId.IndexOf(separator, StringComparison.Ordinal); |
|||
if (indexOfSecondPart > 0 && indexOfSecondPart < secondId.Length - separator.Length - 1) |
|||
{ |
|||
secondId = secondId[(indexOfSecondPart + separator.Length)..]; |
|||
} |
|||
|
|||
id = DomainId.Combine(context.AppId, DomainId.Create(secondId.ToString())); |
|||
} |
|||
|
|||
stream = $"{typeName}-{id}"; |
|||
|
|||
return (stream, id); |
|||
} |
|||
|
|||
public long GetStreamOffset(string streamName) |
|||
{ |
|||
Guard.NotNullOrEmpty(streamName, nameof(streamName)); |
|||
|
|||
if (!streams.TryGetValue(streamName, out var offset)) |
|||
{ |
|||
offset = EtagVersion.Empty; |
|||
} |
|||
|
|||
streams[streamName] = offset + 1; |
|||
|
|||
return offset; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,112 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
|
|||
#pragma warning disable RECS0108 // Warns about static fields in generic types
|
|||
|
|||
namespace Squidex.Infrastructure.States |
|||
{ |
|||
public sealed class BatchContext<T> : IBatchContext<T> |
|||
{ |
|||
private static readonly List<Envelope<IEvent>> EmptyStream = new List<Envelope<IEvent>>(); |
|||
private readonly Type owner; |
|||
private readonly ISnapshotStore<T> snapshotStore; |
|||
private readonly IEventStore eventStore; |
|||
private readonly IEventDataFormatter eventDataFormatter; |
|||
private readonly IStreamNameResolver streamNameResolver; |
|||
private readonly Dictionary<DomainId, (long, List<Envelope<IEvent>>)> @events = new Dictionary<DomainId, (long, List<Envelope<IEvent>>)>(); |
|||
private List<(DomainId Key, T Snapshot, long Version)>? snapshots; |
|||
|
|||
internal BatchContext( |
|||
Type owner, |
|||
ISnapshotStore<T> snapshotStore, |
|||
IEventStore eventStore, |
|||
IEventDataFormatter eventDataFormatter, |
|||
IStreamNameResolver streamNameResolver) |
|||
{ |
|||
this.owner = owner; |
|||
this.snapshotStore = snapshotStore; |
|||
this.eventStore = eventStore; |
|||
this.eventDataFormatter = eventDataFormatter; |
|||
this.streamNameResolver = streamNameResolver; |
|||
} |
|||
|
|||
internal void Add(DomainId key, T snapshot, long version) |
|||
{ |
|||
snapshots ??= new List<(DomainId Key, T Snapshot, long Version)>(); |
|||
snapshots.Add((key, snapshot, version)); |
|||
} |
|||
|
|||
public async Task LoadAsync(IEnumerable<DomainId> ids) |
|||
{ |
|||
var streamNames = ids.ToDictionary(x => x, x => streamNameResolver.GetStreamName(owner, x.ToString())); |
|||
|
|||
if (streamNames.Count == 0) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
var streams = await eventStore.QueryManyAsync(streamNames.Values); |
|||
|
|||
foreach (var (id, streamName) in streamNames) |
|||
{ |
|||
if (streams.TryGetValue(streamName, out var data)) |
|||
{ |
|||
var stream = data.Select(eventDataFormatter.ParseIfKnown).NotNull().ToList(); |
|||
|
|||
events[id] = (data.Count - 1, stream); |
|||
} |
|||
else |
|||
{ |
|||
events[id] = (EtagVersion.Empty, EmptyStream); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public async ValueTask DisposeAsync() |
|||
{ |
|||
await CommitAsync(); |
|||
} |
|||
|
|||
public Task CommitAsync() |
|||
{ |
|||
var current = Interlocked.Exchange(ref snapshots, null!); |
|||
|
|||
if (current == null || current.Count == 0) |
|||
{ |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
return snapshotStore.WriteManyAsync(current); |
|||
} |
|||
|
|||
public IPersistence<T> WithEventSourcing(Type owner, DomainId key, HandleEvent? applyEvent) |
|||
{ |
|||
var (version, streamEvents) = events[key]; |
|||
|
|||
return new BatchPersistence<T>(key, this, version, streamEvents, applyEvent); |
|||
} |
|||
|
|||
public IPersistence<T> WithSnapshotsAndEventSourcing(Type owner, DomainId key, HandleSnapshot<T>? applySnapshot, HandleEvent? applyEvent) |
|||
{ |
|||
var (version, streamEvents) = events[key]; |
|||
|
|||
return new BatchPersistence<T>(key, this, version, streamEvents, applyEvent); |
|||
} |
|||
|
|||
public IPersistence<T> WithSnapshots(Type owner, DomainId key, HandleSnapshot<T>? applySnapshot) |
|||
{ |
|||
throw new NotSupportedException(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,80 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
|
|||
namespace Squidex.Infrastructure.States |
|||
{ |
|||
internal class BatchPersistence<T> : IPersistence<T> |
|||
{ |
|||
private readonly DomainId ownerKey; |
|||
private readonly BatchContext<T> context; |
|||
private readonly IReadOnlyList<Envelope<IEvent>> events; |
|||
private readonly HandleEvent? applyEvent; |
|||
|
|||
public long Version { get; } |
|||
|
|||
internal BatchPersistence(DomainId ownerKey, BatchContext<T> context, long version, IReadOnlyList<Envelope<IEvent>> @events, |
|||
HandleEvent? applyEvent) |
|||
{ |
|||
this.ownerKey = ownerKey; |
|||
this.context = context; |
|||
this.events = events; |
|||
this.applyEvent = applyEvent; |
|||
|
|||
Version = version; |
|||
} |
|||
|
|||
public Task DeleteAsync() |
|||
{ |
|||
throw new NotSupportedException(); |
|||
} |
|||
|
|||
public Task WriteEventsAsync(IReadOnlyList<Envelope<IEvent>> events) |
|||
{ |
|||
throw new NotSupportedException(); |
|||
} |
|||
|
|||
public Task ReadAsync(long expectedVersion = -2) |
|||
{ |
|||
if (applyEvent != null) |
|||
{ |
|||
foreach (var @event in events) |
|||
{ |
|||
if (!applyEvent(@event)) |
|||
{ |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
if (expectedVersion > EtagVersion.Any && expectedVersion != Version) |
|||
{ |
|||
if (Version == EtagVersion.Empty) |
|||
{ |
|||
throw new DomainObjectNotFoundException(ownerKey.ToString()!); |
|||
} |
|||
else |
|||
{ |
|||
throw new InconsistentStateException(Version, expectedVersion); |
|||
} |
|||
} |
|||
|
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public Task WriteSnapshotAsync(T state) |
|||
{ |
|||
context.Add(ownerKey, state, Version); |
|||
|
|||
return Task.CompletedTask; |
|||
} |
|||
} |
|||
} |
|||
@ -1,13 +1,26 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Collections.Generic; |
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
|
|||
namespace Squidex.Infrastructure.States |
|||
{ |
|||
public interface IPersistence : IPersistence<None> |
|||
public interface IPersistence<in TState> |
|||
{ |
|||
long Version { get; } |
|||
|
|||
Task DeleteAsync(); |
|||
|
|||
Task WriteEventsAsync(IReadOnlyList<Envelope<IEvent>> events); |
|||
|
|||
Task WriteSnapshotAsync(TState state); |
|||
|
|||
Task ReadAsync(long expectedVersion = EtagVersion.Any); |
|||
} |
|||
} |
|||
|
|||
@ -0,0 +1,25 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
|
|||
namespace Squidex.Infrastructure.States |
|||
{ |
|||
public delegate bool HandleEvent(Envelope<IEvent> @event); |
|||
|
|||
public delegate void HandleSnapshot<in T>(T state); |
|||
|
|||
public interface IPersistenceFactory<T> |
|||
{ |
|||
IPersistence<T> WithEventSourcing(Type owner, DomainId id, HandleEvent? applyEvent); |
|||
|
|||
IPersistence<T> WithSnapshots(Type owner, DomainId id, HandleSnapshot<T>? applySnapshot); |
|||
|
|||
IPersistence<T> WithSnapshotsAndEventSourcing(Type owner, DomainId id, HandleSnapshot<T>? applySnapshot, HandleEvent? applyEvent); |
|||
} |
|||
} |
|||
@ -1,221 +0,0 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
|
|||
#pragma warning disable RECS0012 // 'if' statement can be re-written as 'switch' statement
|
|||
|
|||
namespace Squidex.Infrastructure.States |
|||
{ |
|||
internal class Persistence<TSnapshot, TKey> : IPersistence<TSnapshot> where TKey : notnull |
|||
{ |
|||
private readonly TKey ownerKey; |
|||
private readonly ISnapshotStore<TSnapshot, TKey> snapshotStore; |
|||
private readonly IEventStore eventStore; |
|||
private readonly IEventDataFormatter eventDataFormatter; |
|||
private readonly PersistenceMode persistenceMode; |
|||
private readonly HandleSnapshot<TSnapshot>? applyState; |
|||
private readonly HandleEvent? applyEvent; |
|||
private readonly Lazy<string> streamName; |
|||
private long versionSnapshot = EtagVersion.Empty; |
|||
private long versionEvents = EtagVersion.Empty; |
|||
private long version = EtagVersion.Empty; |
|||
|
|||
public long Version |
|||
{ |
|||
get => version; |
|||
} |
|||
|
|||
private bool UseSnapshots |
|||
{ |
|||
get => (persistenceMode & PersistenceMode.Snapshots) == PersistenceMode.Snapshots; |
|||
} |
|||
|
|||
private bool UseEventSourcing |
|||
{ |
|||
get => (persistenceMode & PersistenceMode.EventSourcing) == PersistenceMode.EventSourcing; |
|||
} |
|||
|
|||
public Persistence(TKey ownerKey, Type ownerType, |
|||
IEventStore eventStore, |
|||
IEventDataFormatter eventDataFormatter, |
|||
ISnapshotStore<TSnapshot, TKey> snapshotStore, |
|||
IStreamNameResolver streamNameResolver, |
|||
PersistenceMode persistenceMode, |
|||
HandleSnapshot<TSnapshot>? applyState, |
|||
HandleEvent? applyEvent) |
|||
{ |
|||
this.ownerKey = ownerKey; |
|||
this.applyState = applyState; |
|||
this.applyEvent = applyEvent; |
|||
this.eventStore = eventStore; |
|||
this.eventDataFormatter = eventDataFormatter; |
|||
this.persistenceMode = persistenceMode; |
|||
this.snapshotStore = snapshotStore; |
|||
|
|||
streamName = new Lazy<string>(() => streamNameResolver.GetStreamName(ownerType, ownerKey.ToString()!)); |
|||
} |
|||
|
|||
public async Task DeleteAsync() |
|||
{ |
|||
if (UseSnapshots) |
|||
{ |
|||
await snapshotStore.RemoveAsync(ownerKey); |
|||
} |
|||
|
|||
if (UseEventSourcing) |
|||
{ |
|||
await eventStore.DeleteStreamAsync(streamName.Value); |
|||
} |
|||
} |
|||
|
|||
public async Task ReadAsync(long expectedVersion = EtagVersion.Any) |
|||
{ |
|||
versionSnapshot = EtagVersion.Empty; |
|||
versionEvents = EtagVersion.Empty; |
|||
|
|||
if (UseSnapshots) |
|||
{ |
|||
await ReadSnapshotAsync(); |
|||
} |
|||
|
|||
if (UseEventSourcing) |
|||
{ |
|||
await ReadEventsAsync(); |
|||
} |
|||
|
|||
UpdateVersion(); |
|||
|
|||
if (expectedVersion > EtagVersion.Any && expectedVersion != version) |
|||
{ |
|||
if (version == EtagVersion.Empty) |
|||
{ |
|||
throw new DomainObjectNotFoundException(ownerKey.ToString()!); |
|||
} |
|||
else |
|||
{ |
|||
throw new InconsistentStateException(version, expectedVersion); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private async Task ReadSnapshotAsync() |
|||
{ |
|||
var (state, position) = await snapshotStore.ReadAsync(ownerKey); |
|||
|
|||
// Treat all negative values as not-found (empty).
|
|||
position = Math.Max(position, EtagVersion.Empty); |
|||
|
|||
versionSnapshot = position; |
|||
versionEvents = position; |
|||
|
|||
if (applyState != null && position >= 0) |
|||
{ |
|||
applyState(state); |
|||
} |
|||
} |
|||
|
|||
private async Task ReadEventsAsync() |
|||
{ |
|||
var events = await eventStore.QueryAsync(streamName.Value, versionEvents + 1); |
|||
|
|||
var isStopped = false; |
|||
|
|||
foreach (var @event in events) |
|||
{ |
|||
var newVersion = versionEvents + 1; |
|||
|
|||
if (@event.EventStreamNumber != newVersion) |
|||
{ |
|||
throw new InvalidOperationException("Events must follow the snapshot version in consecutive order with no gaps."); |
|||
} |
|||
|
|||
// Skip the parsing for performance reasons if we are not interested, but continue reading to get the version.
|
|||
if (!isStopped) |
|||
{ |
|||
var parsedEvent = eventDataFormatter.ParseIfKnown(@event); |
|||
|
|||
if (applyEvent != null && parsedEvent != null) |
|||
{ |
|||
isStopped = !applyEvent(parsedEvent); |
|||
} |
|||
} |
|||
|
|||
versionEvents++; |
|||
} |
|||
} |
|||
|
|||
public async Task WriteSnapshotAsync(TSnapshot state) |
|||
{ |
|||
var oldVersion = versionSnapshot; |
|||
|
|||
if (oldVersion == EtagVersion.Empty && UseEventSourcing) |
|||
{ |
|||
oldVersion = (versionEvents - 1); |
|||
} |
|||
|
|||
var newVersion = UseEventSourcing ? versionEvents : oldVersion + 1; |
|||
|
|||
if (newVersion == versionSnapshot) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
await snapshotStore.WriteAsync(ownerKey, state, oldVersion, newVersion); |
|||
|
|||
versionSnapshot = newVersion; |
|||
|
|||
UpdateVersion(); |
|||
} |
|||
|
|||
public async Task WriteEventsAsync(IReadOnlyList<Envelope<IEvent>> events) |
|||
{ |
|||
Guard.NotEmpty(events, nameof(events)); |
|||
|
|||
var oldVersion = EtagVersion.Any; |
|||
|
|||
if (UseEventSourcing) |
|||
{ |
|||
oldVersion = versionEvents; |
|||
} |
|||
|
|||
var eventCommitId = Guid.NewGuid(); |
|||
var eventData = events.Select(x => eventDataFormatter.ToEventData(x, eventCommitId, true)).ToArray(); |
|||
|
|||
try |
|||
{ |
|||
await eventStore.AppendAsync(eventCommitId, streamName.Value, oldVersion, eventData); |
|||
} |
|||
catch (WrongEventVersionException ex) |
|||
{ |
|||
throw new InconsistentStateException(ex.CurrentVersion, ex.ExpectedVersion, ex); |
|||
} |
|||
|
|||
versionEvents += eventData.Length; |
|||
} |
|||
|
|||
private void UpdateVersion() |
|||
{ |
|||
if (persistenceMode == PersistenceMode.Snapshots) |
|||
{ |
|||
version = versionSnapshot; |
|||
} |
|||
else if (persistenceMode == PersistenceMode.EventSourcing) |
|||
{ |
|||
version = versionEvents; |
|||
} |
|||
else if (persistenceMode == PersistenceMode.SnapshotsAndEventSourcing) |
|||
{ |
|||
version = Math.Max(versionEvents, versionSnapshot); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,70 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using FakeItEasy; |
|||
using Squidex.Infrastructure; |
|||
using Xunit; |
|||
|
|||
namespace Squidex.Domain.Apps.Entities.Backup |
|||
{ |
|||
public class StreamMapperTests |
|||
{ |
|||
private readonly DomainId appIdOld = DomainId.NewGuid(); |
|||
private readonly DomainId appId = DomainId.NewGuid(); |
|||
private readonly StreamMapper sut; |
|||
|
|||
public StreamMapperTests() |
|||
{ |
|||
sut = new StreamMapper(new RestoreContext(appId, |
|||
A.Fake<IUserMapping>(), |
|||
A.Fake<IBackupReader>(), |
|||
appIdOld)); |
|||
} |
|||
|
|||
[Fact] |
|||
public void Should_map_old_app_id() |
|||
{ |
|||
var result = sut.Map($"app-{appIdOld}"); |
|||
|
|||
Assert.Equal(($"app-{appId}", appId), result); |
|||
} |
|||
|
|||
[Fact] |
|||
public void Should_map_old_app_broken_id() |
|||
{ |
|||
var result = sut.Map($"app-{appIdOld}--{appIdOld}"); |
|||
|
|||
Assert.Equal(($"app-{appId}", appId), result); |
|||
} |
|||
|
|||
[Fact] |
|||
public void Should_map_non_app_id() |
|||
{ |
|||
var result = sut.Map($"content-{appIdOld}--123"); |
|||
|
|||
Assert.Equal(($"content-{appId}--123", DomainId.Create($"{appId}--123")), result); |
|||
} |
|||
|
|||
[Fact] |
|||
public void Should_map_non_app_id_with_double_slash() |
|||
{ |
|||
var result = sut.Map($"content-{appIdOld}--other--id"); |
|||
|
|||
Assert.Equal(($"content-{appId}--other--id", DomainId.Create($"{appId}--other--id")), result); |
|||
} |
|||
|
|||
[Fact] |
|||
public void Should_map_non_combined_id() |
|||
{ |
|||
var id = DomainId.NewGuid(); |
|||
|
|||
var result = sut.Map($"content-{id}"); |
|||
|
|||
Assert.Equal(($"content-{appId}--{id}", DomainId.Create($"{appId}--{id}")), result); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,164 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschränkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Threading.Tasks; |
|||
using FakeItEasy; |
|||
using Squidex.Infrastructure.EventSourcing; |
|||
using Squidex.Infrastructure.TestHelpers; |
|||
using Xunit; |
|||
|
|||
namespace Squidex.Infrastructure.States |
|||
{ |
|||
public class PersistenceBatchTests |
|||
{ |
|||
private readonly ISnapshotStore<int> snapshotStore = A.Fake<ISnapshotStore<int>>(); |
|||
private readonly IEventDataFormatter eventDataFormatter = A.Fake<IEventDataFormatter>(); |
|||
private readonly IEventStore eventStore = A.Fake<IEventStore>(); |
|||
private readonly IStreamNameResolver streamNameResolver = A.Fake<IStreamNameResolver>(); |
|||
private readonly IStore<int> sut; |
|||
|
|||
public PersistenceBatchTests() |
|||
{ |
|||
A.CallTo(() => streamNameResolver.GetStreamName(None.Type, A<string>._)) |
|||
.ReturnsLazily(x => x.GetArgument<string>(1)!); |
|||
|
|||
sut = new Store<int>(snapshotStore, eventStore, eventDataFormatter, streamNameResolver); |
|||
} |
|||
|
|||
[Fact] |
|||
public async Task Should_read_from_preloaded_events() |
|||
{ |
|||
var event1_1 = new MyEvent { MyProperty = "event1_1" }; |
|||
var event1_2 = new MyEvent { MyProperty = "event1_2" }; |
|||
var event2_1 = new MyEvent { MyProperty = "event2_1" }; |
|||
var event2_2 = new MyEvent { MyProperty = "event2_2" }; |
|||
|
|||
var key1 = DomainId.NewGuid(); |
|||
var key2 = DomainId.NewGuid(); |
|||
|
|||
var bulk = sut.WithBatchContext(None.Type); |
|||
|
|||
SetupEventStore(new Dictionary<DomainId, List<MyEvent>> |
|||
{ |
|||
[key1] = new List<MyEvent> { event1_1, event1_2 }, |
|||
[key2] = new List<MyEvent> { event2_1, event2_2 } |
|||
}); |
|||
|
|||
await bulk.LoadAsync(new[] { key1, key2 }); |
|||
|
|||
var persistedEvents1 = Save.Events(); |
|||
var persistence1 = bulk.WithEventSourcing(None.Type, key1, persistedEvents1.Write); |
|||
|
|||
await persistence1.ReadAsync(); |
|||
|
|||
var persistedEvents2 = Save.Events(); |
|||
var persistence2 = bulk.WithEventSourcing(None.Type, key2, persistedEvents2.Write); |
|||
|
|||
await persistence2.ReadAsync(); |
|||
|
|||
Assert.Equal(persistedEvents1.ToArray(), new[] { event1_1, event1_2 }); |
|||
Assert.Equal(persistedEvents2.ToArray(), new[] { event2_1, event2_2 }); |
|||
} |
|||
|
|||
[Fact] |
|||
public async Task Should_provide_empty_events_if_nothing_loaded() |
|||
{ |
|||
var key = DomainId.NewGuid(); |
|||
|
|||
var bulk = sut.WithBatchContext(None.Type); |
|||
|
|||
await bulk.LoadAsync(new[] { key }); |
|||
|
|||
var persistedEvents = Save.Events(); |
|||
var persistence = bulk.WithEventSourcing(None.Type, key, persistedEvents.Write); |
|||
|
|||
await persistence.ReadAsync(); |
|||
|
|||
Assert.Empty(persistedEvents.ToArray()); |
|||
Assert.Empty(persistedEvents.ToArray()); |
|||
} |
|||
|
|||
[Fact] |
|||
public void Should_throw_exception_if_not_preloaded() |
|||
{ |
|||
var key = DomainId.NewGuid(); |
|||
|
|||
var bulk = sut.WithBatchContext(None.Type); |
|||
|
|||
Assert.Throws<KeyNotFoundException>(() => bulk.WithEventSourcing(None.Type, key, null)); |
|||
} |
|||
|
|||
[Fact] |
|||
public async Task Should_write_batched() |
|||
{ |
|||
var key1 = DomainId.NewGuid(); |
|||
var key2 = DomainId.NewGuid(); |
|||
|
|||
var bulk = sut.WithBatchContext(None.Type); |
|||
|
|||
await bulk.LoadAsync(new[] { key1, key2 }); |
|||
|
|||
var persistedEvents1 = Save.Events(); |
|||
var persistence1 = bulk.WithEventSourcing(None.Type, key1, persistedEvents1.Write); |
|||
|
|||
var persistedEvents2 = Save.Events(); |
|||
var persistence2 = bulk.WithEventSourcing(None.Type, key2, persistedEvents2.Write); |
|||
|
|||
await persistence1.WriteSnapshotAsync(12); |
|||
await persistence2.WriteSnapshotAsync(12); |
|||
|
|||
A.CallTo(() => snapshotStore.WriteAsync(A<DomainId>._, A<int>._, A<long>._, A<long>._)) |
|||
.MustNotHaveHappened(); |
|||
|
|||
A.CallTo(() => snapshotStore.WriteManyAsync(A<IEnumerable<(DomainId, int, long)>>._)) |
|||
.MustNotHaveHappened(); |
|||
|
|||
await bulk.CommitAsync(); |
|||
await bulk.DisposeAsync(); |
|||
|
|||
A.CallTo(() => snapshotStore.WriteManyAsync(A<IEnumerable<(DomainId, int, long)>>.That.Matches(x => x.Count() == 2))) |
|||
.MustHaveHappenedOnceExactly(); |
|||
} |
|||
|
|||
private void SetupEventStore(Dictionary<DomainId, List<MyEvent>> streams) |
|||
{ |
|||
var storedStreams = new Dictionary<string, IReadOnlyList<StoredEvent>>(); |
|||
|
|||
foreach (var (id, stream) in streams) |
|||
{ |
|||
var storedStream = new List<StoredEvent>(); |
|||
|
|||
var i = 0; |
|||
|
|||
foreach (var @event in stream) |
|||
{ |
|||
var eventData = new EventData("Type", new EnvelopeHeaders(), "Payload"); |
|||
var eventStored = new StoredEvent(id.ToString(), i.ToString(), i, eventData); |
|||
|
|||
storedStream.Add(eventStored); |
|||
|
|||
A.CallTo(() => eventDataFormatter.Parse(eventStored)) |
|||
.Returns(new Envelope<IEvent>(@event)); |
|||
|
|||
A.CallTo(() => eventDataFormatter.ParseIfKnown(eventStored)) |
|||
.Returns(new Envelope<IEvent>(@event)); |
|||
|
|||
i++; |
|||
} |
|||
|
|||
storedStreams[id.ToString()] = storedStream; |
|||
} |
|||
|
|||
var streamNames = streams.Keys.Select(x => x.ToString()).ToArray(); |
|||
|
|||
A.CallTo(() => eventStore.QueryManyAsync(A<IEnumerable<string>>.That.IsSameSequenceAs(streamNames))) |
|||
.Returns(storedStreams); |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue