diff --git a/.config/dotnet-tools.json b/.config/dotnet-tools.json index 95adf14..da8092f 100644 --- a/.config/dotnet-tools.json +++ b/.config/dotnet-tools.json @@ -3,10 +3,11 @@ "isRoot": true, "tools": { "gitversion.tool": { - "version": "5.12.0", + "version": "6.1.0", "commands": [ "dotnet-gitversion" - ] + ], + "rollForward": false } } } \ No newline at end of file diff --git a/Changelog.md b/Changelog.md index a0b8f19..f8dfe83 100644 --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,14 @@ # NEventStore.Domain Versions +## vNext + +- Async Methods [#17](https://github.com/NEventStore/NEventStore.Domain/issues/17) +- Updated NEventStore reference to version 10.0.0 + +### Breaking Changes + +- Simplified `IRepository` interface, many overloaded methods are now extension methods in `RepositoryExtensions`. + ## 9.1.1 - Updated NEventStore reference to version 9.1.0 diff --git a/readme.markdown b/Readme.md similarity index 100% rename from readme.markdown rename to Readme.md diff --git a/appveyor.yml b/appveyor.yml index 98a01aa..b41c6fc 100644 --- a/appveyor.yml +++ b/appveyor.yml @@ -40,13 +40,13 @@ build: verbosity: minimal after_build: - cmd: >- - REM dotnet pack ./src/NEventStore.Domain/NEventStore.Domain.Core.csproj -c %CONFIGURATION% --no-build -o artifacts -p:PackageVersion=%GitVersion_NuGetVersion% + REM dotnet pack ./src/NEventStore.Domain/NEventStore.Domain.Core.csproj -c %CONFIGURATION% --no-build -o artifacts -p:PackageVersion=%GitVersion_SemVer% REM use .nuspec file, projects with submodule generate wrong PackageReferences - nuget pack ./src/.nuget/NEventStore.Domain.nuspec -properties "version=%GitVersion_NuGetVersion%;configuration=%CONFIGURATION%" -OutputDirectory artifacts + nuget pack ./src/.nuget/NEventStore.Domain.nuspec -properties "version=%GitVersion_SemVer%;configuration=%CONFIGURATION%" -OutputDirectory artifacts test: assemblies: except: diff --git a/build.ps1 b/build.ps1 index fa51aeb..e6151c4 100644 --- a/build.ps1 +++ b/build.ps1 @@ -17,7 +17,7 @@ dotnet restore ./src/NEventStore.Domain.Core.sln --verbosity m Write-Host "Running GitVersion for the Project" $str = dotnet tool run dotnet-gitversion /updateAssemblyInfo | out-string $json = convertFrom-json $str -$nugetversion = $json.NuGetVersion +$nugetversion = $json.SemVer # Now we need to patch the AssemblyInfo for submodules Write-Host "Running GitVersion for the Dependencies" @@ -25,7 +25,7 @@ dotnet tool run dotnet-gitversion ".\dependencies\NEventStore" /updateAssemblyIn # Build Write-Host "Building: "$nugetversion" "$configuration -dotnet build ./src/NEventStore.Domain.Core.sln -c $configuration --no-restore +dotnet build ./src/NEventStore.Domain.Core.sln -c $configuration --no-restore /p:ContinuousIntegrationBuild=true # Testing if ($runtests -eq "y") { @@ -44,4 +44,4 @@ Write-Host "NuGet Packages creation" #dotnet pack ./src/NEventStore.Domain/NEventStore.Domain.Core.csproj --no-build -c $configuration -o $artifacts -p:NuspecFile="" -p:NuspecProperties="pippo=$configuration;version=$nugetversion" #Write-Host nuget pack ./src/NEventStore.Domain/NEventStore.Domain.Core.csproj -properties "version=$nugetversion;configuration=$configuration" -nuget pack ./src/.nuget/NEventStore.Domain.nuspec -properties "version=$nugetversion;configuration=$configuration" -OutputDirectory $artifacts \ No newline at end of file +nuget pack ./src/.nuget/NEventStore.Domain.nuspec -properties "version=$nugetversion;configuration=$configuration" -OutputDirectory $artifacts -Symbols -SymbolPackageFormat snupkg \ No newline at end of file diff --git a/dependencies/NEventStore b/dependencies/NEventStore index a7d4848..aaef5b5 160000 --- a/dependencies/NEventStore +++ b/dependencies/NEventStore @@ -1 +1 @@ -Subproject commit a7d4848ac5ec6b5d6f1e7bbc19a2381ca5d0e721 +Subproject commit aaef5b5a8f07d5774a6fd7291cb09c1c676a8f83 diff --git a/icon.png b/icon.png new file mode 100644 index 0000000..c6c977d Binary files /dev/null and b/icon.png differ diff --git a/src/.editorconfig b/src/.editorconfig new file mode 100644 index 0000000..73679fd --- /dev/null +++ b/src/.editorconfig @@ -0,0 +1,33 @@ +[*] +end_of_line = crlf +indent_style = space +indent_size = 4 + +[*.xml] +indent_style = space + +[*.cs] +csharp_new_line_before_open_brace = all + +# RCS1229: Use async/await when necessary. +dotnet_diagnostic.RCS1229.severity = error + +# IDE0290: Use primary constructor +csharp_style_prefer_primary_constructors = false + +# IDE0028: Simplify collection initialization +dotnet_style_collection_initializer = false + +# IDE0305: Simplify collection initialization +dotnet_diagnostic.IDE0305.severity = none + +# Logging Warnings: temporary disabled, enable them again later on + +# CA1848: Use the LoggerMessage delegates +dotnet_diagnostic.CA1848.severity = none + +# CA2254: Template should be a static expression +dotnet_diagnostic.CA2254.severity = none + +# CS1591: Missing XML comment for publicly visible type or member +dotnet_diagnostic.CS1591.severity = none \ No newline at end of file diff --git a/src/.nuget/NEventStore.Domain.nuspec b/src/.nuget/NEventStore.Domain.nuspec index 73c7e77..55e0da5 100644 --- a/src/.nuget/NEventStore.Domain.nuspec +++ b/src/.nuget/NEventStore.Domain.nuspec @@ -1,36 +1,45 @@ - - NEventStore.Domain - $version$ - NEventStore Domain - NEventStore Dev Team - NEventStore Dev Team - http://neventstore.org - false - A recipe for implementing event sourced domain objects with NEventStore. - A recipe for implementing event sourced domain objects with NEventStore. - es-US - events, event sourcing, cqrs, storage, persistence, database, ddd, domain - - - - - - - - - - - - - - - + with pre-release of the actual version. Ex, if we are on 6.0.0 unstable, dependency from NEventStore should + be latest stable version so it can take prerelease of 6.0.0 version. + With (5.4 we are telling NuGet that we need a version greater than 5.4 so we are allowing all the prerelease + 6.x.x versions + --> + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/Directory.Build.props b/src/Directory.Build.props new file mode 100644 index 0000000..a69983f --- /dev/null +++ b/src/Directory.Build.props @@ -0,0 +1,7 @@ + + + 13.0 + enable + enable + + \ No newline at end of file diff --git a/src/NEventStore.Domain.Core.sln b/src/NEventStore.Domain.Core.sln index ddd198b..8006b97 100644 --- a/src/NEventStore.Domain.Core.sln +++ b/src/NEventStore.Domain.Core.sln @@ -15,12 +15,13 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NEventStore.Domain.Core.Tes EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{A4010EA6-0ECC-445C-BECC-02C45F0427B7}" ProjectSection(SolutionItems) = preProject + .editorconfig = .editorconfig ..\appveyor.yml = ..\appveyor.yml ..\build.ps1 = ..\build.ps1 ..\Changelog.md = ..\Changelog.md ..\GitVersion.yml = ..\GitVersion.yml .nuget\NEventStore.Domain.nuspec = .nuget\NEventStore.Domain.nuspec - ..\readme.markdown = ..\readme.markdown + ..\Readme.md = ..\Readme.md EndProjectSection EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = ".nuget", ".nuget", "{83B575A7-6DA1-4E0A-8331-B59836F898D7}" diff --git a/src/NEventStore.Domain.Tests/NEventStore.Domain.Core.Tests.csproj b/src/NEventStore.Domain.Tests/NEventStore.Domain.Core.Tests.csproj index 34860b1..6074d7b 100644 --- a/src/NEventStore.Domain.Tests/NEventStore.Domain.Core.Tests.csproj +++ b/src/NEventStore.Domain.Tests/NEventStore.Domain.Core.Tests.csproj @@ -1,7 +1,7 @@  - net7.0;net462 + net8.0;net472 false exe @@ -15,12 +15,12 @@ - - - - - - + + + + + + diff --git a/src/NEventStore.Domain.Tests/Persistence/AggregateFactory.cs b/src/NEventStore.Domain.Tests/Persistence/AggregateFactory.cs index 7973308..b89b067 100644 --- a/src/NEventStore.Domain.Tests/Persistence/AggregateFactory.cs +++ b/src/NEventStore.Domain.Tests/Persistence/AggregateFactory.cs @@ -1,17 +1,21 @@ -namespace NEventStore.Domain.Tests.Persistence -{ - using System; - using System.Reflection; - using NEventStore.Domain.Persistence; +using System.Reflection; +using NEventStore.Domain.Persistence; +namespace NEventStore.Domain.Tests.Persistence +{ internal class AggregateFactory : IConstructAggregates - { - public IAggregate Build(Type type, Guid id, IMemento snapshot) - { - ConstructorInfo constructor = type.GetConstructor( - BindingFlags.NonPublic | BindingFlags.Instance, null, new[] { typeof(Guid) }, null); + { + public IAggregate Build(Type type, Guid id, IMemento? snapshot) + { + var constructor = type.GetConstructor( + BindingFlags.NonPublic | BindingFlags.Instance, null, [typeof(Guid)], null); + + if (constructor == null) + { + throw new NotSupportedException(string.Format("The type '{0}' does not have a constructor accepting a Guid.", type)); + } - return constructor.Invoke(new object[] { id }) as IAggregate; - } - } + return (IAggregate)constructor!.Invoke([id]); + } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain.Tests/Persistence/IAggregatePersistenceTests.Async.cs b/src/NEventStore.Domain.Tests/Persistence/IAggregatePersistenceTests.Async.cs new file mode 100644 index 0000000..fd22fed --- /dev/null +++ b/src/NEventStore.Domain.Tests/Persistence/IAggregatePersistenceTests.Async.cs @@ -0,0 +1,433 @@ +using NEventStore.Domain.Core; +using NEventStore.Domain.Persistence; +using NEventStore.Domain.Persistence.EventStore; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using FluentAssertions; +#if MSTEST +using Microsoft.VisualStudio.TestTools.UnitTesting; +#endif +#if NUNIT +#endif +#if XUNIT +using Xunit; +using Xunit.Should; +#endif + +namespace NEventStore.Domain.Tests.Persistence.Async +{ +#if MSTEST + [TestClass] +#endif + public abstract class using_a_configured_repository : SpecificationBase + { + protected IRepository? _repository; + + protected IStoreEvents? _storeEvents; + + protected override void Context() + { + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository = CreateRepository(); + } + + protected EventStoreRepository CreateRepository() + { + return new EventStoreRepository(_storeEvents!, new AggregateFactory(), new ConflictDetector()); + } + } + + public class when_an_aggregate_is_persisted : using_a_configured_repository + { + private TestAggregate? _testAggregate; + + private Guid _id; + + protected override void Context() + { + base.Context(); + _id = Guid.NewGuid(); + _testAggregate = new TestAggregate(_id, "Test"); + } + + protected override Task BecauseAsync() + { + return _repository!.SaveAsync(_testAggregate!, Guid.NewGuid(), null); + } + + [Fact] + public async Task should_be_returned_when_loaded_by_id() + { + (await _repository!.GetByIdAsync(_id).ConfigureAwait(false)) + .Should().NotBeNull(); + } + + [Fact] + public async Task version_should_be_one() + { + (await _repository!.GetByIdAsync(_id).ConfigureAwait(false)) + .Version.Should().Be(1); + } + + [Fact] + public async Task id_should_be_set() + { + (await _repository!.GetByIdAsync(_id).ConfigureAwait(false)) + .Id.Should().Be(_id); + } + + [Fact] + public async Task should_have_name_set() + { + (await _repository!.GetByIdAsync(_id).ConfigureAwait(false)) + .Name.Should().Be(_testAggregate!.Name); + } + } + + public class when_a_persisted_aggregate_is_updated : using_a_configured_repository + { + private Guid _id; + + private const string NewName = "UpdatedName"; + + protected override Task ContextAsync() + { + base.Context(); + _id = Guid.NewGuid(); + return _repository!.SaveAsync(new TestAggregate(_id, "Test"), Guid.NewGuid(), null); + } + + protected override Task BecauseAsync() + { + var aggregate = _repository!.GetById(_id); + aggregate.ChangeName(NewName); + return _repository!.SaveAsync(aggregate, Guid.NewGuid(), null); + } + + [Fact] + public async Task should_have_updated_name() + { + (await _repository!.GetByIdAsync(_id)) + .Name.Should().Be(NewName); + } + + [Fact] + public async Task should_have_updated_version() + { + (await _repository!.GetByIdAsync(_id)) + .Version.Should().Be(2); + } + } + + public class when_a_loading_a_specific_aggregate_version : using_a_configured_repository + { + private Guid _id; + + private const string VersionOneName = "Test"; + private const string NewName = "UpdatedName"; + + protected override Task ContextAsync() + { + base.Context(); + _id = Guid.NewGuid(); + return _repository!.SaveAsync(new TestAggregate(_id, VersionOneName), Guid.NewGuid(), null); + } + + protected override async Task BecauseAsync() + { + var aggregate = _repository!.GetById(_id); + aggregate.ChangeName(NewName); + await _repository!.SaveAsync(aggregate, Guid.NewGuid(), null).ConfigureAwait(false); + _repository!.Dispose(); + } + + [Fact] + public async Task should_be_able_to_load_initial_version() + { + (await _repository!.GetByIdAsync(_id, 1)) + .Name.Should().Be(VersionOneName); + } + } + + public class when_an_aggregate_is_persisted_to_specific_bucket : using_a_configured_repository + { + private TestAggregate? _testAggregate; + + private Guid _id; + + private string _bucket = "TenantB"; + + protected override void Context() + { + base.Context(); + _id = Guid.NewGuid(); + _testAggregate = new TestAggregate(_id, "Test"); + } + + protected override Task BecauseAsync() + { + return _repository!.SaveAsync(_bucket, _testAggregate!, Guid.NewGuid(), null); + } + + [Fact] + public async Task should_be_returned_when_loaded_by_id() + { + (await _repository!.GetByIdAsync(_bucket, _id).ConfigureAwait(false)) + .Name.Should().Be(_testAggregate!.Name); + } + } + + /// + /// + /// Idempotency Check: + /// Internally a DuplicateCommitException will be raised and catch by the repository, + /// the whole commit will be discarded, we assume the it's the same commit issued twice. + /// + /// Issue: #4 + /// + public class when_an_aggregate_is_persisted_using_the_same_commitId_twice : using_a_configured_repository + { + private TestAggregate? _testAggregate; + + private Guid _id; + + protected override void Context() + { + base.Context(); + _id = Guid.NewGuid(); + _testAggregate = new TestAggregate(_id, "Test"); + } + + protected override async Task BecauseAsync() + { + var commitId = Guid.NewGuid(); + await _repository!.SaveAsync(_testAggregate!, commitId, null).ConfigureAwait(false); + + _testAggregate!.ChangeName("one"); + + await _repository!.SaveAsync(_testAggregate, commitId).ConfigureAwait(false); + } + + [Fact] + public async Task the_second_commit_was_silently_discarded_and_not_written_to_database() + { + var aggregate = (await _repository!.GetByIdAsync(_id).ConfigureAwait(false)); + aggregate.Name.Should().Be("Test"); + aggregate.Version.Should().Be(1); + } + + [Fact] + public void the_aggregate_still_has_pending_changes() + { + var uncommittedEvents = ((IAggregate)_testAggregate!).GetUncommittedEvents(); + uncommittedEvents.Count.Should().BeGreaterThan(0); + var enumerator = uncommittedEvents.GetEnumerator(); + enumerator.MoveNext(); + enumerator.Current.Should().BeOfType(); + } + } + + public class when_an_aggregate_is_persisted_by_two_repositories : SpecificationBase + { + protected IRepository? _repository1; + protected IRepository? _repository2; + + protected IStoreEvents? _storeEvents; + private Guid _aggregateId; + private TestAggregate? aggregate; + private Exception? _thrown; + + protected override void Context() + { + base.Context(); + + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository1 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + _repository2 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + + _aggregateId = Guid.NewGuid(); + aggregate = new TestAggregate(_aggregateId, "my name is.."); + } + + protected override async Task BecauseAsync() + { + await _repository1!.SaveAsync(aggregate!, Guid.NewGuid()).ConfigureAwait(false); + aggregate!.ChangeName("one"); + + _thrown = await Catch.ExceptionAsync(() => _repository2!.SaveAsync(aggregate, Guid.NewGuid())); + } + + [Fact] + public void should_not_throw_a_ConflictingCommandException() + { + _thrown.Should().BeNull(); + } + + [Fact] + public async Task should_have_updated_name_if_loaded_by_repository_that_saved_it_last() + { + (await _repository2!.GetByIdAsync(_aggregateId).ConfigureAwait(false)) + .Name.Should().Be("one"); + } + + /// + /// current repository implementation act as a session cache! + /// + [Fact] + public async Task should_have_original_name_if_loaded_by_repository_that_saved_it_first() + { + (await _repository1!.GetByIdAsync(_aggregateId).ConfigureAwait(false)) + .Name.Should().Be("my name is.."); + } + } + + public class when_an_aggregate_is_persisted_concurrently_by_two_clients : SpecificationBase + { + protected IRepository? _repository1; + protected IRepository? _repository2; + + protected IStoreEvents? _storeEvents; + private Guid _aggregateId; + private Exception? _thrown; + + protected override Task ContextAsync() + { + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository1 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + _repository2 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + + _aggregateId = Guid.NewGuid(); + var aggregate = new TestAggregate(_aggregateId, "my name is.."); + return _repository1.SaveAsync(aggregate, Guid.NewGuid()); + } + + protected override async Task BecauseAsync() + { + var agg1 = await _repository1!.GetByIdAsync(_aggregateId).ConfigureAwait(false); + var agg2 = await _repository2!.GetByIdAsync(_aggregateId).ConfigureAwait(false); + agg1.ChangeName("one"); + agg2.ChangeName("two"); + + await _repository1!.SaveAsync(agg1, Guid.NewGuid()).ConfigureAwait(false); + + _thrown = await Catch.ExceptionAsync(() => _repository2!.SaveAsync(agg2, Guid.NewGuid())).ConfigureAwait(false); + } + + [Fact] + public void should_throw_a_ConflictingCommandException() + { + _thrown.Should().BeOfType(); + } + } + + public class when_an_aggregate_is_persisted_concurrently_by_two_clients_using_new_operator : SpecificationBase + { + protected IRepository? _repository1; + protected IRepository? _repository2; + + protected IStoreEvents? _storeEvents; + private Guid _aggregateId; + private Exception? _thrown; + + protected override Task ContextAsync() + { + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository1 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + _repository2 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + + _aggregateId = Guid.NewGuid(); + var aggregate = new TestAggregate(_aggregateId, "my name is.."); + return _repository1.SaveAsync(aggregate, Guid.NewGuid()); + } + + protected override async Task BecauseAsync() + { + var agg1 = await _repository1!.GetByIdAsync(_aggregateId).ConfigureAwait(false); + var agg2 = new TestAggregate(_aggregateId, "two"); + agg1.ChangeName("one"); + + await _repository1!.SaveAsync(agg1, Guid.NewGuid()).ConfigureAwait(false); + + _thrown = await Catch.ExceptionAsync(() => _repository2!.SaveAsync(agg2, Guid.NewGuid())).ConfigureAwait(false); + } + + [Fact] + public void should_throw_a_ConflictingCommandException() + { + _thrown.Should().BeOfType(); + } + } + + public class when_the_same_aggregate_is_created_and_persisted_concurrently : SpecificationBase + { + protected IRepository? _repository1; + protected IRepository? _repository2; + + protected IStoreEvents? _storeEvents; + private Guid _aggregateId; + private Exception? _thrown; + + protected override void Context() + { + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository1 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + _repository2 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + + _aggregateId = Guid.NewGuid(); + } + + protected override async Task BecauseAsync() + { + var agg1 = new TestAggregate(_aggregateId, "one"); + var agg2 = new TestAggregate(_aggregateId, "two"); + + await _repository1!.SaveAsync(agg1, Guid.NewGuid()).ConfigureAwait(false); + + _thrown = await Catch.ExceptionAsync(() => _repository2!.SaveAsync(agg2, Guid.NewGuid())).ConfigureAwait(false); + } + + [Fact] + public void should_throw_a_ConflictingCommandException() + { + _thrown.Should().BeOfType(); + } + } + + public class when_aggregate_is_reloaded_with_snapshot : using_a_configured_repository + { + private TestAggregate? _testAggregate; + private TestAggregate? _reloadedAggregate; + private Guid _id; + + protected override Task ContextAsync() + { + base.Context(); + _id = Guid.NewGuid(); + _testAggregate = new TestAggregate(_id, "Test"); + return _repository!.SaveAsync(_testAggregate, Guid.NewGuid()); //save at version 1. + } + + protected override async Task BecauseAsync() + { + var otherRepository = CreateRepository(); + var aggregate = await otherRepository.GetByIdAsync(_id).ConfigureAwait(false); //load at version 1 + aggregate.ChangeName("Name changed"); + await otherRepository.SaveAsync(Bucket.Default, aggregate, Guid.NewGuid()).ConfigureAwait(false); //save in version 2 + //Now save the snapshot. + + var memento = ((IAggregate)aggregate).GetSnapshot(); + var snapshot = new Snapshot(Bucket.Default, aggregate.Id.ToString(), aggregate.Version, memento!); + + await _storeEvents!.Advanced.AddSnapshotAsync(snapshot, CancellationToken.None).ConfigureAwait(false); + + //now reload, + _reloadedAggregate = await otherRepository.GetByIdAsync(_id).ConfigureAwait(false); + } + + [Fact] + public void should_have_correct_version() + { + _reloadedAggregate!.Version.Should().Be(2); + } + } +} \ No newline at end of file diff --git a/src/NEventStore.Domain.Tests/Persistence/IAggregatePersistenceTests.cs b/src/NEventStore.Domain.Tests/Persistence/IAggregatePersistenceTests.cs index c18c842..7722d45 100644 --- a/src/NEventStore.Domain.Tests/Persistence/IAggregatePersistenceTests.cs +++ b/src/NEventStore.Domain.Tests/Persistence/IAggregatePersistenceTests.cs @@ -1,47 +1,45 @@ -namespace NEventStore.Domain.Tests.Persistence -{ - using System; - using NEventStore.Domain.Core; - using NEventStore.Domain.Persistence; - using NEventStore.Domain.Persistence.EventStore; - using NEventStore.Persistence.AcceptanceTests; - using NEventStore.Persistence.AcceptanceTests.BDD; - using FluentAssertions; +using NEventStore.Domain.Core; +using NEventStore.Domain.Persistence; +using NEventStore.Domain.Persistence.EventStore; +using NEventStore.Persistence.AcceptanceTests; +using NEventStore.Persistence.AcceptanceTests.BDD; +using FluentAssertions; #if MSTEST - using Microsoft.VisualStudio.TestTools.UnitTesting; +using Microsoft.VisualStudio.TestTools.UnitTesting; #endif #if NUNIT - using NUnit.Framework; #endif #if XUNIT - using Xunit; - using Xunit.Should; +using Xunit; +using Xunit.Should; #endif +namespace NEventStore.Domain.Tests.Persistence +{ #if MSTEST [TestClass] #endif public abstract class using_a_configured_repository : SpecificationBase { - protected IRepository _repository; + protected IRepository? _repository; - protected IStoreEvents _storeEvents; + protected IStoreEvents? _storeEvents; protected override void Context() { - this._storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); - this._repository = CreateRepository(); + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository = CreateRepository(); } protected EventStoreRepository CreateRepository() { - return new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); + return new EventStoreRepository(_storeEvents!, new AggregateFactory(), new ConflictDetector()); } } public class when_an_aggregate_is_persisted : using_a_configured_repository { - private TestAggregate _testAggregate; + private TestAggregate? _testAggregate; private Guid _id; @@ -54,31 +52,31 @@ protected override void Context() protected override void Because() { - _repository.Save(_testAggregate, Guid.NewGuid(), null); + _repository!.Save(_testAggregate!, Guid.NewGuid(), null); } [Fact] public void should_be_returned_when_loaded_by_id() { - _repository.GetById(_id).Should().NotBeNull(); + _repository!.GetById(_id).Should().NotBeNull(); } [Fact] public void version_should_be_one() { - _repository.GetById(_id).Version.Should().Be(1); + _repository!.GetById(_id).Version.Should().Be(1); } [Fact] public void id_should_be_set() { - _repository.GetById(_id).Id.Should().Be(_id); + _repository!.GetById(_id).Id.Should().Be(_id); } [Fact] public void should_have_name_set() { - _repository.GetById(_id).Name.Should().Be(_testAggregate.Name); + _repository!.GetById(_id).Name.Should().Be(_testAggregate!.Name); } } @@ -92,26 +90,26 @@ protected override void Context() { base.Context(); _id = Guid.NewGuid(); - _repository.Save(new TestAggregate(_id, "Test"), Guid.NewGuid(), null); + _repository!.Save(new TestAggregate(_id, "Test"), Guid.NewGuid(), null); } protected override void Because() { - var aggregate = _repository.GetById(_id); + var aggregate = _repository!.GetById(_id); aggregate.ChangeName(NewName); - _repository.Save(aggregate, Guid.NewGuid(), null); + _repository!.Save(aggregate, Guid.NewGuid(), null); } [Fact] public void should_have_updated_name() { - _repository.GetById(_id).Name.Should().Be(NewName); + _repository!.GetById(_id).Name.Should().Be(NewName); } [Fact] public void should_have_updated_version() { - _repository.GetById(_id).Version.Should().Be(2); + _repository!.GetById(_id).Version.Should().Be(2); } } @@ -126,62 +124,62 @@ protected override void Context() { base.Context(); _id = Guid.NewGuid(); - _repository.Save(new TestAggregate(_id, VersionOneName), Guid.NewGuid(), null); + _repository!.Save(new TestAggregate(_id, VersionOneName), Guid.NewGuid(), null); } protected override void Because() { - var aggregate = _repository.GetById(_id); + var aggregate = _repository!.GetById(_id); aggregate.ChangeName(NewName); - _repository.Save(aggregate, Guid.NewGuid(), null); - _repository.Dispose(); + _repository!.Save(aggregate, Guid.NewGuid(), null); + _repository!.Dispose(); } [Fact] public void should_be_able_to_load_initial_version() { - _repository.GetById(_id, 1).Name.Should().Be(VersionOneName); + _repository!.GetById(_id, 1).Name.Should().Be(VersionOneName); } } public class when_an_aggregate_is_persisted_to_specific_bucket : using_a_configured_repository { - private TestAggregate _testAggregate; + private TestAggregate? _testAggregate; private Guid _id; - private string _bucket; + private string _bucket = "TenantB"; protected override void Context() { base.Context(); _id = Guid.NewGuid(); - _bucket = "TenantB"; _testAggregate = new TestAggregate(_id, "Test"); } protected override void Because() { - _repository.Save(_bucket, _testAggregate, Guid.NewGuid(), null); + _repository!.Save(_bucket, _testAggregate!, Guid.NewGuid(), null); } [Fact] public void should_be_returned_when_loaded_by_id() { - _repository.GetById(_bucket, _id).Name.Should().Be(_testAggregate.Name); + _repository!.GetById(_bucket, _id).Name.Should().Be(_testAggregate!.Name); } } /// + /// /// Idempotency Check: /// Internally a DuplicateCommitException will be raised and catch by the repository, /// the whole commit will be discarded, we assume the it's the same commit issued twice. - /// - /// Issue: #4 + /// + /// Issue: #4 /// public class when_an_aggregate_is_persisted_using_the_same_commitId_twice : using_a_configured_repository { - private TestAggregate _testAggregate; + private TestAggregate? _testAggregate; private Guid _id; @@ -195,17 +193,17 @@ protected override void Context() protected override void Because() { var commitId = Guid.NewGuid(); - _repository.Save(_testAggregate, commitId, null); + _repository!.Save(_testAggregate!, commitId, null); - _testAggregate.ChangeName("one"); + _testAggregate!.ChangeName("one"); - _repository.Save(_testAggregate, commitId); + _repository!.Save(_testAggregate, commitId); } [Fact] public void the_second_commit_was_silently_discarded_and_not_written_to_database() { - var aggregate = _repository.GetById(_id); + var aggregate = _repository!.GetById(_id); aggregate.Name.Should().Be("Test"); aggregate.Version.Should().Be(1); } @@ -213,7 +211,7 @@ public void the_second_commit_was_silently_discarded_and_not_written_to_database [Fact] public void the_aggregate_still_has_pending_changes() { - var uncommittedEvents = ((IAggregate)_testAggregate).GetUncommittedEvents(); + var uncommittedEvents = ((IAggregate)_testAggregate!).GetUncommittedEvents(); uncommittedEvents.Count.Should().BeGreaterThan(0); var enumerator = uncommittedEvents.GetEnumerator(); enumerator.MoveNext(); @@ -223,21 +221,21 @@ public void the_aggregate_still_has_pending_changes() public class when_an_aggregate_is_persisted_by_two_repositories : SpecificationBase { - protected IRepository _repository1; - protected IRepository _repository2; + protected IRepository? _repository1; + protected IRepository? _repository2; - protected IStoreEvents _storeEvents; + protected IStoreEvents? _storeEvents; private Guid _aggregateId; - private TestAggregate aggregate; - private Exception _thrown; + private TestAggregate? aggregate; + private Exception? _thrown; protected override void Context() { base.Context(); - this._storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); - this._repository1 = new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); - this._repository2 = new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository1 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + _repository2 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); _aggregateId = Guid.NewGuid(); aggregate = new TestAggregate(_aggregateId, "my name is.."); @@ -245,10 +243,10 @@ protected override void Context() protected override void Because() { - _repository1.Save(aggregate, Guid.NewGuid()); - aggregate.ChangeName("one"); + _repository1!.Save(aggregate!, Guid.NewGuid()); + aggregate!.ChangeName("one"); - _thrown = Catch.Exception(() => _repository2.Save(aggregate, Guid.NewGuid())); + _thrown = Catch.Exception(() => _repository2!.Save(aggregate, Guid.NewGuid())); } [Fact] @@ -260,7 +258,7 @@ public void should_not_throw_a_ConflictingCommandException() [Fact] public void should_have_updated_name_if_loaded_by_repository_that_saved_it_last() { - _repository2.GetById(_aggregateId).Name.Should().Be("one"); + _repository2!.GetById(_aggregateId).Name.Should().Be("one"); } /// @@ -269,26 +267,26 @@ public void should_have_updated_name_if_loaded_by_repository_that_saved_it_last( [Fact] public void should_have_original_name_if_loaded_by_repository_that_saved_it_first() { - _repository1.GetById(_aggregateId).Name.Should().Be("my name is.."); + _repository1!.GetById(_aggregateId).Name.Should().Be("my name is.."); } } public class when_an_aggregate_is_persisted_concurrently_by_two_clients : SpecificationBase { - protected IRepository _repository1; - protected IRepository _repository2; + protected IRepository? _repository1; + protected IRepository? _repository2; - protected IStoreEvents _storeEvents; + protected IStoreEvents? _storeEvents; private Guid _aggregateId; - private Exception _thrown; + private Exception? _thrown; protected override void Context() { base.Context(); - this._storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); - this._repository1 = new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); - this._repository2 = new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository1 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + _repository2 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); _aggregateId = Guid.NewGuid(); var aggregate = new TestAggregate(_aggregateId, "my name is.."); @@ -297,14 +295,14 @@ protected override void Context() protected override void Because() { - var agg1 = _repository1.GetById(_aggregateId); - var agg2 = _repository2.GetById(_aggregateId); + var agg1 = _repository1!.GetById(_aggregateId); + var agg2 = _repository2!.GetById(_aggregateId); agg1.ChangeName("one"); agg2.ChangeName("two"); - _repository1.Save(agg1, Guid.NewGuid()); + _repository1!.Save(agg1, Guid.NewGuid()); - _thrown = Catch.Exception(() => _repository2.Save(agg2, Guid.NewGuid())); + _thrown = Catch.Exception(() => _repository2!.Save(agg2, Guid.NewGuid())); } [Fact] @@ -316,20 +314,20 @@ public void should_throw_a_ConflictingCommandException() public class when_an_aggregate_is_persisted_concurrently_by_two_clients_using_new_operator : SpecificationBase { - protected IRepository _repository1; - protected IRepository _repository2; + protected IRepository? _repository1; + protected IRepository? _repository2; - protected IStoreEvents _storeEvents; + protected IStoreEvents? _storeEvents; private Guid _aggregateId; - private Exception _thrown; + private Exception? _thrown; protected override void Context() { base.Context(); - this._storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); - this._repository1 = new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); - this._repository2 = new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository1 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + _repository2 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); _aggregateId = Guid.NewGuid(); var aggregate = new TestAggregate(_aggregateId, "my name is.."); @@ -338,13 +336,13 @@ protected override void Context() protected override void Because() { - var agg1 = _repository1.GetById(_aggregateId); + var agg1 = _repository1!.GetById(_aggregateId); var agg2 = new TestAggregate(_aggregateId, "two"); agg1.ChangeName("one"); - _repository1.Save(agg1, Guid.NewGuid()); + _repository1!.Save(agg1, Guid.NewGuid()); - _thrown = Catch.Exception(() => _repository2.Save(agg2, Guid.NewGuid())); + _thrown = Catch.Exception(() => _repository2!.Save(agg2, Guid.NewGuid())); } [Fact] @@ -356,20 +354,20 @@ public void should_throw_a_ConflictingCommandException() public class when_the_same_aggregate_is_created_and_persisted_concurrently : SpecificationBase { - protected IRepository _repository1; - protected IRepository _repository2; + protected IRepository? _repository1; + protected IRepository? _repository2; - protected IStoreEvents _storeEvents; + protected IStoreEvents? _storeEvents; private Guid _aggregateId; - private Exception _thrown; + private Exception? _thrown; protected override void Context() { base.Context(); - this._storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); - this._repository1 = new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); - this._repository2 = new EventStoreRepository(this._storeEvents, new AggregateFactory(), new ConflictDetector()); + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository1 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); + _repository2 = new EventStoreRepository(_storeEvents, new AggregateFactory(), new ConflictDetector()); _aggregateId = Guid.NewGuid(); } @@ -380,9 +378,9 @@ protected override void Because() var agg1 = new TestAggregate(_aggregateId, "one"); var agg2 = new TestAggregate(_aggregateId, "two"); - _repository1.Save(agg1, Guid.NewGuid()); + _repository1!.Save(agg1, Guid.NewGuid()); - _thrown = Catch.Exception(() => _repository2.Save(agg2, Guid.NewGuid())); + _thrown = Catch.Exception(() => _repository2!.Save(agg2, Guid.NewGuid())); } [Fact] @@ -394,8 +392,8 @@ public void should_throw_a_ConflictingCommandException() public class when_aggregate_is_reloaded_with_snapshot : using_a_configured_repository { - private TestAggregate _testAggregate; - private TestAggregate _reloadedAggregate; + private TestAggregate? _testAggregate; + private TestAggregate? _reloadedAggregate; private Guid _id; protected override void Context() @@ -403,7 +401,7 @@ protected override void Context() base.Context(); _id = Guid.NewGuid(); _testAggregate = new TestAggregate(_id, "Test"); - _repository.Save(_testAggregate, Guid.NewGuid()); //save at version 1. + _repository!.Save(_testAggregate, Guid.NewGuid()); //save at version 1. } protected override void Because() @@ -415,9 +413,9 @@ protected override void Because() //Now save the snapshot. var memento = ((IAggregate)aggregate).GetSnapshot(); - var snapshot = new Snapshot(Bucket.Default, aggregate.Id.ToString(), aggregate.Version, memento); + var snapshot = new Snapshot(Bucket.Default, aggregate.Id.ToString(), aggregate.Version, memento!); - _storeEvents.Advanced.AddSnapshot(snapshot); + _storeEvents!.Advanced.AddSnapshot(snapshot); //now reload, _reloadedAggregate = otherRepository.GetById(_id); @@ -426,7 +424,7 @@ protected override void Because() [Fact] public void should_have_correct_version() { - _reloadedAggregate.Version.Should().Be(2); + _reloadedAggregate!.Version.Should().Be(2); } } } \ No newline at end of file diff --git a/src/NEventStore.Domain.Tests/Persistence/ISagaPersistenceTests.Async.cs b/src/NEventStore.Domain.Tests/Persistence/ISagaPersistenceTests.Async.cs new file mode 100644 index 0000000..ad20291 --- /dev/null +++ b/src/NEventStore.Domain.Tests/Persistence/ISagaPersistenceTests.Async.cs @@ -0,0 +1,56 @@ +using NEventStore.Domain.Persistence; +using NEventStore.Domain.Persistence.EventStore; +using NEventStore.Persistence.AcceptanceTests.BDD; +using FluentAssertions; +#if MSTEST +using Microsoft.VisualStudio.TestTools.UnitTesting; +#endif +#if NUNIT +#endif +#if XUNIT +using Xunit; +using Xunit.Should; +#endif + +namespace NEventStore.Domain.Tests.Persistence.EventStore.Async +{ +#if MSTEST + [TestClass] +#endif + public class using_a_sagaeventstorerepository : SpecificationBase + { + protected ISagaRepository? _repository; + + protected IStoreEvents? _storeEvents; + + protected override void Context() + { + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository = new SagaEventStoreRepository(_storeEvents, new SagaFactory()); + } + } + + public class when_a_saga_is_loaded : using_a_sagaeventstorerepository + { + private TestSaga? _testSaga; + + private string _id = "something"; + + protected override void Context() + { + base.Context(); + _testSaga = new TestSaga(_id); + } + + protected override Task BecauseAsync() + { + return _repository!.SaveAsync(_testSaga!, Guid.NewGuid(), null); + } + + [Fact] + public async Task should_be_returned_when_loaded_by_id() + { + (await _repository!.GetByIdAsync(_id).ConfigureAwait(false)).Id.Should().Be(_testSaga!.Id); + } + } +} diff --git a/src/NEventStore.Domain.Tests/Persistence/ISagaPersistenceTests.cs b/src/NEventStore.Domain.Tests/Persistence/ISagaPersistenceTests.cs index f51625a..d373b47 100644 --- a/src/NEventStore.Domain.Tests/Persistence/ISagaPersistenceTests.cs +++ b/src/NEventStore.Domain.Tests/Persistence/ISagaPersistenceTests.cs @@ -1,59 +1,56 @@ -namespace NEventStore.Domain.Tests.Persistence.EventStore -{ - using System; - using NEventStore.Domain.Persistence; - using NEventStore.Domain.Persistence.EventStore; - using NEventStore.Persistence.AcceptanceTests.BDD; - using FluentAssertions; +using NEventStore.Domain.Persistence; +using NEventStore.Domain.Persistence.EventStore; +using NEventStore.Persistence.AcceptanceTests.BDD; +using FluentAssertions; #if MSTEST - using Microsoft.VisualStudio.TestTools.UnitTesting; +using Microsoft.VisualStudio.TestTools.UnitTesting; #endif #if NUNIT - using NUnit.Framework; #endif #if XUNIT - using Xunit; - using Xunit.Should; +using Xunit; +using Xunit.Should; #endif +namespace NEventStore.Domain.Tests.Persistence.EventStore +{ #if MSTEST [TestClass] #endif public class using_a_sagaeventstorerepository : SpecificationBase - { - protected ISagaRepository _repository; - - protected IStoreEvents _storeEvents; - - protected override void Context() - { - this._storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); - this._repository = new SagaEventStoreRepository(this._storeEvents, new SagaFactory()); - } - } - - public class when_a_saga_is_loaded : using_a_sagaeventstorerepository - { - private TestSaga _testSaga; - - private string _id; - - protected override void Context() - { - base.Context(); - _id = "something"; - _testSaga = new TestSaga(_id); - } - - protected override void Because() - { - _repository.Save(_testSaga, Guid.NewGuid(), null); - } - - [Fact] - public void should_be_returned_when_loaded_by_id() - { - _repository.GetById(_id).Id.Should().Be(_testSaga.Id); - } - } + { + protected ISagaRepository? _repository; + + protected IStoreEvents? _storeEvents; + + protected override void Context() + { + _storeEvents = Wireup.Init().UsingInMemoryPersistence().Build(); + _repository = new SagaEventStoreRepository(_storeEvents, new SagaFactory()); + } + } + + public class when_a_saga_is_loaded : using_a_sagaeventstorerepository + { + private TestSaga? _testSaga; + + private string _id = "something"; + + protected override void Context() + { + base.Context(); + _testSaga = new TestSaga(_id); + } + + protected override void Because() + { + _repository!.Save(_testSaga!, Guid.NewGuid(), null); + } + + [Fact] + public void should_be_returned_when_loaded_by_id() + { + _repository!.GetById(_id).Id.Should().Be(_testSaga!.Id); + } + } } diff --git a/src/NEventStore.Domain.Tests/Persistence/SagaFactory.cs b/src/NEventStore.Domain.Tests/Persistence/SagaFactory.cs index 52ae20c..3a8a4de 100644 --- a/src/NEventStore.Domain.Tests/Persistence/SagaFactory.cs +++ b/src/NEventStore.Domain.Tests/Persistence/SagaFactory.cs @@ -1,20 +1,24 @@ +using System.Reflection; +using NEventStore.Domain.Persistence; + namespace NEventStore.Domain.Tests.Persistence { - using System; - using System.Reflection; - using NEventStore.Domain.Persistence; - internal class SagaFactory : IConstructSagas - { - public ISaga Build(Type type, string id) - { - ConstructorInfo constructor = type.GetConstructor( - BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance, - null, - new[] { typeof(string) }, - null); + { + public ISaga Build(Type type, string id) + { + var constructor = type.GetConstructor( + BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance, + null, + new[] { typeof(string) }, + null); + + if (constructor == null) + { + throw new NotSupportedException(string.Format("The type '{0}' does not have a constructor accepting a Guid.", type)); + } - return constructor.Invoke(new object[] { id }) as ISaga; - } - } + return (ISaga)constructor.Invoke(new object[] { id }); + } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Core/AggregateBase.cs b/src/NEventStore.Domain/Core/AggregateBase.cs index 10c40ac..31e2c3b 100644 --- a/src/NEventStore.Domain/Core/AggregateBase.cs +++ b/src/NEventStore.Domain/Core/AggregateBase.cs @@ -5,100 +5,104 @@ namespace NEventStore.Domain.Core using System.Collections.Generic; public abstract class AggregateBase : IAggregate, IEquatable - { - private readonly ICollection uncommittedEvents = new LinkedList(); - - private IRouteEvents registeredRoutes; - - protected AggregateBase() - : this(null) - {} - - protected AggregateBase(IRouteEvents handler) - { - if (handler == null) - { - return; - } - - this.RegisteredRoutes = handler; - this.RegisteredRoutes.Register(this); - } - - protected IRouteEvents RegisteredRoutes - { - get - { - return this.registeredRoutes ?? (this.registeredRoutes = new ConventionEventRouter(true, this)); - } - set - { - if (value == null) - { - throw new InvalidOperationException("AggregateBase must have an event router to function"); - } - - this.registeredRoutes = value; - } - } - - public Guid Id { get; protected set; } - - public int Version { get; protected set; } - - void IAggregate.ApplyEvent(object @event) - { - this.RegisteredRoutes.Dispatch(@event); - this.Version++; - } - - ICollection IAggregate.GetUncommittedEvents() - { - return (ICollection)this.uncommittedEvents; - } - - void IAggregate.ClearUncommittedEvents() - { - this.uncommittedEvents.Clear(); - } - - IMemento IAggregate.GetSnapshot() - { - IMemento snapshot = this.GetSnapshot(); - snapshot.Id = this.Id; - snapshot.Version = this.Version; - return snapshot; - } - - public virtual bool Equals(IAggregate other) - { - return null != other && other.Id == this.Id; - } - - protected void Register(Action route) - { - this.RegisteredRoutes.Register(route); - } - - protected void RaiseEvent(object @event) - { - ((IAggregate)this).ApplyEvent(@event); - this.uncommittedEvents.Add(@event); - } - - protected virtual IMemento GetSnapshot() - { - return null; - } - - public override int GetHashCode() - { - return this.Id.GetHashCode(); - } - - public override bool Equals(object obj) - { - return this.Equals(obj as IAggregate); - } - } + { + private readonly ICollection uncommittedEvents = new LinkedList(); + + private IRouteEvents? registeredRoutes; + + protected AggregateBase() + : this(null) + { } + + protected AggregateBase(IRouteEvents? handler) + { + if (handler == null) + { + return; + } + + this.RegisteredRoutes = handler; + this.RegisteredRoutes.Register(this); + } + + protected IRouteEvents RegisteredRoutes + { + get + { + return this.registeredRoutes ?? (this.registeredRoutes = new ConventionEventRouter(true, this)); + } + set + { + if (value == null) + { + throw new InvalidOperationException("AggregateBase must have an event router to function"); + } + + this.registeredRoutes = value; + } + } + + public Guid Id { get; protected set; } + + public int Version { get; protected set; } + + void IAggregate.ApplyEvent(object @event) + { + this.RegisteredRoutes.Dispatch(@event); + this.Version++; + } + + ICollection IAggregate.GetUncommittedEvents() + { + return (ICollection)this.uncommittedEvents; + } + + void IAggregate.ClearUncommittedEvents() + { + this.uncommittedEvents.Clear(); + } + + IMemento? IAggregate.GetSnapshot() + { + var snapshot = this.GetSnapshot(); + if (snapshot == null) + { + return null; + } + snapshot.Id = this.Id; + snapshot.Version = this.Version; + return snapshot; + } + + public virtual bool Equals(IAggregate other) + { + return null != other && other.Id == this.Id; + } + + protected void Register(Action route) + { + this.RegisteredRoutes.Register(route); + } + + protected void RaiseEvent(object @event) + { + ((IAggregate)this).ApplyEvent(@event); + this.uncommittedEvents.Add(@event); + } + + protected virtual IMemento? GetSnapshot() + { + return null; + } + + public override int GetHashCode() + { + return this.Id.GetHashCode(); + } + + public override bool Equals(object obj) + { + return this.Equals(obj as IAggregate); + } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Core/RegistrationEventRouter.cs b/src/NEventStore.Domain/Core/RegistrationEventRouter.cs index be6502c..6b6fc17 100644 --- a/src/NEventStore.Domain/Core/RegistrationEventRouter.cs +++ b/src/NEventStore.Domain/Core/RegistrationEventRouter.cs @@ -1,11 +1,11 @@ namespace NEventStore.Domain.Core { - using System; - using System.Collections.Generic; + using System; + using System.Collections.Generic; - public class RegistrationEventRouter : IRouteEvents + public class RegistrationEventRouter : IRouteEvents { - private readonly IDictionary> handlers = new Dictionary>(); + private readonly Dictionary> handlers = new Dictionary>(); private IAggregate regsitered; diff --git a/src/NEventStore.Domain/IAggregate.cs b/src/NEventStore.Domain/IAggregate.cs index f24c614..14ec44a 100644 --- a/src/NEventStore.Domain/IAggregate.cs +++ b/src/NEventStore.Domain/IAggregate.cs @@ -1,17 +1,16 @@ +using System.Collections; + namespace NEventStore.Domain { - using System; - using System.Collections; - public interface IAggregate - { - Guid Id { get; } - int Version { get; } + { + Guid Id { get; } + int Version { get; } - void ApplyEvent(object @event); - ICollection GetUncommittedEvents(); - void ClearUncommittedEvents(); + void ApplyEvent(object @event); + ICollection GetUncommittedEvents(); + void ClearUncommittedEvents(); - IMemento GetSnapshot(); - } + IMemento? GetSnapshot(); + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/NEventStore.Domain.Core.csproj b/src/NEventStore.Domain/NEventStore.Domain.Core.csproj index c5efc17..a07ed47 100644 --- a/src/NEventStore.Domain/NEventStore.Domain.Core.csproj +++ b/src/NEventStore.Domain/NEventStore.Domain.Core.csproj @@ -1,50 +1,88 @@  - - netstandard2.0;net462 - false - NEventStore.Domain - NEventStore.Domain - - - - NEventStore.Domain - NEventStore Domain - NEventStore Dev Team - http://neventstore.org - false - A recipe for implementing event sourced domain objects with NEventStore. - events, event sourcing, cqrs, storage, persistence, database - - True - - - - TRACE;DEBUG - - - - - - - - - - - - - True - True - ExceptionMessages.resx - - - - - - ResXFileCodeGenerator - ExceptionMessages.Designer.cs - NEventStore.Domain.Persistence.EventStore - - + + netstandard2.0;net462 + false + NEventStore.Domain + NEventStore.Domain + + + + NEventStore.Domain + NEventStore Domain + NEventStore Dev Team + http://neventstore.org + false + A recipe for implementing event sourced domain objects with NEventStore. + events, event sourcing, cqrs, storage, persistence, database + + True + true + true + snupkg + true + true + True + True + NEventStore Dev Team + icon.png + Readme.md + https://github.com/NEventStore/NEventStore.Domain.git + git + license.txt + True + True + latest-recommended + + + + TRACE;DEBUG + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + True + \ + + + True + \ + + + True + \ + + + + + + + + + + + True + True + ExceptionMessages.resx + + + + + + ResXFileCodeGenerator + ExceptionMessages.Designer.cs + NEventStore.Domain.Persistence.EventStore + + \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/ConflictingCommandException.cs b/src/NEventStore.Domain/Persistence/ConflictingCommandException.cs index 31c9e4b..8aa9a95 100644 --- a/src/NEventStore.Domain/Persistence/ConflictingCommandException.cs +++ b/src/NEventStore.Domain/Persistence/ConflictingCommandException.cs @@ -4,43 +4,43 @@ namespace NEventStore.Domain.Persistence using System.Runtime.Serialization; /// - /// Represents a command that could not be executed because it conflicted with the command of another user or actor. - /// - [Serializable] - public class ConflictingCommandException : Exception - { - /// - /// Initializes a new instance of the ConflictingCommandException class. - /// - public ConflictingCommandException() - {} + /// Represents a command that could not be executed because it conflicted with the command of another user or actor. + /// + [Serializable] + public class ConflictingCommandException : Exception + { + /// + /// Initializes a new instance of the ConflictingCommandException class. + /// + public ConflictingCommandException() + {} - /// - /// Initializes a new instance of the ConflictingCommandException class. - /// - /// The message that describes the error. - public ConflictingCommandException(string message) - : base(message) - {} + /// + /// Initializes a new instance of the ConflictingCommandException class. + /// + /// The message that describes the error. + public ConflictingCommandException(string message) + : base(message) + {} - /// - /// Initializes a new instance of the ConflictingCommandException class. - /// - /// The message that describes the error. - /// The message that is the cause of the current exception. - public ConflictingCommandException(string message, Exception innerException) - : base(message, innerException) - {} + /// + /// Initializes a new instance of the ConflictingCommandException class. + /// + /// The message that describes the error. + /// The message that is the cause of the current exception. + public ConflictingCommandException(string message, Exception innerException) + : base(message, innerException) + {} #if !NETSTANDARD1_6 - /// - /// Initializes a new instance of the ConflictingCommandException class. - /// - /// The SerializationInfo that holds the serialized object data of the exception being thrown. - /// The StreamingContext that contains contextual information about the source or destination. - protected ConflictingCommandException(SerializationInfo info, StreamingContext context) - : base(info, context) - {} + /// + /// Initializes a new instance of the ConflictingCommandException class. + /// + /// The SerializationInfo that holds the serialized object data of the exception being thrown. + /// The StreamingContext that contains contextual information about the source or destination. + protected ConflictingCommandException(SerializationInfo info, StreamingContext context) + : base(info, context) + {} #endif - } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.Async.cs b/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.Async.cs new file mode 100644 index 0000000..7cf61b6 --- /dev/null +++ b/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.Async.cs @@ -0,0 +1,84 @@ +using NEventStore.Persistence; + +namespace NEventStore.Domain.Persistence.EventStore +{ + public partial class EventStoreRepository + { + async Task IRepository.GetByIdAsync(string bucketId, Guid id, int version, CancellationToken cancellationToken) + { + var snapshot = await GetSnapshotAsync(bucketId, id, version, cancellationToken).ConfigureAwait(false); + IEventStream stream = await OpenStreamAsync(bucketId, id, version, snapshot, cancellationToken: cancellationToken).ConfigureAwait(false); + var aggregate = GetAggregate(snapshot, stream); + + ApplyEventsToAggregate(version, stream, aggregate); + + return aggregate; + } + + public async Task SaveAsync(string bucketId, IAggregate aggregate, Guid commitId, Action>? updateHeaders, CancellationToken cancellationToken) + { + Dictionary headers = PrepareHeaders(aggregate, updateHeaders); + while (true) + { + IEventStream stream = PrepareStream(bucketId, aggregate, headers); + int commitEventCount = stream.CommittedEvents.Count; + + try + { + await stream.CommitChangesAsync(commitId, cancellationToken).ConfigureAwait(false); + aggregate.ClearUncommittedEvents(); + return; + } + catch (DuplicateCommitException) + { + stream.ClearChanges(); + // Issue: #4 and test: when_an_aggregate_is_persisted_using_the_same_commitId_twice + // should we rethrow the exception here? or provide a feedback whether the save was successful ? + return; + } + catch (ConcurrencyException e) + { + var conflict = ThrowOnConflict(stream, commitEventCount); + stream.ClearChanges(); + + if (conflict) + { + throw new ConflictingCommandException(e.Message, e); + } + } + catch (StorageException e) + { + throw new PersistenceException(e.Message, e); + } + } + } + + private async Task GetSnapshotAsync(string bucketId, Guid id, int version, CancellationToken cancellationToken) + { + var snapshotId = bucketId + id; + if (!_snapshots.TryGetValue(snapshotId, out ISnapshot? snapshot)) + { + snapshot = await _eventStore.Advanced.GetSnapshotAsync(bucketId, id, version, cancellationToken).ConfigureAwait(false); + _snapshots[snapshotId] = snapshot; + } + + return snapshot; + } + + private async Task OpenStreamAsync(string bucketId, Guid id, int version, ISnapshot? snapshot, CancellationToken cancellationToken) + { + IEventStream stream; + var streamId = bucketId + "+" + id; + if (_streams.TryGetValue(streamId, out stream)) + { + return stream; + } + + stream = snapshot == null + ? await _eventStore.OpenStreamAsync(bucketId, id, 0, version, cancellationToken: cancellationToken).ConfigureAwait(false) + : await _eventStore.OpenStreamAsync(snapshot, version, cancellationToken).ConfigureAwait(false); + + return _streams[streamId] = stream; + } + } +} diff --git a/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs b/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs index 7e15a74..fa6e379 100644 --- a/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs +++ b/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs @@ -1,213 +1,184 @@ +using NEventStore.Persistence; + namespace NEventStore.Domain.Persistence.EventStore { - using System; - using System.Collections.Generic; - using System.Linq; - using NEventStore.Persistence; - - public class EventStoreRepository : IRepository - { - private const string AggregateTypeHeader = "AggregateType"; - - private readonly IDetectConflicts _conflictDetector; - - private readonly IStoreEvents _eventStore; - - private readonly IConstructAggregates _factory; - - private readonly IDictionary _snapshots = new Dictionary(); - - private readonly IDictionary _streams = new Dictionary(); - - public EventStoreRepository(IStoreEvents eventStore, IConstructAggregates factory, IDetectConflicts conflictDetector) - { - _eventStore = eventStore; - _factory = factory; - _conflictDetector = conflictDetector; - } - - public void Dispose() - { - Dispose(true); - GC.SuppressFinalize(this); - } - - public virtual TAggregate GetById(Guid id) where TAggregate : class, IAggregate - { - return GetById(Bucket.Default, id); - } - - public virtual TAggregate GetById(Guid id, int versionToLoad) where TAggregate : class, IAggregate - { - return GetById(Bucket.Default, id, versionToLoad); - } - - public TAggregate GetById(string bucketId, Guid id) where TAggregate : class, IAggregate - { - return GetById(bucketId, id, int.MaxValue); - } - - public TAggregate GetById(string bucketId, Guid id, int versionToLoad) where TAggregate : class, IAggregate - { - ISnapshot snapshot = GetSnapshot(bucketId, id, versionToLoad); - IEventStream stream = OpenStream(bucketId, id, versionToLoad, snapshot); - IAggregate aggregate = GetAggregate(snapshot, stream); - - ApplyEventsToAggregate(versionToLoad, stream, aggregate); - - return aggregate as TAggregate; - } - - public virtual void Save(IAggregate aggregate, Guid commitId, Action> updateHeaders) - { - Save(Bucket.Default, aggregate, commitId, updateHeaders); - - } - - public void Save(string bucketId, IAggregate aggregate, Guid commitId, Action> updateHeaders) - { - Dictionary headers = PrepareHeaders(aggregate, updateHeaders); - while (true) - { - IEventStream stream = PrepareStream(bucketId, aggregate, headers); - int commitEventCount = stream.CommittedEvents.Count; - - try - { - stream.CommitChanges(commitId); - aggregate.ClearUncommittedEvents(); - return; - } - catch (DuplicateCommitException) - { - stream.ClearChanges(); + public partial class EventStoreRepository : IRepository + { + private const string AggregateTypeHeader = "AggregateType"; + + private readonly IDetectConflicts _conflictDetector; + + private readonly IStoreEvents _eventStore; + + private readonly IConstructAggregates _factory; + + private readonly Dictionary _snapshots = []; + + private readonly Dictionary _streams = []; + + public EventStoreRepository(IStoreEvents eventStore, IConstructAggregates factory, IDetectConflicts conflictDetector) + { + _eventStore = eventStore; + _factory = factory; + _conflictDetector = conflictDetector; + } + + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + public TAggregate GetById(string bucketId, Guid id, int version) where TAggregate : class, IAggregate + { + var snapshot = GetSnapshot(bucketId, id, version); + IEventStream stream = OpenStream(bucketId, id, version, snapshot); + var aggregate = GetAggregate(snapshot, stream); + + ApplyEventsToAggregate(version, stream, aggregate); + + return aggregate; + } + + public void Save(string bucketId, IAggregate aggregate, Guid commitId, Action>? updateHeaders) + { + Dictionary headers = PrepareHeaders(aggregate, updateHeaders); + while (true) + { + IEventStream stream = PrepareStream(bucketId, aggregate, headers); + int commitEventCount = stream.CommittedEvents.Count; + + try + { + stream.CommitChanges(commitId); + aggregate.ClearUncommittedEvents(); + return; + } + catch (DuplicateCommitException) + { + stream.ClearChanges(); // Issue: #4 and test: when_an_aggregate_is_persisted_using_the_same_commitId_twice - // should we rethtow the exception here? or provide a feedback whether the save was successful ? + // should we rethrow the exception here? or provide a feedback whether the save was successful ? return; - } - catch (ConcurrencyException e) - { + } + catch (ConcurrencyException e) + { var conflict = ThrowOnConflict(stream, commitEventCount); stream.ClearChanges(); if (conflict) - { - throw new ConflictingCommandException(e.Message, e); - } - } - catch (StorageException e) - { - throw new PersistenceException(e.Message, e); - } - } - } - - protected virtual void Dispose(bool disposing) - { - if (!disposing) - { - return; - } - - lock (_streams) - { - foreach (var stream in _streams) - { - stream.Value.Dispose(); - } - - _snapshots.Clear(); - _streams.Clear(); - } - } - - private static void ApplyEventsToAggregate(int versionToLoad, IEventStream stream, IAggregate aggregate) - { - if (versionToLoad == 0 || aggregate.Version < versionToLoad) - { - foreach (var @event in stream.CommittedEvents.Select(x => x.Body)) - { - aggregate.ApplyEvent(@event); - } - } - } - - private IAggregate GetAggregate(ISnapshot snapshot, IEventStream stream) - { - IMemento memento = snapshot == null ? null : snapshot.Payload as IMemento; - return _factory.Build(typeof(TAggregate), stream.StreamId.ToGuid(), memento); - } - - private ISnapshot GetSnapshot(string bucketId, Guid id, int version) - { - ISnapshot snapshot; - var snapshotId = bucketId + id; - if (!_snapshots.TryGetValue(snapshotId, out snapshot)) - { - _snapshots[snapshotId] = snapshot = _eventStore.Advanced.GetSnapshot(bucketId, id, version); - } - - return snapshot; - } - - private IEventStream OpenStream(string bucketId, Guid id, int version, ISnapshot snapshot) - { - IEventStream stream; - var streamId = bucketId + "+" + id; - if (_streams.TryGetValue(streamId, out stream)) - { - return stream; - } - - stream = snapshot == null + { + throw new ConflictingCommandException(e.Message, e); + } + } + catch (StorageException e) + { + throw new PersistenceException(e.Message, e); + } + } + } + + protected virtual void Dispose(bool disposing) + { + if (!disposing) + { + return; + } + + lock (_streams) + { + foreach (var stream in _streams) + { + stream.Value.Dispose(); + } + + _snapshots.Clear(); + _streams.Clear(); + } + } + + private static void ApplyEventsToAggregate(int versionToLoad, IEventStream stream, IAggregate aggregate) + { + if (versionToLoad == 0 || aggregate.Version < versionToLoad) + { + foreach (var @event in stream.CommittedEvents.Select(x => x.Body)) + { + aggregate.ApplyEvent(@event); + } + } + } + + private TAggregate GetAggregate(ISnapshot? snapshot, IEventStream stream) where TAggregate : class, IAggregate + { + var memento = snapshot == null ? null : snapshot.Payload as IMemento; + return (TAggregate)_factory.Build(typeof(TAggregate), stream.StreamId.ToGuid(), memento); + } + + private ISnapshot? GetSnapshot(string bucketId, Guid id, int version) + { + var snapshotId = bucketId + id; + if (!_snapshots.TryGetValue(snapshotId, out ISnapshot? snapshot)) + { + _snapshots[snapshotId] = snapshot = _eventStore.Advanced.GetSnapshot(bucketId, id, version); + } + + return snapshot; + } + + private IEventStream OpenStream(string bucketId, Guid id, int version, ISnapshot? snapshot) + { + IEventStream stream; + var streamId = bucketId + "+" + id; + if (_streams.TryGetValue(streamId, out stream)) + { + return stream; + } + + stream = snapshot == null ? _eventStore.OpenStream(bucketId, id, 0, version) - : _eventStore.OpenStream(snapshot, version); - - return _streams[streamId] = stream; - } - - private IEventStream PrepareStream(string bucketId, IAggregate aggregate, Dictionary headers) - { - IEventStream stream; - var streamId = bucketId + "+" + aggregate.Id; - if (!_streams.TryGetValue(streamId, out stream)) - { - _streams[streamId] = stream = _eventStore.CreateStream(bucketId, aggregate.Id); - } - - foreach (var item in headers) - { - stream.UncommittedHeaders[item.Key] = item.Value; - } - - aggregate.GetUncommittedEvents() - .Cast() - .Select(x => new EventMessage { Body = x }) - .ToList() - .ForEach(stream.Add); - - return stream; - } - - private static Dictionary PrepareHeaders( - IAggregate aggregate, Action> updateHeaders) - { - var headers = new Dictionary(); - - headers[AggregateTypeHeader] = aggregate.GetType().FullName; - if (updateHeaders != null) - { - updateHeaders(headers); - } - - return headers; - } - - private bool ThrowOnConflict(IEventStream stream, int skip) - { - IEnumerable committed = stream.CommittedEvents.Skip(skip).Select(x => x.Body); - IEnumerable uncommitted = stream.UncommittedEvents.Select(x => x.Body); - return _conflictDetector.ConflictsWith(uncommitted, committed); - } - } + : _eventStore.OpenStream(snapshot, version); + + return _streams[streamId] = stream; + } + + private IEventStream PrepareStream(string bucketId, IAggregate aggregate, Dictionary headers) + { + var streamId = bucketId + "+" + aggregate.Id; + if (!_streams.TryGetValue(streamId, out IEventStream stream)) + { + _streams[streamId] = stream = _eventStore.CreateStream(bucketId, aggregate.Id); + } + + foreach (var item in headers) + { + stream.UncommittedHeaders[item.Key] = item.Value; + } + + aggregate.GetUncommittedEvents() + .Cast() + .Select(x => new EventMessage { Body = x }) + .ToList() + .ForEach(stream.Add); + + return stream; + } + + private static Dictionary PrepareHeaders( + IAggregate aggregate, Action>? updateHeaders) + { + var headers = new Dictionary(); + + headers[AggregateTypeHeader] = aggregate.GetType().FullName; + updateHeaders?.Invoke(headers); + + return headers; + } + + private bool ThrowOnConflict(IEventStream stream, int skip) + { + IEnumerable committed = stream.CommittedEvents.Skip(skip).Select(x => x.Body); + IEnumerable uncommitted = stream.UncommittedEvents.Select(x => x.Body); + return _conflictDetector.ConflictsWith(uncommitted, committed); + } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.Async.cs b/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.Async.cs new file mode 100644 index 0000000..016f765 --- /dev/null +++ b/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.Async.cs @@ -0,0 +1,65 @@ +using NEventStore.Persistence; + +namespace NEventStore.Domain.Persistence.EventStore +{ + public partial class SagaEventStoreRepository + { + async Task ISagaRepository.GetByIdAsync(string bucketId, string sagaId, CancellationToken cancellationToken) + { + var stream = await OpenStreamAsync(bucketId, sagaId, cancellationToken).ConfigureAwait(false); + return BuildSaga(sagaId, stream); + } + + public async Task SaveAsync(string bucketId, ISaga saga, Guid commitId, Action>? updateHeaders, CancellationToken cancellationToken) + { + if (saga == null) + { + throw new ArgumentNullException(nameof(saga), ExceptionMessages.NullArgument); + } + + Dictionary headers = PrepareHeaders(saga, updateHeaders); + IEventStream stream = PrepareStream(bucketId, saga, headers); + + await PersistAsync(stream, commitId, cancellationToken).ConfigureAwait(false); + + saga.ClearUncommittedEvents(); + saga.ClearUndispatchedMessages(); + } + + private async Task OpenStreamAsync(string bucketId, string sagaId, CancellationToken cancellationToken) + { + var sagaKey = bucketId + "+" + sagaId; + if (_streams.TryGetValue(sagaKey, out IEventStream stream)) + { + return stream; + } + + try + { + stream = await _eventStore.OpenStreamAsync(bucketId, sagaId, 0, int.MaxValue, cancellationToken).ConfigureAwait(false); + } + catch (StreamNotFoundException) + { + stream = _eventStore.CreateStream(bucketId, sagaId); + } + + return _streams[sagaKey] = stream; + } + + private static async Task PersistAsync(IEventStream stream, Guid commitId, CancellationToken cancellationToken) + { + try + { + await stream.CommitChangesAsync(commitId, cancellationToken); + } + catch (DuplicateCommitException) + { + stream.ClearChanges(); + } + catch (StorageException e) + { + throw new PersistenceException(e.Message, e); + } + } + } +} diff --git a/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs b/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs index ac7c561..e6b2f45 100644 --- a/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs +++ b/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs @@ -1,175 +1,167 @@ +using NEventStore.Persistence; + namespace NEventStore.Domain.Persistence.EventStore { - using System; - using System.Collections.Generic; - using System.Linq; - using NEventStore.Persistence; - - public class SagaEventStoreRepository : ISagaRepository, IDisposable - { - private const string SagaTypeHeader = "SagaType"; - - private string UndispatchedMessageHeader = "UndispatchedMessage."; - - private readonly IStoreEvents _eventStore; - - private readonly IConstructSagas _factory; - - private readonly IDictionary _streams = new Dictionary(); - - public SagaEventStoreRepository(IStoreEvents eventStore, IConstructSagas factory) - { - _eventStore = eventStore; - _factory = factory; - } - - /// - /// Creates and instance of the Saga Repository - /// - /// - /// - /// allows the user to specify the prefix used to store the undispatched commands inside the Commit.Headers dictionary. - /// It is useful for all those databases that do not allow to have . (dot) in property names (like MongoDB, if we chose to not serialize the Headers dictionary as - /// ArrayOfArrays). - public SagaEventStoreRepository(IStoreEvents eventStore, IConstructSagas factory, string undispatchedMessageHeader) - : this(eventStore, factory) - { - UndispatchedMessageHeader = undispatchedMessageHeader; - } - - public void Dispose() - { - Dispose(true); - GC.SuppressFinalize(this); - } - - public TSaga GetById(string bucketId, string sagaId) where TSaga : class, ISaga - { - return BuildSaga(sagaId, OpenStream(bucketId, sagaId)); - } - - public void Save(string bucketId, ISaga saga, Guid commitId, Action> updateHeaders) - { - if (saga == null) - { - throw new ArgumentNullException("saga", ExceptionMessages.NullArgument); - } - - Dictionary headers = PrepareHeaders(saga, updateHeaders); - IEventStream stream = PrepareStream(bucketId, saga, headers); - - Persist(stream, commitId); - - saga.ClearUncommittedEvents(); - saga.ClearUndispatchedMessages(); - } - - protected virtual void Dispose(bool disposing) - { - if (!disposing) - { - return; - } - - lock (_streams) - { - foreach (var stream in _streams) - { - stream.Value.Dispose(); - } - - _streams.Clear(); - } - } - - private IEventStream OpenStream(string bucketId, string sagaId) - { - IEventStream stream; - var sagaKey = bucketId + "+" + sagaId; - if (_streams.TryGetValue(sagaKey, out stream)) - { - return stream; - } - - try - { - stream = _eventStore.OpenStream(bucketId, sagaId, 0, int.MaxValue); - } - catch (StreamNotFoundException) - { - stream = _eventStore.CreateStream(bucketId, sagaId); - } - - return _streams[sagaKey] = stream; - } - - private TSaga BuildSaga(string sagaId, IEventStream stream) where TSaga : class, ISaga - { - var saga = (TSaga)_factory.Build(typeof(TSaga), sagaId); - foreach (var @event in stream.CommittedEvents.Select(x => x.Body)) - { - saga.Transition(@event); - } - - saga.ClearUncommittedEvents(); - saga.ClearUndispatchedMessages(); - - return saga; - } - - private Dictionary PrepareHeaders( - ISaga saga, Action> updateHeaders) - { - var headers = new Dictionary(); - - headers[SagaTypeHeader] = saga.GetType().FullName; - if (updateHeaders != null) - { - updateHeaders(headers); - } - - int i = 0; - foreach (var command in saga.GetUndispatchedMessages()) - { - headers[UndispatchedMessageHeader + i++] = command; - } - - return headers; - } - - private IEventStream PrepareStream(string bucketId, ISaga saga, Dictionary headers) - { - IEventStream stream; - var sagaKey = bucketId + "+" + saga.Id; - if (!_streams.TryGetValue(sagaKey, out stream)) - { - _streams[sagaKey] = stream = _eventStore.CreateStream(bucketId, saga.Id); - } - - foreach (var item in headers) - { - stream.UncommittedHeaders[item.Key] = item.Value; - } - - saga.GetUncommittedEvents().Cast().Select(x => new EventMessage { Body = x }).ToList().ForEach(stream.Add); - - return stream; - } - - private static void Persist(IEventStream stream, Guid commitId) - { - try - { - stream.CommitChanges(commitId); - } - catch (DuplicateCommitException) - { - stream.ClearChanges(); - } - catch (StorageException e) - { - throw new PersistenceException(e.Message, e); - } - } - } + public partial class SagaEventStoreRepository : ISagaRepository, IDisposable + { + private const string SagaTypeHeader = "SagaType"; + + private string UndispatchedMessageHeader = "UndispatchedMessage."; + + private readonly IStoreEvents _eventStore; + + private readonly IConstructSagas _factory; + + private readonly Dictionary _streams = []; + + public SagaEventStoreRepository(IStoreEvents eventStore, IConstructSagas factory) + { + _eventStore = eventStore; + _factory = factory; + } + + /// + /// Creates and instance of the Saga Repository + /// + /// + /// + /// allows the user to specify the prefix used to store the un-dispatched commands inside the Commit.Headers dictionary. + /// It is useful for all those databases that do not allow to have . (dot) in property names (like MongoDB, if we chose to not serialize the Headers dictionary as + /// ArrayOfArrays). + public SagaEventStoreRepository(IStoreEvents eventStore, IConstructSagas factory, string undispatchedMessageHeader) + : this(eventStore, factory) + { + UndispatchedMessageHeader = undispatchedMessageHeader; + } + + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + public TSaga GetById(string bucketId, string sagaId) where TSaga : class, ISaga + { + return BuildSaga(sagaId, OpenStream(bucketId, sagaId)); + } + + public void Save(string bucketId, ISaga saga, Guid commitId, Action>? updateHeaders) + { + if (saga == null) + { + throw new ArgumentNullException(nameof(saga), ExceptionMessages.NullArgument); + } + + Dictionary headers = PrepareHeaders(saga, updateHeaders); + IEventStream stream = PrepareStream(bucketId, saga, headers); + + Persist(stream, commitId); + + saga.ClearUncommittedEvents(); + saga.ClearUndispatchedMessages(); + } + + protected virtual void Dispose(bool disposing) + { + if (!disposing) + { + return; + } + + lock (_streams) + { + foreach (var stream in _streams) + { + stream.Value.Dispose(); + } + + _streams.Clear(); + } + } + + private IEventStream OpenStream(string bucketId, string sagaId) + { + var sagaKey = bucketId + "+" + sagaId; + if (_streams.TryGetValue(sagaKey, out IEventStream stream)) + { + return stream; + } + + try + { + stream = _eventStore.OpenStream(bucketId, sagaId, 0, int.MaxValue); + } + catch (StreamNotFoundException) + { + stream = _eventStore.CreateStream(bucketId, sagaId); + } + + return _streams[sagaKey] = stream; + } + + private TSaga BuildSaga(string sagaId, IEventStream stream) where TSaga : class, ISaga + { + var saga = (TSaga)_factory.Build(typeof(TSaga), sagaId); + foreach (var @event in stream.CommittedEvents.Select(x => x.Body)) + { + saga.Transition(@event); + } + + saga.ClearUncommittedEvents(); + saga.ClearUndispatchedMessages(); + + return saga; + } + + private Dictionary PrepareHeaders( + ISaga saga, Action>? updateHeaders) + { + var headers = new Dictionary(); + + headers[SagaTypeHeader] = saga.GetType().FullName; + updateHeaders?.Invoke(headers); + + int i = 0; + foreach (var command in saga.GetUndispatchedMessages()) + { + headers[UndispatchedMessageHeader + i++] = command; + } + + return headers; + } + + private IEventStream PrepareStream(string bucketId, ISaga saga, Dictionary headers) + { + var sagaKey = bucketId + "+" + saga.Id; + if (!_streams.TryGetValue(sagaKey, out IEventStream stream)) + { + _streams[sagaKey] = stream = _eventStore.CreateStream(bucketId, saga.Id); + } + + foreach (var item in headers) + { + stream.UncommittedHeaders[item.Key] = item.Value; + } + + saga.GetUncommittedEvents().Cast().Select(x => new EventMessage { Body = x }).ToList().ForEach(stream.Add); + + return stream; + } + + private static void Persist(IEventStream stream, Guid commitId) + { + try + { + stream.CommitChanges(commitId); + } + catch (DuplicateCommitException) + { + stream.ClearChanges(); + } + catch (StorageException e) + { + throw new PersistenceException(e.Message, e); + } + } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/IConstructAggregates.cs b/src/NEventStore.Domain/Persistence/IConstructAggregates.cs index 73bd94b..c318520 100644 --- a/src/NEventStore.Domain/Persistence/IConstructAggregates.cs +++ b/src/NEventStore.Domain/Persistence/IConstructAggregates.cs @@ -1,9 +1,7 @@ namespace NEventStore.Domain.Persistence { - using System; - public interface IConstructAggregates - { - IAggregate Build(Type type, Guid id, IMemento snapshot); - } + { + IAggregate Build(Type type, Guid id, IMemento? snapshot); + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/IConstructSagas.cs b/src/NEventStore.Domain/Persistence/IConstructSagas.cs index 62e6777..c2e8f1c 100644 --- a/src/NEventStore.Domain/Persistence/IConstructSagas.cs +++ b/src/NEventStore.Domain/Persistence/IConstructSagas.cs @@ -1,9 +1,7 @@ namespace NEventStore.Domain.Persistence { - using System; - public interface IConstructSagas - { - ISaga Build(Type type, string id); - } + { + ISaga Build(Type type, string id); + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/IRepository.cs b/src/NEventStore.Domain/Persistence/IRepository.cs index af7cb14..56be2a1 100644 --- a/src/NEventStore.Domain/Persistence/IRepository.cs +++ b/src/NEventStore.Domain/Persistence/IRepository.cs @@ -1,20 +1,13 @@ namespace NEventStore.Domain.Persistence { - using System; - using System.Collections.Generic; - public interface IRepository : IDisposable - { - TAggregate GetById(Guid id) where TAggregate : class, IAggregate; - - TAggregate GetById(Guid id, int version) where TAggregate : class, IAggregate; - - TAggregate GetById(string bucketId, Guid id) where TAggregate : class, IAggregate; + { + TAggregate GetById(string bucketId, Guid id, int version) where TAggregate : class, IAggregate; - TAggregate GetById(string bucketId, Guid id, int version) where TAggregate : class, IAggregate; + Task GetByIdAsync(string bucketId, Guid id, int version, CancellationToken cancellationToken = default) where TAggregate : class, IAggregate; - void Save(IAggregate aggregate, Guid commitId, Action> updateHeaders); + void Save(string bucketId, IAggregate aggregate, Guid commitId, Action>? updateHeaders); - void Save(string bucketId, IAggregate aggregate, Guid commitId, Action> updateHeaders); - } + Task SaveAsync(string bucketId, IAggregate aggregate, Guid commitId, Action>? updateHeaders, CancellationToken cancellationToken = default); + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/ISagaRepository.cs b/src/NEventStore.Domain/Persistence/ISagaRepository.cs index d0da6c6..a727103 100644 --- a/src/NEventStore.Domain/Persistence/ISagaRepository.cs +++ b/src/NEventStore.Domain/Persistence/ISagaRepository.cs @@ -1,12 +1,13 @@ namespace NEventStore.Domain.Persistence { - using System; - using System.Collections.Generic; - public interface ISagaRepository - { - TSaga GetById(string bucketId, string sagaId) where TSaga : class, ISaga; + { + TSaga GetById(string bucketId, string sagaId) where TSaga : class, ISaga; + + Task GetByIdAsync(string bucketId, string sagaId, CancellationToken cancellationToken = default) where TSaga : class, ISaga; + + void Save(string bucketId, ISaga saga, Guid commitId, Action>? updateHeaders); - void Save(string bucketId, ISaga saga, Guid commitId, Action> updateHeaders); - } + Task SaveAsync(string bucketId, ISaga saga, Guid commitId, Action>? updateHeaders, CancellationToken cancellationToken = default); + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/PersistenceException.cs b/src/NEventStore.Domain/Persistence/PersistenceException.cs index 967ccbc..96c68d5 100644 --- a/src/NEventStore.Domain/Persistence/PersistenceException.cs +++ b/src/NEventStore.Domain/Persistence/PersistenceException.cs @@ -4,43 +4,43 @@ namespace NEventStore.Domain.Persistence using System.Runtime.Serialization; /// - /// Represents a general failure of the persistence infrastructure. - /// - [Serializable] - public class PersistenceException : Exception - { - /// - /// Initializes a new instance of the PersistenceException class. - /// - public PersistenceException() - {} + /// Represents a general failure of the persistence infrastructure. + /// + [Serializable] + public class PersistenceException : Exception + { + /// + /// Initializes a new instance of the PersistenceException class. + /// + public PersistenceException() + { } - /// - /// Initializes a new instance of the PersistenceException class. - /// - /// The message that describes the error. - public PersistenceException(string message) - : base(message) - {} + /// + /// Initializes a new instance of the PersistenceException class. + /// + /// The message that describes the error. + public PersistenceException(string message) + : base(message) + { } - /// - /// Initializes a new instance of the PersistenceException class. - /// - /// The message that describes the error. - /// The message that is the cause of the current exception. - public PersistenceException(string message, Exception innerException) - : base(message, innerException) - {} + /// + /// Initializes a new instance of the PersistenceException class. + /// + /// The message that describes the error. + /// The message that is the cause of the current exception. + public PersistenceException(string message, Exception innerException) + : base(message, innerException) + { } #if !NETSTANDARD1_6 - /// - /// Initializes a new instance of the PersistenceException class. - /// - /// The SerializationInfo that holds the serialized object data of the exception being thrown. - /// The StreamingContext that contains contextual information about the source or destination. - protected PersistenceException(SerializationInfo info, StreamingContext context) - : base(info, context) - {} + /// + /// Initializes a new instance of the PersistenceException class. + /// + /// The SerializationInfo that holds the serialized object data of the exception being thrown. + /// The StreamingContext that contains contextual information about the source or destination. + protected PersistenceException(SerializationInfo info, StreamingContext context) + : base(info, context) + { } #endif - } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs b/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs index b854a4f..0c832f8 100644 --- a/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs +++ b/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs @@ -1,17 +1,65 @@ namespace NEventStore.Domain.Persistence { - using System; - public static class RepositoryExtensions - { - public static void Save(this IRepository repository, IAggregate aggregate, Guid commitId) - { - repository.Save(aggregate, commitId, a => { }); - } - - public static void Save(this IRepository repository, string bucketId, IAggregate aggregate, Guid commitId) - { - repository.Save(bucketId, aggregate, commitId, a => { }); - } - } + { + public static TAggregate GetById(this IRepository repository, Guid id) where TAggregate : class, IAggregate + { + return repository.GetById(Bucket.Default, id, int.MaxValue); + } + + public static TAggregate GetById(this IRepository repository, Guid id, int version) where TAggregate : class, IAggregate + { + return repository.GetById(Bucket.Default, id, version); + } + + public static TAggregate GetById(this IRepository repository, string bucketId, Guid id) where TAggregate : class, IAggregate + { + return repository.GetById(bucketId, id, int.MaxValue); + } + + public static void Save(this IRepository repository, IAggregate aggregate, Guid commitId, Action>? updateHeaders) + { + repository.Save(Bucket.Default, aggregate, commitId, updateHeaders); + } + + public static void Save(this IRepository repository, IAggregate aggregate, Guid commitId) + { + repository.Save(aggregate, commitId, null); + } + + public static void Save(this IRepository repository, string bucketId, IAggregate aggregate, Guid commitId) + { + repository.Save(bucketId, aggregate, commitId, null); + } + + public static Task GetByIdAsync(this IRepository repository, Guid id, CancellationToken cancellationToken = default) where TAggregate : class, IAggregate + { + return repository.GetByIdAsync(Bucket.Default, id, int.MaxValue, cancellationToken); + } + + public static Task GetByIdAsync(this IRepository repository, Guid id, int version, CancellationToken cancellationToken = default) where TAggregate : class, IAggregate + { + return repository.GetByIdAsync(Bucket.Default, id, version, cancellationToken); + } + + public static Task GetByIdAsync(this IRepository repository, string bucketId, Guid id, CancellationToken cancellationToken = default) where TAggregate : class, IAggregate + { + return repository.GetByIdAsync(bucketId, id, int.MaxValue, cancellationToken); + } + + public static Task SaveAsync(this IRepository repository, IAggregate aggregate, Guid commitId, Action>? updateHeaders, CancellationToken cancellationToken = default) + { + return repository.SaveAsync(Bucket.Default, aggregate, commitId, updateHeaders, cancellationToken); + } + + public static Task SaveAsync(this IRepository repository, IAggregate aggregate, Guid commitId, CancellationToken cancellationToken = default) + { + return repository.SaveAsync(aggregate, commitId, null, cancellationToken); + } + + public static Task SaveAsync(this IRepository repository, string bucketId, IAggregate aggregate, Guid commitId, CancellationToken cancellationToken = default) + { + return repository.SaveAsync(bucketId, aggregate, commitId, null, cancellationToken); + } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs b/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs index 38247b5..7065965 100644 --- a/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs +++ b/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs @@ -1,10 +1,13 @@ namespace NEventStore.Domain.Persistence { - using System; - using System.Collections.Generic; - public static class SagaRepositoryExtensions { + public static TSaga GetById(this ISagaRepository sagaRepository, string sagaId) + where TSaga : class, ISaga + { + return sagaRepository.GetById(Bucket.Default, sagaId); + } + public static TSaga GetById(this ISagaRepository sagaRepository, Guid sagaId) where TSaga : class, ISaga { @@ -15,15 +18,31 @@ public static void Save( this ISagaRepository sagaRepository, ISaga saga, Guid commitId, - Action> updateHeaders) + Action>? updateHeaders) { sagaRepository.Save(Bucket.Default, saga, commitId, updateHeaders); } - public static TSaga GetById(this ISagaRepository sagaRepository, string sagaId) + public static Task GetByIdAsync(this ISagaRepository sagaRepository, string sagaId, CancellationToken cancellationToken = default) where TSaga : class, ISaga { - return sagaRepository.GetById(Bucket.Default, sagaId); + return sagaRepository.GetByIdAsync(Bucket.Default, sagaId, cancellationToken); + } + + public static Task GetByIdAsync(this ISagaRepository sagaRepository, Guid sagaId, CancellationToken cancellationToken = default) + where TSaga : class, ISaga + { + return sagaRepository.GetByIdAsync(Bucket.Default, sagaId.ToString(), cancellationToken); + } + + public static Task SaveAsync( + this ISagaRepository sagaRepository, + ISaga saga, + Guid commitId, + Action>? updateHeaders, + CancellationToken cancellationToken = default) + { + return sagaRepository.SaveAsync(Bucket.Default, saga, commitId, updateHeaders, cancellationToken); } } } \ No newline at end of file