[DEVEX-227] Added Polly to support retries of business logic

Author: oskardudycz

src/Kurrent.Client/Kurrent.Client.csproj

@@ -12,7 +12,8 @@
 		<PackageReference Include="Microsoft.Extensions.Logging" Version="9.0.0"/>
 		<PackageReference Include="System.Diagnostics.DiagnosticSource" Version="9.0.0"/>
 		<PackageReference Include="System.Linq.Async" Version="6.0.1"/>
-	  <PackageReference Include="OpenTelemetry.Api" Version="1.10.0" />
+	  <PackageReference Include="OpenTelemetry.Api" Version="1.11.2" />
+		<PackageReference Include="Polly" Version="8.5.2" />
 
 		<PackageReference Include="Grpc.Tools" Version="2.68.1">
 			<PrivateAssets>all</PrivateAssets>

src/Kurrent.Client/Streams/DecisionMaking/Aggregate.cs

@@ -7,7 +7,7 @@ public interface IAggregate<in TEvent> : IState<TEvent> {
 	Message[] DequeueUncommittedMessages();
 }
 
-public interface IAggregate : IAggregate<object>;
+public interface IAggregate : IAggregate<object>, IState;
 
 public class Aggregate : Aggregate<object>, IAggregate;
 
@@ -24,9 +24,9 @@ Message[] IAggregate<TEvent>.DequeueUncommittedMessages() {
 		return dequeuedEvents;
 	}
 
-	protected void Enqueue(TEvent message) {
-		Apply(message);
-		_uncommittedEvents.Enqueue(Message.From(message));
+	protected void Enqueue(TEvent @event) {
+		Apply(@event);
+		_uncommittedEvents.Enqueue(Message.From(@event));
 	}
 
 	protected void Enqueue(Message message) {

src/Kurrent.Client/Streams/DecisionMaking/StateStore.cs

@@ -55,7 +55,7 @@ public static Task<IWriteResult> AddAsync<TState>(
 			null,
 			ct
 		);
-	
+
 	public static Task<IWriteResult> AddAsync<TState>(
 		this IStateStore<TState> stateStore,
 		string streamName,
@@ -68,13 +68,14 @@ public static Task<IWriteResult> AddAsync<TState>(
 			new AppendToStreamOptions { ExpectedStreamState = StreamState.NoStream },
 			ct
 		);
-	
+
 	public static Task<IWriteResult> AddAsync<TState, TEvent>(
 		this IStateStore<TState> stateStore,
 		string streamName,
 		IEnumerable<TEvent> events,
 		CancellationToken ct = default
-	) where TState : notnull where TEvent : notnull =>
+	) where TState : notnull
+	  where TEvent : notnull =>
 		stateStore.AddAsync(
 			streamName,
 			events.Select(e => Message.From(e)),
@@ -87,7 +88,8 @@ public static Task<IWriteResult> UpdateAsync<TState, TEvent>(
 		string streamName,
 		IEnumerable<TEvent> events,
 		CancellationToken ct = default
-	) where TState : notnull where TEvent : notnull =>
+	) where TState : notnull
+	  where TEvent : notnull =>
 		stateStore.UpdateAsync(
 			streamName,
 			events.Select(e => Message.From(e)),
@@ -114,7 +116,8 @@ public static Task<IWriteResult> UpdateAsync<TState, TEvent>(
 		IEnumerable<TEvent> events,
 		StreamRevision expectedStreamRevision,
 		CancellationToken ct = default
-	) where TState : notnull where TEvent : notnull =>
+	) where TState : notnull
+	  where TEvent : notnull =>
 		stateStore.UpdateAsync(
 			streamName,
 			events.Select(e => Message.From(e)),
@@ -154,7 +157,21 @@ public static Task<IWriteResult> Handle<TState, TEvent>(
 		string streamName,
 		Func<TState, TEvent[]> handle,
 		CancellationToken ct = default
-	) where TState : notnull where TEvent : notnull =>
+	) where TState : notnull
+	  where TEvent : notnull =>
+		stateStore.Handle(
+			streamName,
+			(state, _) => new ValueTask<Message[]>(handle(state).Select(m => Message.From(m)).ToArray()),
+			null,
+			ct
+		);
+
+	public static Task<IWriteResult> Handle<TState>(
+		this IStateStore<TState> stateStore,
+		string streamName,
+		Func<TState, object[]> handle,
+		CancellationToken ct = default
+	) where TState : notnull =>
 		stateStore.Handle(
 			streamName,
 			(state, _) => new ValueTask<Message[]>(handle(state).Select(m => Message.From(m)).ToArray()),
@@ -168,7 +185,22 @@ public static Task<IWriteResult> Handle<TState, TEvent>(
 		Func<TState, TEvent[]> handle,
 		DecideOptions<TState>? decideOptions,
 		CancellationToken ct = default
-	) where TState : notnull where TEvent : notnull =>
+	) where TState : notnull
+	  where TEvent : notnull =>
+		stateStore.Handle(
+			streamName,
+			(state, _) => new ValueTask<Message[]>(handle(state).Select(m => Message.From(m)).ToArray()),
+			decideOptions,
+			ct
+		);
+
+	public static Task<IWriteResult> Handle<TState>(
+		this IStateStore<TState> stateStore,
+		string streamName,
+		Func<TState, object[]> handle,
+		DecideOptions<TState>? decideOptions,
+		CancellationToken ct = default
+	) where TState : notnull =>
 		stateStore.Handle(
 			streamName,
 			(state, _) => new ValueTask<Message[]>(handle(state).Select(m => Message.From(m)).ToArray()),
@@ -179,7 +211,7 @@ public static Task<IWriteResult> Handle<TState, TEvent>(
 
 public class StateStore<TState>(KurrentClient client, StateStoreOptions<TState> options)
 	: IStateStore<TState>
-	where TState : notnull{
+	where TState : notnull {
 	public virtual Task<StateAtPointInTime<TState>> Get(string streamName, CancellationToken ct = default) =>
 		client.GetStateAsync(streamName, options.StateBuilder, options.GetStreamStateOptions, ct);
 

test/Kurrent.Client.Tests.Common/Kurrent.Client.Tests.Common.csproj

@@ -10,7 +10,7 @@
 	<ItemGroup>
 		<PackageReference Include="Ductus.FluentDocker" Version="2.10.59"/>
 		<PackageReference Include="Humanizer.Core" Version="2.14.1"/>
-		<PackageReference Include="Polly" Version="8.5.0"/>
+		<PackageReference Include="Polly" Version="8.5.2"/>
 		<PackageReference Include="Polly.Contrib.WaitAndRetry" Version="1.1.1"/>
 		<PackageReference Include="Serilog" Version="4.2.0"/>
 		<PackageReference Include="Serilog.AspNetCore" Version="9.0.0"/>