Browse Source

Two bugfixes.

pull/808/head
Sebastian 4 years ago
parent
commit
15ba9a6d3f
  1. 1
      backend/src/Squidex.Infrastructure.MongoDb/EventSourcing/MongoEventStore_Writer.cs
  2. 14
      backend/src/Squidex.Infrastructure/Commands/CommandExtensions.cs
  3. 31
      backend/src/Squidex.Infrastructure/Commands/DomainObject.cs
  4. 4
      backend/src/Squidex.Infrastructure/Commands/SnapshotList.cs
  5. 2
      backend/src/Squidex/Areas/Api/Controllers/Contents/Models/BulkUpdateContentsJobDto.cs
  6. 8
      backend/tests/Squidex.Infrastructure.Tests/Commands/DomainObjectTests.cs

1
backend/src/Squidex.Infrastructure.MongoDb/EventSourcing/MongoEventStore_Writer.cs

@ -44,7 +44,6 @@ namespace Squidex.Infrastructure.EventSourcing
Guard.NotEmpty(commitId, nameof(commitId)); Guard.NotEmpty(commitId, nameof(commitId));
Guard.NotNullOrEmpty(streamName, nameof(streamName)); Guard.NotNullOrEmpty(streamName, nameof(streamName));
Guard.NotNull(events, nameof(events)); Guard.NotNull(events, nameof(events));
Guard.LessThan(events.Count, MaxCommitSize, "events.Count");
Guard.GreaterEquals(expectedVersion, EtagVersion.Any, nameof(expectedVersion)); Guard.GreaterEquals(expectedVersion, EtagVersion.Any, nameof(expectedVersion));
using (Telemetry.Activities.StartActivity("ContentQueryService/AppendAsync")) using (Telemetry.Activities.StartActivity("ContentQueryService/AppendAsync"))

14
backend/src/Squidex.Infrastructure/Commands/CommandExtensions.cs

@ -5,6 +5,8 @@
// All rights reserved. Licensed under the MIT license. // All rights reserved. Licensed under the MIT license.
// ========================================================================== // ==========================================================================
using Squidex.Infrastructure.EventSourcing;
namespace Squidex.Infrastructure.Commands namespace Squidex.Infrastructure.Commands
{ {
public static class CommandExtensions public static class CommandExtensions
@ -13,5 +15,17 @@ namespace Squidex.Infrastructure.Commands
{ {
return commandMiddleware.HandleAsync(context, x => Task.CompletedTask); return commandMiddleware.HandleAsync(context, x => Task.CompletedTask);
} }
public static Envelope<IEvent> Migrate<T>(this Envelope<IEvent> @event, T snapshot)
{
if (@event.Payload is IMigratedStateEvent<T> migratable)
{
var payload = migratable.Migrate(snapshot);
@event = new Envelope<IEvent>(payload, @event.Headers);
}
return @event;
}
} }
} }

31
backend/src/Squidex.Infrastructure/Commands/DomainObject.cs

@ -67,14 +67,18 @@ namespace Squidex.Infrastructure.Commands
var allEvents = factory.WithEventSourcing(GetType(), UniqueId, @event => var allEvents = factory.WithEventSourcing(GetType(), UniqueId, @event =>
{ {
var newVersion = snapshot.Version + 1; @event = @event.Migrate(snapshot);
if (snapshots.Contains(newVersion)) var (newSnapshot, isChanged) = ApplyEvent(@event, true, snapshot, snapshot.Version, false);
// Can only be null in case of errors or inconsistent streams.
if (newSnapshot != null)
{ {
return false; snapshot = newSnapshot;
} }
return ApplyEvent(@event, true, snapshot, snapshot.Version, false); // If all snapshorts from this one here are valid we can stop.
return newSnapshot != null && !snapshots.ContainsThisAndNewer(newSnapshot.Version);
}); });
await allEvents.ReadAsync(); await allEvents.ReadAsync();
@ -97,14 +101,9 @@ namespace Squidex.Infrastructure.Commands
}), }),
@event => @event =>
{ {
if (@event.Payload is IMigratedStateEvent<T> migratable) @event = @event.Migrate(Snapshot);
{
var payload = migratable.Migrate(Snapshot);
@event = new Envelope<IEvent>(payload, @event.Headers);
}
return ApplyEvent(@event, true, Snapshot, Version, true); return ApplyEvent(@event, true, Snapshot, Version, true).Success;
}); });
} }
@ -146,7 +145,7 @@ namespace Squidex.Infrastructure.Commands
@event.SetAggregateId(uniqueId); @event.SetAggregateId(uniqueId);
if (ApplyEvent(@event, false, Snapshot, Version, true)) if (ApplyEvent(@event, false, Snapshot, Version, true).Success)
{ {
uncomittedEvents.Add(@event); uncomittedEvents.Add(@event);
} }
@ -299,13 +298,13 @@ namespace Squidex.Infrastructure.Commands
snapshots.ResetTo(previousSnapshot, previousVersion); snapshots.ResetTo(previousSnapshot, previousVersion);
} }
private bool ApplyEvent(Envelope<IEvent> @event, bool isLoading, T snapshot, long version, bool clean) private (T?, bool Success) ApplyEvent(Envelope<IEvent> @event, bool isLoading, T snapshot, long version, bool clean)
{ {
if (IsDeleted(snapshot)) if (IsDeleted(snapshot))
{ {
if (!CanRecreate(@event.Payload)) if (!CanRecreate(@event.Payload))
{ {
return false; return default;
} }
snapshot = new T snapshot = new T
@ -324,14 +323,14 @@ namespace Squidex.Infrastructure.Commands
var isChanged = !ReferenceEquals(snapshot, newSnapshot); var isChanged = !ReferenceEquals(snapshot, newSnapshot);
if (!ReferenceEquals(newSnapshot, snapshot)) if (isChanged)
{ {
newSnapshot.Version = newVersion; newSnapshot.Version = newVersion;
snapshots.Add(newSnapshot, newVersion, clean); snapshots.Add(newSnapshot, newVersion, clean);
} }
return isChanged; return (newSnapshot, isChanged);
} }
private async Task ReadAsync() private async Task ReadAsync()

4
backend/src/Squidex.Infrastructure/Commands/SnapshotList.cs

@ -69,11 +69,11 @@ namespace Squidex.Infrastructure.Commands
return (null, false); return (null, false);
} }
public bool Contains(long version) public bool ContainsThisAndNewer(long version)
{ {
var index = GetIndex(version); var index = GetIndex(version);
return items.ElementAtOrDefault(index) != null; return items.Skip(index).All(x => x != null);
} }
public void Add(T snapshot, long version, bool clean = false) public void Add(T snapshot, long version, bool clean = false)

2
backend/src/Squidex/Areas/Api/Controllers/Contents/Models/BulkUpdateContentsJobDto.cs

@ -67,7 +67,7 @@ namespace Squidex.Areas.Api.Controllers.Contents.Models
public BulkUpdateJob ToJob() public BulkUpdateJob ToJob()
{ {
return SimpleMapper.Map(this, new BulkUpdateJob()); return SimpleMapper.Map(this, new BulkUpdateJob { Query = Query });
} }
} }
} }

8
backend/tests/Squidex.Infrastructure.Tests/Commands/DomainObjectTests.cs

@ -453,15 +453,18 @@ namespace Squidex.Infrastructure.Commands
await sut.ExecuteAsync(new CreateAuto { Value = 3 }); await sut.ExecuteAsync(new CreateAuto { Value = 3 });
await sut.ExecuteAsync(new UpdateAuto { Value = 4 }); await sut.ExecuteAsync(new UpdateAuto { Value = 4 });
await sut.ExecuteAsync(new UpdateAuto { Value = 5 });
var version_Empty = await sut.GetSnapshotAsync(EtagVersion.Empty); var version_Empty = await sut.GetSnapshotAsync(EtagVersion.Empty);
var version_0 = await sut.GetSnapshotAsync(0); var version_0 = await sut.GetSnapshotAsync(0);
var version_1 = await sut.GetSnapshotAsync(1); var version_1 = await sut.GetSnapshotAsync(1);
var version_2 = await sut.GetSnapshotAsync(2);
Assert.Empty(sut.GetUncomittedEvents()); Assert.Empty(sut.GetUncomittedEvents());
AssertSnapshot(version_Empty, 0, EtagVersion.Empty); AssertSnapshot(version_Empty, 0, EtagVersion.Empty);
AssertSnapshot(version_0, 3, 0); AssertSnapshot(version_0, 3, 0);
AssertSnapshot(version_1, 4, 1); AssertSnapshot(version_1, 4, 1);
AssertSnapshot(version_2, 5, 2);
A.CallTo(() => persistenceFactory.WithEventSourcing(typeof(MyDomainObject), id, A<HandleEvent>._)) A.CallTo(() => persistenceFactory.WithEventSourcing(typeof(MyDomainObject), id, A<HandleEvent>._))
.MustNotHaveHappened(); .MustNotHaveHappened();
@ -477,18 +480,21 @@ namespace Squidex.Infrastructure.Commands
await sut.ExecuteAsync(new CreateAuto { Value = 3 }); await sut.ExecuteAsync(new CreateAuto { Value = 3 });
await sut.ExecuteAsync(new UpdateAuto { Value = 4 }); await sut.ExecuteAsync(new UpdateAuto { Value = 4 });
await sut.ExecuteAsync(new UpdateAuto { Value = 5 });
var version_Empty = await sut.GetSnapshotAsync(EtagVersion.Empty); var version_Empty = await sut.GetSnapshotAsync(EtagVersion.Empty);
var version_0 = await sut.GetSnapshotAsync(0); var version_0 = await sut.GetSnapshotAsync(0);
var version_1 = await sut.GetSnapshotAsync(1); var version_1 = await sut.GetSnapshotAsync(1);
var version_2 = await sut.GetSnapshotAsync(2);
Assert.Empty(sut.GetUncomittedEvents()); Assert.Empty(sut.GetUncomittedEvents());
AssertSnapshot(version_Empty, 0, EtagVersion.Empty); AssertSnapshot(version_Empty, 0, EtagVersion.Empty);
AssertSnapshot(version_0, 3, 0); AssertSnapshot(version_0, 3, 0);
AssertSnapshot(version_1, 4, 1); AssertSnapshot(version_1, 4, 1);
AssertSnapshot(version_2, 5, 2);
A.CallTo(() => persistenceFactory.WithEventSourcing(typeof(MyDomainObject), id, A<HandleEvent>._)) A.CallTo(() => persistenceFactory.WithEventSourcing(typeof(MyDomainObject), id, A<HandleEvent>._))
.MustHaveHappened(); .MustHaveHappenedOnceExactly();
} }
private static void AssertSnapshot(MyDomainState state, int value, long version, bool isDeleted = false) private static void AssertSnapshot(MyDomainState state, int value, long version, bool isDeleted = false)

Loading…
Cancel
Save