From cd9efdef6167d31829df30410e7ac9597dc2c4cf Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 17 Jan 2025 12:36:47 +0100 Subject: [PATCH 1/7] Improved nuget package, Update dependencies and project configurations - Bump `gitversion.tool` to version `6.1.0`. - Add entry for `NEventStore` version `10.0.0` in `Changelog.md`. - Modify `build.ps1` to utilize the updated `gitversion` tool. - Change subproject commit hash in the `NEventStore` directory. - Add new binary file `icon.png`. - Update `.editorconfig` with new formatting rules and diagnostics. - Change `NEventStore` dependency range in `NEventStore.Domain.nuspec`. - Set language version to `13.0` and enable nullable types in `Directory.Build.props`. - Update target frameworks and package references in `NEventStore.Domain.Core.Tests.csproj`. - Enhance `NEventStore.Domain.Core.csproj` with additional package generation properties. --- .config/dotnet-tools.json | 5 +- Changelog.md | 5 + readme.markdown => Readme.md | 0 build.ps1 | 4 +- dependencies/NEventStore | 2 +- icon.png | Bin 0 -> 5872 bytes src/.editorconfig | 30 +++++ src/.nuget/NEventStore.Domain.nuspec | 73 ++++++----- src/Directory.Build.props | 7 + src/NEventStore.Domain.Core.sln | 2 +- .../NEventStore.Domain.Core.Tests.csproj | 14 +- .../NEventStore.Domain.Core.csproj | 123 +++++++++++------- 12 files changed, 174 insertions(+), 91 deletions(-) rename readme.markdown => Readme.md (100%) create mode 100644 icon.png create mode 100644 src/.editorconfig create mode 100644 src/Directory.Build.props 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..1d34bdd 100644 --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,10 @@ # NEventStore.Domain Versions +## vNext + +- Async Methods [#17](https://github.com/NEventStore/NEventStore.Domain/issues/17) +- Updated NEventStore reference to version 10.0.0 + ## 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/build.ps1 b/build.ps1 index fa51aeb..7b2361a 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" @@ -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..b92f91a 160000 --- a/dependencies/NEventStore +++ b/dependencies/NEventStore @@ -1 +1 @@ -Subproject commit a7d4848ac5ec6b5d6f1e7bbc19a2381ca5d0e721 +Subproject commit b92f91ac90c743d048b8beef5f41486681b0f191 diff --git a/icon.png b/icon.png new file mode 100644 index 0000000000000000000000000000000000000000..c6c977dab588a6cf30126896d32b0a36f5a1f38a GIT binary patch literal 5872 zcmai2cTkgGmkyx?=^g1MbfhUwdMDBmq>1$2BndT?h)9PBh=52DRHAf{jvyidq=q7f zUZi)B-ZsCT{bqM|zS;fbzIX0>X6}2=bDs0O=iUSpBVEc{thWFF03}3E%M3pw{}~_> z{IgTVu@65Hz|3?t02PDm>-YtcySkw|08ou0zjP$VugRY3S;GJTs?L7~LAO8J1pr`r z0nt)74|%ek7Yb*vXzNxTpIJAG+hRiipQ-6_gTT@>ZZtTm!WyCEMnxgcF4_RKP_!{z zhfAwF#Z@Q54O56jAqk~_rm3qlX{V`i>w#!$s`$b1^uZC^nG;>XJhoX;qUls!HK^X`_6cPcg z?Am}QssCl>Yr3%gb|_l^RH$UQ?)*)m*{9Q`E!%8KCuFf-&%3f|kY0r-T;7rAy7ljGhV#FmV7oD1KXQkIj8Zgh*q5-V32lzV zm#)bl+(Xxn7y7??zbaPoo9xp%O3+NN@Qv$}6~)vXCN~SmPB0sgavz59M1^=#zRN=t zV$Yi^M&5yV(88j;fgKb&;Nu{>>3??qnNeDGu2V>z44Ag5vjgEUA1;fPLTsgByFN7E zke(8w{(864;l5vU2Aqn@gyPZ?utvb_L$mvKVbofi%=66;;&@$hYNEx=3i5Ye115iY zPaHd4bdA(p+CCXE1!3z4>J0(6+FAb%O#csx|IUm8eap9;G@0(d`{Mjjhbj;_=q>lm?%-M*h{c{Xj}RSG-FAv zB9;bNj(XD}#iOS;mR-g!dXKv=#b>(b785Y~DuQMBE=%s#uLT)(e#jrYviHADmR?bM z*Jcad`N{JES2j=dCGuH{O@PiA2P~v8HD-;5aVgr~L>(1XsFqEN9tI2E-bPKyi)BwcBTNl5^SIrT7~mTe{Ny|TH9H`C1s#`a(@`~0%DfP}FkwcfBDJ|al9 zqXEk#Wy)kEuY|=I^b~&kHsTO+GbH@TUwQD{E;KyiRH?~b6pgOuCPG`#6_W-o{OYhX zt&;?PVMH|{C%7O%GW@$5{QB3SFTFCh6Zh4=l3Shq@E`Ubf7@oVPQXlqNjjPSgGQWX zXu7PjvEMYXlKIGD!LOfA5XX>igdBCegHvG%O{~ni3MM!(4)>hvaD?u+O~&o6STnUI zBvqd@DyHpmUpTu7B&BHIT=1Roj~nWwZ0-Qx6eB_e{Wj<)Hyrzp3Mb=W-b}g4xXD>zb76Kr_KVXrm1Sb~WYc59vQ*%Qy?%qF>N%Wpeu1poC(={Q zf9Pw@+z;*H7JUF~=Bhivx(9)YoN#TJH{XomYOrHzJKP$DmU&|kpM7H|B+wJjLWs# zV!_ZM;zsU%pIdiOPJ*>;2Iae*#TJXrJ$*mEJTO!v+-$zQe|8c!7N{!kS|C-me4+8= z`i~^4^=np^%HW^ko40@B-qLl4CxX>ADrq>`r3;bdOPg0VirQswe^Icp^QW2Itd1%v6yBK*NYzY9*k53_2^Jniny&^Ci9(CDZXe zK(j@@vt8Wt(8yzlkE!ucZ*hBm$SaFCy-F4v=*>}?Z_dmEGK>JVlwO^G;I({2sU~Po zi84t3kTJN*9LzjVW={9&)x_dzsd}ZwC|@hRieIxbB>FQeLQHLDZC-fj&uo4OAoD>{ zVdG?)G|9_!nqICF%)(Nb?XY;{ZD=|;p-A6?#d{0FAJIZA$t%uoyA;!1%H0>evk;kj zgUWB(N-ID&aRLjCuCRhX65=k$& zkNaz_%8pA_q^*XE6iKE1^yH$i>h#jDM6$Fs35GYvk~zn6R^2FzV0O~I``@9dD{K|n zj>}Y>#$=^3wxKJ}xfCoX_YU3ldukl4d|32%jNq)-${BP_gGnb8L1P@su-lrMS8>9@ zFOfD$nSVJ-g#5p>wFo=1*8PZZc4~CidNtbmU0ah<_0Qv?!rwiv#kUFIBc5yHxPntO zw^IwL0#F72iyQK$hONP^Gt>^4bLfI(@Zz=l9cq;Cd;08sF$=z^Ud8A&^!3E(rJuhj zo~ZXH8T6umzboJMQTgnZ+ym)t_OjLbT*SetZ-9~R!AS z-%Led%FVs3nUW3ST|5kg09hi2m)`aSEB6Z$a>dm^>Xn1n25hv;do9}?F4fhJ#wv_~ zO_X0ci0|pVCicqIy2)3cmPV5+^WI-5em#ud7M$~`cC7Ek^ROmz)l;0+0K`)v^rOBR z8AM)~f8K60+f@qtAUjQ~<~)7Bk;%g^YLX9Nx$%7ToYc82kU9P8QFT+f+=p_YN87}t z+SOAWrxG?+)G+o?_EjoIfZTv7hqX54vyG^Lu1Hw7|BN%xq*3bs=%oK$5BNVC?SGjY z?*IsX_t#~6i&O?k$zFg{0R1u;w=KoS?9W^N`Ce{OVk>Vx9&X=jK?`GcMct!L7RNjqMZLRqrnM_H4j!w{N_uX&$d~~rm27@ocB|3(1@bP z#22GZg5-&3-7CA*I3IWtXr-W1NzDX#_3+-#Vm;ua%9WhBto#}#tJTxIue}MJMfKASYB)c(`=XF z6{7KtU79ReVN-V}-v043l!EzyWmqwKtfYi~)}v_4OK9n|5|A>;OeSe=-g33GESmxl z2;{!s?x1v{Oi3J^1|~}xq|P1J@Kl%W$b(-_yiK_DvnnfHXAH0fJS@0YbX^{&oO@x~ z_)+he`Otx$uEWl$F?@$XYDPoPPBPDrR^2i@#Q@fL2l+M5$kEJ?u#%0APLN#x6dySa zF#0m$?CE+097Pc%6@D~?6w|K zwY43e{I)HBlpnfsJ3ssYa(=YATGe1vu5>aDb8&TDSe@7#C*PP|&rM5v_6Tug+!uUr zbuiDlJC$QP>gK8<35`ZO47`!qn*Eyl<_#;5kg#d_wQaUCqJC70jHjo!cj0%nbAPr} z=cg2VvG0aZ7);y6<$Zj7JWW$mQ)*Z(PE5ccn_K$s-Mp;vk_$MI|5$G$t zpQa`z+%DEZHSJG)rW;KS@|AFPHdf`2_7{^R5*2qEHC^1@lcJ-6``Zn3iJqe$6Ykj7 z%j0asn5unn()&GpDEaZ5bW;gC{EnqU6>>QFOM7AU8m}O@w?r&x3_n(cH5q; zOXgC|7q@Gc-*UBfyE;Gm(bXjtu&9NPhJ#Bxu`xS%E+T7f%_BNG8X*649dNPY(R}7*k#_B<#(yS?07l14Y`;`~WPaLpC+V9`Lr(s~wg2V3x1O$}X z)FtNJb-zEU5VFU;+kT@`VqAc6s(zWBz2>3&+STZ-d`BEtWY4d1TLr|i*ww{JcXtWT z3tLb}0=?kx=0H(WD)#v#i8hDfLiM#H%3e-uYwOmV>kFT`w!DoEFAYu2+~l&I;T0AK ze46kX-`^;$tQY0~B*mH%JCG|^i#YKE^t;{lZ@-c)h`Men=G%nm>+|vPJ?GJk-CNGG zmyWy)G0In3cX?=S-HX`$+P#IV6&4rQshl|c?TnJCCa0w27ZFMP_Dzk1f(2<)9waup zd8DPIQ{&Wi>)_yk{U{+J;eAPo;x#Gg)-7IM-ss_Blk>mpOW>OReuIw&IU1&>w3=~L zuFF3&%dD!udqyBzTa}pZ*!(b7y&8L;QdGpt8F}gm2s_ysSVE}|)()v8buC_JnXorf zky#ec68rqFQWtyNXaW`xNUW);DQ!lc9j+$%PSq=978Ml<)F&}ZFc~W!y{f|b6g4(l z{lH)nC!Qb-t;Cq(X7-o5wZp?(dxnOXH9)2o7W_M3PoC`mvh|noGK|Ob6I-lH9dL7X zl7y!|PhSm-a@d#CwPMuUtr(g`yd(4vKNb zjaa?!lt3&Mhsa=o>YRr!VouDeN=?kVM(ekaEKdy0#mO!+pKdC{hYuf&>hU?H1vubp zT&H0}%3}6y(}AY+f5Vg=9UYCdOGXuNr>Ccq?t|?3+7lKQ=22BGn4fppX+1P5p`D)z zp5w>Az?T%>>;5%tF|NU%YpTIN*o#`N{Lx*{5itBdM>-f=>uyXD8%u5Zi?8p_>LfLr zY%(6#YVQegvxg7Qj;H){gT_L8XuMN3kxT>_Pwr-xCn9^cr*UvhMG&|?aP-mv*Cw^zsjsT-^h4dH9`4}mG)yY zW-^~{&%fRf5f!zrav(q)Zv}jwSck8{?*j?*+9^gx0x%d%4IbETkm;jGu~?5kZ5KN* z8DA#6f;l-ku_y+Kt*@`AE*Tjb@`Ha|8+OzW`w^hWy_m!w_sn-Z7rx_r+Ey;=R-5zl z=K~QD5%v$Qt?aqw3!_huesp)&!d8v&ELfjzs@-(r4Bu1#-q{H;HZJ+{5AT7rBGzV_RCnWITv*!ro2LgfFcipMjWPRo@ zcH6rhB&))XymU~UGiCbTcl_o^9$8w#@b!a9Vbk56J@y~8j)V5|?vJ7)&)6&uwgf=HC1@Y+j^z`W1R&n_R1<7j0QY_#Z=`Rh$!}3o|D+9TB zAzQ+uV~{QREc9?ip^LGuu5N3ty+FztLqkvh91OwBA-2>~f`p2FZOlzi52P5p1I8;~ zz!F9wyzapVr!llC-kV<)c72m|-e)Vh{_!Z3dv7|wQNu@(%*6P3kgWe~zua7SXFM(6 z-bsU?$MN=5R(3Ya$jHd)=@JU-Raf@7DIE#8$mlwH+zd~^Yms!&mQ_g$EhHVkRrk;k zidy9qnVd|4PnuP0h!hza*+A-(@{a8swHNlx+d%}k - - 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..9d9a0f3 100644 --- a/src/NEventStore.Domain.Core.sln +++ b/src/NEventStore.Domain.Core.sln @@ -20,7 +20,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution ..\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/NEventStore.Domain.Core.csproj b/src/NEventStore.Domain/NEventStore.Domain.Core.csproj index c5efc17..7b65de2 100644 --- a/src/NEventStore.Domain/NEventStore.Domain.Core.csproj +++ b/src/NEventStore.Domain/NEventStore.Domain.Core.csproj @@ -1,50 +1,81 @@  - - 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 + + + + + + + + True + \ + + + True + \ + + + True + \ + + + + + + + + + + + True + True + ExceptionMessages.resx + + + + + + ResXFileCodeGenerator + ExceptionMessages.Designer.cs + NEventStore.Domain.Persistence.EventStore + + \ No newline at end of file From 12ebf9bb3cab300da0bda5c0f0e514f32c083a42 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 17 Jan 2025 13:21:38 +0100 Subject: [PATCH 2/7] Refactor IRepository and related components for null safety - Updated Changelog.md to reflect breaking changes in IRepository. - Adjusted .editorconfig for diagnostic severity settings. - Added .editorconfig to solution items in NEventStore.Domain.Core.sln. - Modified AggregateFactory.Build method to accept optional IMemento? parameter. - Refactored AggregateBase to use nullable types for improved safety. - Updated IAggregate, IConstructAggregates, and IConstructSagas interfaces for nullable types. - Improved XML comments in ConflictingCommandException and PersistenceException classes. - Refactored EventStoreRepository and SagaEventStoreRepository for better handling of nulls and events. - Enhanced RepositoryExtensions with new methods for aggregate retrieval and saving. - Introduced SagaRepositoryExtensions for additional saga repository methods. --- Changelog.md | 4 + src/.editorconfig | 5 +- src/NEventStore.Domain.Core.sln | 1 + .../Persistence/AggregateFactory.cs | 30 +- src/NEventStore.Domain/Core/AggregateBase.cs | 196 ++++----- src/NEventStore.Domain/IAggregate.cs | 21 +- .../ConflictingCommandException.cs | 68 ++-- .../EventStore/EventStoreRepository.cs | 380 ++++++++---------- .../EventStore/SagaEventStoreRepository.cs | 13 +- .../Persistence/IConstructAggregates.cs | 8 +- .../Persistence/IConstructSagas.cs | 8 +- .../Persistence/IRepository.cs | 19 +- .../Persistence/ISagaRepository.cs | 11 +- .../Persistence/PersistenceException.cs | 68 ++-- .../Persistence/RepositoryExtensions.cs | 44 +- .../Persistence/SagaRepositoryExtensions.cs | 3 - 16 files changed, 433 insertions(+), 446 deletions(-) diff --git a/Changelog.md b/Changelog.md index 1d34bdd..f8dfe83 100644 --- a/Changelog.md +++ b/Changelog.md @@ -5,6 +5,10 @@ - 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/src/.editorconfig b/src/.editorconfig index 6f92936..73679fd 100644 --- a/src/.editorconfig +++ b/src/.editorconfig @@ -27,4 +27,7 @@ dotnet_diagnostic.IDE0305.severity = none dotnet_diagnostic.CA1848.severity = none # CA2254: Template should be a static expression -dotnet_diagnostic.CA2254.severity = none \ No newline at end of file +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/NEventStore.Domain.Core.sln b/src/NEventStore.Domain.Core.sln index 9d9a0f3..8006b97 100644 --- a/src/NEventStore.Domain.Core.sln +++ b/src/NEventStore.Domain.Core.sln @@ -15,6 +15,7 @@ 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 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/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/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/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.cs b/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs index 7e15a74..86a05fd 100644 --- a/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs +++ b/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs @@ -1,213 +1,189 @@ +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(); + { + 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) + { + 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 ? _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) + { + 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); + } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs b/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs index ac7c561..23967f2 100644 --- a/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs +++ b/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs @@ -1,10 +1,7 @@ +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"; @@ -15,7 +12,7 @@ public class SagaEventStoreRepository : ISagaRepository, IDisposable private readonly IConstructSagas _factory; - private readonly IDictionary _streams = new Dictionary(); + private readonly Dictionary _streams = []; public SagaEventStoreRepository(IStoreEvents eventStore, IConstructSagas factory) { @@ -28,7 +25,7 @@ public SagaEventStoreRepository(IStoreEvents eventStore, IConstructSagas factory /// /// /// - /// allows the user to specify the prefix used to store the undispatched commands inside the Commit.Headers dictionary. + /// 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) @@ -52,7 +49,7 @@ public void Save(string bucketId, ISaga saga, Guid commitId, Action headers = PrepareHeaders(saga, updateHeaders); 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..273d5d7 100644 --- a/src/NEventStore.Domain/Persistence/IRepository.cs +++ b/src/NEventStore.Domain/Persistence/IRepository.cs @@ -1,20 +1,9 @@ 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; - - void Save(IAggregate aggregate, Guid commitId, Action> updateHeaders); + { + TAggregate GetById(string bucketId, Guid id, int version) where TAggregate : class, IAggregate; - void Save(string bucketId, IAggregate aggregate, Guid commitId, Action> updateHeaders); - } + void Save(string bucketId, IAggregate aggregate, Guid commitId, Action> updateHeaders); + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/ISagaRepository.cs b/src/NEventStore.Domain/Persistence/ISagaRepository.cs index d0da6c6..2707a34 100644 --- a/src/NEventStore.Domain/Persistence/ISagaRepository.cs +++ b/src/NEventStore.Domain/Persistence/ISagaRepository.cs @@ -1,12 +1,9 @@ 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; - void Save(string bucketId, ISaga saga, Guid commitId, Action> updateHeaders); - } + void Save(string bucketId, ISaga saga, Guid commitId, Action> updateHeaders); + } } \ 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..a7de22e 100644 --- a/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs +++ b/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs @@ -1,17 +1,37 @@ 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 => { }); - } + { + static readonly Action> DoNotUpdateHeaders = _ => { }; + + 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, DoNotUpdateHeaders); + } - public static void Save(this IRepository repository, string bucketId, IAggregate aggregate, Guid commitId) - { - repository.Save(bucketId, aggregate, commitId, a => { }); - } - } + public static void Save(this IRepository repository, string bucketId, IAggregate aggregate, Guid commitId) + { + repository.Save(bucketId, aggregate, commitId, DoNotUpdateHeaders); + } + } } \ No newline at end of file diff --git a/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs b/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs index 38247b5..a3ae8d0 100644 --- a/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs +++ b/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs @@ -1,8 +1,5 @@ namespace NEventStore.Domain.Persistence { - using System; - using System.Collections.Generic; - public static class SagaRepositoryExtensions { public static TSaga GetById(this ISagaRepository sagaRepository, Guid sagaId) From beb0f49cf4a537bcd1bff24c0a8eb22d1b7f584a Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 17 Jan 2025 16:04:43 +0100 Subject: [PATCH 3/7] Refs: #17 Add async methods for repositories and improve error handling This commit introduces asynchronous methods `SaveAsync` and `GetByIdAsync` in the `IRepository` and `ISagaRepository` interfaces, along with their implementations in the respective repository classes. Nullable reference types are now used for better null handling. Test classes have been updated to validate the new asynchronous operations. The `SagaFactory` class has been modified to throw an exception if a constructor is not found, enhancing error handling. Additionally, the `EventStoreRepository` and `SagaEventStoreRepository` classes have been updated to improve performance and responsiveness during I/O operations. These changes enhance the functionality and robustness of the persistence layer, allowing for better scalability and performance in handling aggregates and sagas. --- .../IAggregatePersistenceTests.Async.cs | 433 ++++++++++++++++++ .../Persistence/IAggregatePersistenceTests.cs | 190 ++++---- .../ISagaPersistenceTests.Async.cs | 56 +++ .../Persistence/ISagaPersistenceTests.cs | 91 ++-- .../Persistence/SagaFactory.cs | 34 +- .../EventStore/EventStoreRepository.Async.cs | 84 ++++ .../EventStore/EventStoreRepository.cs | 17 +- .../SagaEventStoreRepository.Async.cs | 65 +++ .../EventStore/SagaEventStoreRepository.cs | 329 +++++++------ .../Persistence/IRepository.cs | 6 +- .../Persistence/ISagaRepository.cs | 6 +- .../Persistence/RepositoryExtensions.cs | 38 +- .../Persistence/SagaRepositoryExtensions.cs | 28 +- 13 files changed, 1031 insertions(+), 346 deletions(-) create mode 100644 src/NEventStore.Domain.Tests/Persistence/IAggregatePersistenceTests.Async.cs create mode 100644 src/NEventStore.Domain.Tests/Persistence/ISagaPersistenceTests.Async.cs create mode 100644 src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.Async.cs create mode 100644 src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.Async.cs 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/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 86a05fd..fa6e379 100644 --- a/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs +++ b/src/NEventStore.Domain/Persistence/EventStore/EventStoreRepository.cs @@ -2,7 +2,7 @@ namespace NEventStore.Domain.Persistence.EventStore { - public class EventStoreRepository : IRepository + public partial class EventStoreRepository : IRepository { private const string AggregateTypeHeader = "AggregateType"; @@ -40,7 +40,7 @@ public TAggregate GetById(string bucketId, Guid id, int version) whe return aggregate; } - public void Save(string bucketId, IAggregate aggregate, Guid commitId, Action> updateHeaders) + public void Save(string bucketId, IAggregate aggregate, Guid commitId, Action>? updateHeaders) { Dictionary headers = PrepareHeaders(aggregate, updateHeaders); while (true) @@ -116,9 +116,8 @@ private TAggregate GetAggregate(ISnapshot? snapshot, IEventStream st private ISnapshot? GetSnapshot(string bucketId, Guid id, int version) { - ISnapshot? snapshot; var snapshotId = bucketId + id; - if (!_snapshots.TryGetValue(snapshotId, out snapshot)) + if (!_snapshots.TryGetValue(snapshotId, out ISnapshot? snapshot)) { _snapshots[snapshotId] = snapshot = _eventStore.Advanced.GetSnapshot(bucketId, id, version); } @@ -144,9 +143,8 @@ private IEventStream OpenStream(string bucketId, Guid id, int version, ISnapshot private IEventStream PrepareStream(string bucketId, IAggregate aggregate, Dictionary headers) { - IEventStream stream; var streamId = bucketId + "+" + aggregate.Id; - if (!_streams.TryGetValue(streamId, out stream)) + if (!_streams.TryGetValue(streamId, out IEventStream stream)) { _streams[streamId] = stream = _eventStore.CreateStream(bucketId, aggregate.Id); } @@ -166,15 +164,12 @@ private IEventStream PrepareStream(string bucketId, IAggregate aggregate, Dictio } private static Dictionary PrepareHeaders( - IAggregate aggregate, Action> updateHeaders) + IAggregate aggregate, Action>? updateHeaders) { var headers = new Dictionary(); headers[AggregateTypeHeader] = aggregate.GetType().FullName; - if (updateHeaders != null) - { - updateHeaders(headers); - } + updateHeaders?.Invoke(headers); return headers; } 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 23967f2..e6b2f45 100644 --- a/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs +++ b/src/NEventStore.Domain/Persistence/EventStore/SagaEventStoreRepository.cs @@ -2,171 +2,166 @@ namespace NEventStore.Domain.Persistence.EventStore { - public 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) - { - 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/IRepository.cs b/src/NEventStore.Domain/Persistence/IRepository.cs index 273d5d7..56be2a1 100644 --- a/src/NEventStore.Domain/Persistence/IRepository.cs +++ b/src/NEventStore.Domain/Persistence/IRepository.cs @@ -4,6 +4,10 @@ public interface IRepository : IDisposable { TAggregate GetById(string bucketId, Guid id, int version) where TAggregate : class, IAggregate; - void Save(string bucketId, IAggregate aggregate, Guid commitId, Action> updateHeaders); + Task GetByIdAsync(string bucketId, Guid id, int version, CancellationToken cancellationToken = default) where TAggregate : class, IAggregate; + + 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 2707a34..a727103 100644 --- a/src/NEventStore.Domain/Persistence/ISagaRepository.cs +++ b/src/NEventStore.Domain/Persistence/ISagaRepository.cs @@ -4,6 +4,10 @@ public interface ISagaRepository { TSaga GetById(string bucketId, string sagaId) where TSaga : class, ISaga; - void Save(string bucketId, ISaga saga, Guid commitId, Action> updateHeaders); + Task GetByIdAsync(string bucketId, string sagaId, CancellationToken cancellationToken = default) where TSaga : class, ISaga; + + 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/RepositoryExtensions.cs b/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs index a7de22e..0c832f8 100644 --- a/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs +++ b/src/NEventStore.Domain/Persistence/RepositoryExtensions.cs @@ -2,8 +2,6 @@ { public static class RepositoryExtensions { - static readonly Action> DoNotUpdateHeaders = _ => { }; - public static TAggregate GetById(this IRepository repository, Guid id) where TAggregate : class, IAggregate { return repository.GetById(Bucket.Default, id, int.MaxValue); @@ -19,19 +17,49 @@ public static TAggregate GetById(this IRepository repository, string return repository.GetById(bucketId, id, int.MaxValue); } - public static void Save(this IRepository repository, IAggregate aggregate, Guid commitId, Action> updateHeaders) + 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, DoNotUpdateHeaders); + repository.Save(aggregate, commitId, null); } public static void Save(this IRepository repository, string bucketId, IAggregate aggregate, Guid commitId) { - repository.Save(bucketId, aggregate, commitId, DoNotUpdateHeaders); + 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 a3ae8d0..7065965 100644 --- a/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs +++ b/src/NEventStore.Domain/Persistence/SagaRepositoryExtensions.cs @@ -2,6 +2,12 @@ namespace NEventStore.Domain.Persistence { 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 { @@ -12,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 From a2d288f10cb7f769d0977d31179d68d230dfe323 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 17 Jan 2025 16:36:44 +0100 Subject: [PATCH 4/7] Fixed AppVeyor build --- appveyor.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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: From 7a7d891208afaecedcb610c0a6ad4a6f031380e6 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 17 Jan 2025 18:52:25 +0100 Subject: [PATCH 5/7] Update NEventStore and dependencies; enhance source linking - Updated subproject commit for NEventStore. - Changed NEventStore dependency version range to stable releases in `NEventStore.Domain.nuspec`. - Added `using System` directive and simplified `handlers` dictionary type in `RegistrationEventRouter.cs`. - Added `PackageReference` for `Microsoft.SourceLink.GitHub` in `NEventStore.Domain.Core.csproj` to improve GitHub integration. --- dependencies/NEventStore | 2 +- src/.nuget/NEventStore.Domain.nuspec | 4 ++-- src/NEventStore.Domain/Core/RegistrationEventRouter.cs | 8 ++++---- src/NEventStore.Domain/NEventStore.Domain.Core.csproj | 7 +++++++ 4 files changed, 14 insertions(+), 7 deletions(-) diff --git a/dependencies/NEventStore b/dependencies/NEventStore index b92f91a..f04b46a 160000 --- a/dependencies/NEventStore +++ b/dependencies/NEventStore @@ -1 +1 @@ -Subproject commit b92f91ac90c743d048b8beef5f41486681b0f191 +Subproject commit f04b46a40a1309754e30f33bb1c1ae7a67b7030e diff --git a/src/.nuget/NEventStore.Domain.nuspec b/src/.nuget/NEventStore.Domain.nuspec index b0cd845..55e0da5 100644 --- a/src/.nuget/NEventStore.Domain.nuspec +++ b/src/.nuget/NEventStore.Domain.nuspec @@ -26,10 +26,10 @@ --> - + - + 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/NEventStore.Domain.Core.csproj b/src/NEventStore.Domain/NEventStore.Domain.Core.csproj index 7b65de2..a07ed47 100644 --- a/src/NEventStore.Domain/NEventStore.Domain.Core.csproj +++ b/src/NEventStore.Domain/NEventStore.Domain.Core.csproj @@ -39,6 +39,13 @@ TRACE;DEBUG + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + From c9127016665f5538b186ae5f7d07502ddebe4640 Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Tue, 21 Jan 2025 10:51:58 +0100 Subject: [PATCH 6/7] Update build process and subproject commit reference Modified `build.ps1` to enable continuous integration builds by adding `/p:ContinuousIntegrationBuild=true` to the `dotnet build` command. Updated the `$nugetversion` variable to pull from `$json.SemVer` for better version management. Also, updated the commit hash in the `NEventStore` subproject to reflect the latest changes. --- build.ps1 | 2 +- dependencies/NEventStore | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/build.ps1 b/build.ps1 index 7b2361a..e6151c4 100644 --- a/build.ps1 +++ b/build.ps1 @@ -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") { diff --git a/dependencies/NEventStore b/dependencies/NEventStore index f04b46a..82c2da7 160000 --- a/dependencies/NEventStore +++ b/dependencies/NEventStore @@ -1 +1 @@ -Subproject commit f04b46a40a1309754e30f33bb1c1ae7a67b7030e +Subproject commit 82c2da7a983c5d2b1e89e9e6ed396febaadfd24b From 6ad289150a060be48d146a942143e6f28ed06dde Mon Sep 17 00:00:00 2001 From: Alessandro Giorgetti Date: Fri, 24 Jan 2025 11:01:29 +0100 Subject: [PATCH 7/7] Update NEventStore subproject to 10.0.0 Updated the NEventStore subproject from commit `82c2da7a983c5d2b1e89e9e6ed396febaadfd24b` to `aaef5b5a8f07d5774a6fd7291cb09c1c676a8f83`. --- dependencies/NEventStore | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dependencies/NEventStore b/dependencies/NEventStore index 82c2da7..aaef5b5 160000 --- a/dependencies/NEventStore +++ b/dependencies/NEventStore @@ -1 +1 @@ -Subproject commit 82c2da7a983c5d2b1e89e9e6ed396febaadfd24b +Subproject commit aaef5b5a8f07d5774a6fd7291cb09c1c676a8f83