[DEVEX-227] Updated ProjectState to keep dictionary of states instead of the single state

That gives possibility to cache multiple single stream projection states instead of rolling always a single one

Author: oskardudycz

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

@@ -150,6 +150,29 @@ public class GetStreamStateOptions<TState> : ReadStreamOptions where TState : no
 }
 
 public static class KurrentClientGettingStateClientExtensions {
+	public static async Task<StateAtPointInTime<TState>> GetStateAsync<TState>(
+		this KurrentClient eventStore,
+		string streamName,
+		IStateBuilder<TState> stateBuilder,
+		GetStreamStateOptions<TState>? options,
+		CancellationToken ct = default
+	) where TState : notnull {
+		StateAtPointInTime<TState>? stateAtPointInTime = null;
+
+		options ??= new GetStreamStateOptions<TState>();
+
+		if (options.GetSnapshot != null)
+			stateAtPointInTime = await options.GetSnapshot(
+				GetSnapshotOptions.ForStream(streamName),
+				ct
+			);
+
+		options.StreamPosition = stateAtPointInTime?.LastStreamPosition ?? StreamPosition.Start;
+
+		return await eventStore.ReadStreamAsync(streamName, options, ct)
+			.GetStateAsync(stateBuilder, ct);
+	}
+	
 	public static async Task<StateAtPointInTime<TState>> GetStateAsync<TState>(
 		this IAsyncEnumerable<ResolvedEvent> messages,
 		TState initialState,
@@ -178,21 +201,33 @@ public static async IAsyncEnumerable<StateAtPointInTime<TState>> ProjectState<TS
 		this IAsyncEnumerable<ResolvedEvent> messages,
 		TState initialState,
 		Func<TState, ResolvedEvent, TState> evolve,
+		Func<ResolvedEvent, string>? getProjectedId,
 		[EnumeratorCancellation] CancellationToken ct
 	) where TState : notnull {
-		var state = initialState;
-
 		if (messages is KurrentClient.ReadStreamResult readStreamResult) {
 			if (await readStreamResult.ReadState.ConfigureAwait(false) == ReadState.StreamNotFound) {
-				yield return new StateAtPointInTime<TState>(state);
+				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,
@@ -201,29 +236,13 @@ [EnumeratorCancellation] CancellationToken ct
 		}
 	}
 
-	public static async Task<StateAtPointInTime<TState>> GetStateAsync<TState>(
-		this KurrentClient eventStore,
-		string streamName,
-		IStateBuilder<TState> stateBuilder,
-		GetStreamStateOptions<TState>? options,
-		CancellationToken ct = default
-	) where TState : notnull {
-		StateAtPointInTime<TState>? stateAtPointInTime = null;
-
-		options ??= new GetStreamStateOptions<TState> ();
-		
-		if (options.GetSnapshot != null) {
-			stateAtPointInTime = await options.GetSnapshot(
-				GetSnapshotOptions.ForStream(streamName),
-				ct
-			);
-		}
-
-		options.StreamPosition = stateAtPointInTime?.LastStreamPosition ?? StreamPosition.Start;
-
-		return await eventStore.ReadStreamAsync(streamName, options, ct)
-			.GetStateAsync(stateBuilder, ct);
-	}
+	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,