[DEVEX-227] Added state cache for ProjectState function

Author: oskardudycz

src/Kurrent.Client/Streams/GettingState/ProjectState.cs

@@ -0,0 +1,81 @@
+using System.Runtime.CompilerServices;
+using EventStore.Client;
+
+namespace Kurrent.Client.Streams.GettingState;
+
+public class ProjectStateOptions<TState> {
+	public Func<ResolvedEvent, string>? GetProjectedId { get; set; }
+	
+	public IStateCache<TState>? StateCache { get; set; }
+}
+
+public static class KurrentClientProjectStateExtensions {
+	public static async IAsyncEnumerable<StateAtPointInTime<TState>> ProjectState<TState>(
+		this IAsyncEnumerable<ResolvedEvent> messages,
+		TState initialState,
+		Func<TState, ResolvedEvent, TState> evolve,
+		ProjectStateOptions<TState>? options,
+		[EnumeratorCancellation] CancellationToken ct
+	) where TState : notnull {
+		if (messages is KurrentClient.ReadStreamResult readStreamResult) {
+			if (await readStreamResult.ReadState.ConfigureAwait(false) == ReadState.StreamNotFound) {
+				yield return new StateAtPointInTime<TState>(initialState);
+
+				yield break;
+			}
+		}
+
+		var getProjectedId = options?.GetProjectedId ?? (resolvedEvent => resolvedEvent.OriginalStreamId);
+		var stateCache     = options?.StateCache ?? new DictionaryStateCache<TState>();
+
+		await foreach (var resolvedEvent in messages.WithCancellation(ct)) {
+			var projectedId = getProjectedId(resolvedEvent);
+			
+			var state = await stateCache.GetValueOrDefaultAsync(projectedId, initialState, ct).ConfigureAwait(false);
+			
+			state = evolve(state, resolvedEvent);
+
+			await stateCache.SetValueAsync(projectedId, state, ct).ConfigureAwait(false);
+
+			yield return new StateAtPointInTime<TState>(
+				state,
+				resolvedEvent.Event.EventNumber,
+				resolvedEvent.Event.Position
+			);
+		}
+	}
+
+	public static IAsyncEnumerable<StateAtPointInTime<TState>> ProjectState<TState>(
+		this IAsyncEnumerable<ResolvedEvent> messages,
+		TState initialState,
+		Func<TState, ResolvedEvent, TState> evolve,
+		CancellationToken ct
+	) where TState : notnull =>
+		messages.ProjectState(initialState, evolve, null, ct);
+}
+
+
+public interface IStateCache<TState> {
+	public ValueTask<TState> GetValueOrDefaultAsync(string key, TState defaultValue, CancellationToken ct = default);
+
+	public ValueTask SetValueAsync(string key, TState state, CancellationToken ct = default);
+}
+
+public class DictionaryStateCache<TState> : IStateCache<TState> {
+	readonly Dictionary<string, TState> _states = new Dictionary<string, TState>();
+
+	public ValueTask<TState> GetValueOrDefaultAsync(string key, TState defaultValue, CancellationToken ct = default) {
+#if NET48
+			var state = _states.TryGetValue(key, out TState? value) ? value : defaultValue;
+#else
+		var state = _states.GetValueOrDefault(key, defaultValue);
+#endif
+		return new ValueTask<TState>(state);
+	}
+
+	public ValueTask SetValueAsync(string key, TState state, CancellationToken ct = default) {
+		_states[key] = state;
+
+		return new ValueTask();
+	}
+}

src/Kurrent.Client/Streams/GettingState/StateBuilder.cs

@@ -172,7 +172,7 @@ public static async Task<StateAtPointInTime<TState>> GetStateAsync<TState>(
 		return await eventStore.ReadStreamAsync(streamName, options, ct)
 			.GetStateAsync(stateBuilder, ct);
 	}
-	
+
 	public static async Task<StateAtPointInTime<TState>> GetStateAsync<TState>(
 		this IAsyncEnumerable<ResolvedEvent> messages,
 		TState initialState,
@@ -197,53 +197,6 @@ CancellationToken ct
 		return new StateAtPointInTime<TState>(state, lastEvent?.Event.EventNumber, lastEvent?.Event.Position);
 	}
 
-	public static async IAsyncEnumerable<StateAtPointInTime<TState>> ProjectState<TState>(
-		this IAsyncEnumerable<ResolvedEvent> messages,
-		TState initialState,
-		Func<TState, ResolvedEvent, TState> evolve,
-		Func<ResolvedEvent, string>? getProjectedId,
-		[EnumeratorCancellation] CancellationToken ct
-	) where TState : notnull {
-		if (messages is KurrentClient.ReadStreamResult readStreamResult) {
-			if (await readStreamResult.ReadState.ConfigureAwait(false) == ReadState.StreamNotFound) {
-				yield return new StateAtPointInTime<TState>(initialState);
-
-				yield break;
-			}
-		}
-
-		var states = new Dictionary<string, TState>();
-
-		getProjectedId ??= resolvedEvent => resolvedEvent.OriginalStreamId;
-
-		await foreach (var resolvedEvent in messages.WithCancellation(ct)) {
-			var projectedId = getProjectedId(resolvedEvent);
-			#if NET48
-			var state = states.TryGetValue(projectedId, out TState? value) ? value : initialState;
-			#else
-			var state = states.GetValueOrDefault(projectedId, initialState);
-			#endif
-			
-			state = evolve(state, resolvedEvent);
-
-			states[projectedId] = state;
-
-			yield return new StateAtPointInTime<TState>(
-				state,
-				resolvedEvent.Event.EventNumber,
-				resolvedEvent.Event.Position
-			);
-		}
-	}
-
-	public static IAsyncEnumerable<StateAtPointInTime<TState>> ProjectState<TState>(
-		this IAsyncEnumerable<ResolvedEvent> messages,
-		TState initialState,
-		Func<TState, ResolvedEvent, TState> evolve,
-		CancellationToken ct
-	) where TState : notnull =>
-		messages.ProjectState(initialState, evolve, null, ct);
-
 	public static Task<StateAtPointInTime<TState>> GetStateAsync<TState>(
 		this KurrentClient eventStore,
 		string streamName,