Bring back ProjectionFactory

Also use IServiceScopeFactory instead of IServiceProvider in
ProjectionProcessor to make it more clear that we are only creating scopes

Author: Andreas Gehrke

src/Atc.Cosmos.EventStore.Cqrs/DependencyInjection/EventStoreOptionsBuilderExtensions.cs

@@ -28,6 +28,7 @@ public static EventStoreOptionsBuilder UseCQRS(
         builder.Services.AddSingleton<IProjectionOptionsFactory, ProjectionOptionsFactory>();
 
         builder.Services.AddSingleton(typeof(ProjectionMetadata<>), typeof(ProjectionMetadata<>));
+        builder.Services.AddTransient<IProjectionFactory, DefaultProjectionFactory>();
 
         builder.Services.TryAddSingleton<IProjectionTelemetry, ProjectionTelemetry>();
 

src/Atc.Cosmos.EventStore.Cqrs/ProjectionFactory.cs

@@ -0,0 +1,40 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Atc.Cosmos.EventStore.Cqrs;
+
+/// <summary>
+/// Responsible for creating <see cref="IProjection"/> instances.
+/// </summary>
+public interface IProjectionFactory
+{
+    /// <summary>
+    /// Creates a projection of type <typeparamref name="TProjection"/> for the event stream
+    /// identified by <paramref name="streamId"/>.
+    /// </summary>
+    /// <param name="streamId">ID of the stream being projected.</param>
+    /// <param name="cancellationToken">Cancellation.</param>
+    /// <typeparam name="TProjection">Type of projection to create.</typeparam>
+    /// <returns>The created projection.</returns>
+    public Task<IProjection> CreateAsync<TProjection>(EventStreamId streamId, CancellationToken cancellationToken)
+        where TProjection : IProjection;
+}
+
+/// <summary>
+/// The default projection factory which just creates projections by
+/// getting them from the DI-container.
+/// </summary>
+internal sealed class DefaultProjectionFactory : IProjectionFactory
+{
+    private readonly IServiceProvider serviceProvider;
+
+    public DefaultProjectionFactory(IServiceProvider serviceProvider)
+    {
+        this.serviceProvider = serviceProvider;
+    }
+
+    public Task<IProjection> CreateAsync<TProjection>(EventStreamId streamId, CancellationToken cancellationToken)
+        where TProjection : IProjection
+    {
+        return Task.FromResult<IProjection>(serviceProvider.GetRequiredService<TProjection>());
+    }
+}

src/Atc.Cosmos.EventStore.Cqrs/Projections/ProjectionProcessor.cs

@@ -10,18 +10,18 @@ internal class ProjectionProcessor<TProjection> : IProjectionProcessor<TProjecti
     private readonly IReadOnlyCollection<ProjectionFilter> filters;
     private readonly IProjectionTelemetry telemetry;
     private readonly ProjectionMetadata<TProjection> projectionMetadata;
-    private readonly IServiceProvider serviceProvider;
+    private readonly IServiceScopeFactory serviceScopeFactory;
     private readonly string projectionName;
 
     public ProjectionProcessor(
         IProjectionOptionsFactory optionsFactory,
         IProjectionTelemetry telemetry,
         ProjectionMetadata<TProjection> projectionMetadata,
-        IServiceProvider serviceProvider)
+        IServiceScopeFactory serviceScopeFactory)
     {
         this.telemetry = telemetry;
         this.projectionMetadata = projectionMetadata;
-        this.serviceProvider = serviceProvider;
+        this.serviceScopeFactory = serviceScopeFactory;
         filters = optionsFactory
             .GetOptions<TProjection>()
             .Filters;
@@ -49,10 +49,6 @@ public async Task<ProjectionAction> ProcessBatchAsync(
 
         foreach (var events in groupedEvents)
         {
-            await using var scope = serviceProvider.CreateAsyncScope();
-
-            var projection = scope.ServiceProvider.GetRequiredService<TProjection>();
-
             using var operation = batchTelemetry.StartProjection(events.Key);
 
             if (!projectionMetadata.CanConsumeOneOrMoreEvents(events))
@@ -62,11 +58,16 @@ public async Task<ProjectionAction> ProcessBatchAsync(
                 continue;
             }
 
+            var eventStreamId = EventStreamId.FromStreamId(events.Key);
+            await using var scope = serviceScopeFactory.CreateAsyncScope();
+            var projectionFactory = scope.ServiceProvider.GetRequiredService<IProjectionFactory>();
+            var projection = await projectionFactory.CreateAsync<TProjection>(eventStreamId, cancellationToken);
+
             try
             {
                 await projection
                     .InitializeAsync(
-                        events.Key,
+                        eventStreamId,
                         cancellationToken)
                     .ConfigureAwait(false);