Projections#1
Conversation
There was a problem hiding this comment.
Pull Request Overview
This PR implements a comprehensive projections feature for the Strata Framework, adding asynchronous event processing capabilities to event-sourced grains. The implementation provides two projection approaches: Host Managed Projections (using Orleans OneWay calls) and Projection Managed Grains (using Orleans streams).
- Core projection infrastructure with type-safe interfaces and configuration options
- Host Managed Projections via ProjectionGrain with internal queuing and worker threads
- Projection Managed Grains via EventRecipientGrain with stream-based event processing
Reviewed Changes
Copilot reviewed 21 out of 21 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| src/Strata/Projections/IProjection.cs | Core generic projection interface defining Handle method |
| src/Strata/Projections/IProjectionGrain.cs | Orleans grain interface for projection processing with OneWay calls |
| src/Strata/Projections/ProjectionGrain.cs | Main projection grain implementation with internal queue and worker threads |
| src/Strata/Projections/ProjectionOptions.cs | Configuration options with validation attributes |
| src/Strata/Projections/ProjectionRegistry.cs | Registry for managing projection type registrations and mappings |
| src/Strata/Projections/ProjectionStateManager.cs | State management for stateful projections with serialization support |
| src/Strata/Projections/StreamEventProcessor.cs | Stream event processing with reflection-based method routing |
| src/Strata/Projections/EventRecipientGrain.cs | Base class for stream-based projection grains |
| src/Strata/Projections/GrainExtensions.cs | Extension methods for registering projections with EventSourcedGrain |
| src/Strata/Projections/ServiceCollectionExtensions.cs | Dependency injection configuration extensions |
| src/Strata/EventSourcedGrain.cs | Integration of projection processing into event raising |
| tests and examples | Comprehensive test coverage and usage examples |
Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
| /// <returns>A collection of projection IDs.</returns> | ||
| public IEnumerable<string> GetAllProjectionIds() | ||
| { | ||
| return _stateCache.Keys.ToList(); |
There was a problem hiding this comment.
The ToList() call creates an unnecessary copy of the keys collection. Consider returning IEnumerable directly or use _stateCache.Keys.ToArray() if materialization is needed for thread safety.
| return _stateCache.Keys.ToList(); | |
| return _stateCache.Keys; |
| public ProjectionStateManager(ILogger<ProjectionStateManager> logger) | ||
| { | ||
| _logger = logger ?? throw new ArgumentNullException(nameof(logger)); | ||
| _stateCache = new Dictionary<string, object>(); | ||
| _stateVersions = new Dictionary<string, int>(); | ||
| _jsonOptions = new JsonSerializerOptions | ||
| { | ||
| PropertyNamingPolicy = JsonNamingPolicy.CamelCase, | ||
| WriteIndented = false | ||
| }; |
There was a problem hiding this comment.
[nitpick] Consider making JsonSerializerOptions configurable through dependency injection or options pattern rather than hardcoding the configuration. This would allow different serialization behaviors for different scenarios.
| public ProjectionStateManager(ILogger<ProjectionStateManager> logger) | |
| { | |
| _logger = logger ?? throw new ArgumentNullException(nameof(logger)); | |
| _stateCache = new Dictionary<string, object>(); | |
| _stateVersions = new Dictionary<string, int>(); | |
| _jsonOptions = new JsonSerializerOptions | |
| { | |
| PropertyNamingPolicy = JsonNamingPolicy.CamelCase, | |
| WriteIndented = false | |
| }; | |
| public ProjectionStateManager( | |
| ILogger<ProjectionStateManager> logger, | |
| JsonSerializerOptions jsonOptions) | |
| { | |
| _logger = logger ?? throw new ArgumentNullException(nameof(logger)); | |
| _stateCache = new Dictionary<string, object>(); | |
| _stateVersions = new Dictionary<string, int>(); | |
| _jsonOptions = jsonOptions ?? throw new ArgumentNullException(nameof(jsonOptions)); |
| var projectionGrain = grain.GrainFactory.GetGrain<IProjectionGrain>( | ||
| $"{grain.GetGrainId()}_{projectionType.Name}"); |
There was a problem hiding this comment.
[nitpick] The grain ID construction using string concatenation creates a magic string format. Consider extracting this to a constant or helper method to ensure consistency across the codebase.
| // Process projections asynchronously (fire-and-forget) | ||
| _ = Task.Run(async () => await ProcessProjectionsAsync(@event)); |
There was a problem hiding this comment.
Using Task.Run for CPU-bound work is appropriate, but consider using ConfigureAwait(false) in the async lambda to avoid capturing synchronization context unnecessarily.
|
|
||
| try | ||
| { | ||
| var instance = Activator.CreateInstance(projectionTypeInfo); |
There was a problem hiding this comment.
[nitpick] Using Activator.CreateInstance assumes parameterless constructors. Consider supporting dependency injection by integrating with the service provider to create projection instances with their dependencies.
Adds Journaling Grain Base using new Durable Objects
…lementation contract
Adds Projections
Full documentation in projections.md
Task list is in projections.tasks.md