refactor: Make processor generic (#22)

Author: NilsLattek

README.md

@@ -1,3 +1,11 @@
 # underground
 
-work in progress...
+work in progress...
+
+## Example
+
+Run example:
+
+```bash
+dotnet run --project example/ConsoleApp/
+```

example/ConsoleApp/AppDbContext.cs

@@ -4,9 +4,10 @@
 
 namespace ConsoleApp;
 
-sealed class AppDbContext(DbContextOptions<AppDbContext> options) : DbContext(options), IOutboxDbContext
+sealed class AppDbContext(DbContextOptions<AppDbContext> options) : DbContext(options), IOutboxDbContext, IInboxDbContext
 {
     public DbSet<OutboxMessage> OutboxMessages { get; set; }
+    public DbSet<InboxMessage> InboxMessages { get; set; }
 
     // example on how to apply custom configuration to the outbox model
     protected override void OnModelCreating(ModelBuilder modelBuilder)

example/ConsoleApp/ExampleMessageHandler.cs

@@ -7,7 +7,7 @@ public class ExampleMessageHandler : IOutboxMessageHandler<ExampleMessage>
 {
     public Task HandleAsync(ExampleMessage message, CancellationToken cancellationToken)
     {
-        Console.WriteLine("received: " + message.Id);
+        Console.WriteLine("received outbox: " + message.Id);
         return Task.CompletedTask;
     }
 }

example/ConsoleApp/InboxMessageHandler.cs

@@ -0,0 +1,13 @@
+
+using Underground.Outbox;
+
+namespace ConsoleApp;
+
+public class InboxMessageHandler : IInboxMessageHandler<ExampleMessage>
+{
+    public Task HandleAsync(ExampleMessage message, CancellationToken cancellationToken)
+    {
+        Console.WriteLine("received inbox: " + message.Id);
+        return Task.CompletedTask;
+    }
+}

example/ConsoleApp/Program.cs

@@ -26,10 +26,15 @@
 {
     cfg.AddHandler<ExampleMessageHandler>();
 });
+builder.Services.AddInboxServices<AppDbContext>(cfg =>
+{
+    cfg.AddHandler<InboxMessageHandler>();
+});
 
 IHost host = builder.Build();
 
 var outbox = host.Services.GetRequiredService<IOutbox>();
+var inbox = host.Services.GetRequiredService<IInbox>();
 var dbContext = host.Services.GetRequiredService<AppDbContext>();
 await dbContext.Database.EnsureCreatedAsync();
 
@@ -40,6 +45,9 @@
         var partition = (i % 3).ToString();
         var message = new OutboxMessage(Guid.NewGuid(), DateTime.UtcNow, new ExampleMessage($"partition {partition}: {i}"), partition);
         await outbox.AddMessageAsync(dbContext, message, CancellationToken.None);
+
+        var inboxMessage = new InboxMessage(Guid.NewGuid(), DateTime.UtcNow, new ExampleMessage($"inbox message: {i}"));
+        await inbox.AddMessageAsync(dbContext, inboxMessage, CancellationToken.None);
     }
 
     await transaction.CommitAsync();

src/Underground.Outbox/Configuration/ConfigureOutboxServices.cs

@@ -23,34 +23,60 @@ Action<OutboxServiceConfiguration> configuration
         var serviceConfig = new OutboxServiceConfiguration();
         configuration.Invoke(serviceConfig);
 
-        // register all assigned handlers
-        services.TryAddEnumerable(serviceConfig.HandlersWithLifetime);
-
         services.AddScoped<IOutboxDbContext>(sp => sp.GetRequiredService<TContext>());
         services.AddScoped<AddMessageToOutbox>();
         services.AddScoped<IOutbox, Outbox>();
-        services.AddScoped<IMessageDispatcher, DirectInvocationDispatcher>();
-        services.AddScoped<IMessageExceptionHandler, DiscardMessageOnExceptionHandler>();
-        services.AddScoped<ProcessExceptionFromHandler>();
+        services.AddScoped<IMessageDispatcher<OutboxMessage>, OutboxDispatcher>();
+
+        AddGenericServices<OutboxMessage, IOutboxDbContext>(services, serviceConfig);
+
+        return services;
+    }
+
+    public static IServiceCollection AddInboxServices<TContext>(
+        this IServiceCollection services,
+        Action<InboxServiceConfiguration> configuration
+    ) where TContext : DbContext, IInboxDbContext
+    {
+        var serviceConfig = new InboxServiceConfiguration();
+        configuration.Invoke(serviceConfig);
+
+        services.AddScoped<IInboxDbContext>(sp => sp.GetRequiredService<TContext>());
+        services.AddScoped<AddMessageToInbox>();
+        services.AddScoped<IInbox, Inbox>();
+        services.AddScoped<IMessageDispatcher<InboxMessage>, InboxDispatcher>();
+
+        AddGenericServices<InboxMessage, IInboxDbContext>(services, serviceConfig);
+
+        return services;
+    }
+
+    private static void AddGenericServices<TEntity, TContext>(this IServiceCollection services, ServiceConfiguration serviceConfig)
+    where TEntity : class, IMessage
+    where TContext : IDbContext
+    {
+        // register all assigned handlers
+        services.TryAddEnumerable(serviceConfig.HandlersWithLifetime);
+
+        services.AddScoped<IMessageExceptionHandler<TEntity>, DiscardMessageOnExceptionHandler<TEntity>>();
+        services.AddScoped<ProcessExceptionFromHandler<TEntity>>();
         services.AddSingleton(
-            provider => new OutboxProcessor(
+            provider => new Processor<TEntity>(
                 serviceConfig,
                 provider.GetRequiredService<IServiceScopeFactory>(),
-                provider.GetRequiredService<IMessageDispatcher>(),
-                provider.GetRequiredService<ILogger<OutboxProcessor>>()
+                provider.GetRequiredService<IMessageDispatcher<TEntity>>(),
+                provider.GetRequiredService<ILogger<Processor<TEntity>>>()
             )
         );
-        services.AddHostedService<OutboxBackgroundService>();
+        services.AddHostedService<BackgroundService<TEntity>>();
 
         var serviceProvider = services.BuildServiceProvider();
-        var dbContext = serviceProvider.GetRequiredService<IOutboxDbContext>();
+        var dbContext = serviceProvider.GetRequiredService<TContext>();
         var connectionString = dbContext.Database.GetConnectionString();
         if (string.IsNullOrEmpty(connectionString))
         {
             throw new ArgumentException("Database connection string is not set. Please ensure the DbContext is properly configured.");
         }
         services.AddSingleton<IDistributedLockProvider>(_ => new PostgresDistributedSynchronizationProvider(connectionString));
-
-        return services;
     }
 }

src/Underground.Outbox/Configuration/InboxServiceConfiguration.cs

@@ -0,0 +1,19 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Underground.Outbox.Configuration;
+
+public class InboxServiceConfiguration : ServiceConfiguration
+{
+    public override ServiceConfiguration AddHandler(HandlerType messageHandlerType, ServiceLifetime serviceLifetime = ServiceLifetime.Transient)
+    {
+        var interfaceType = messageHandlerType.GetInterface("Underground.Outbox.IInboxMessageHandler`1");
+
+        if (interfaceType?.IsGenericType == true)
+        {
+            Console.WriteLine($"Added handler for {interfaceType} with {messageHandlerType} ");
+            HandlersWithLifetime.Add(new ServiceDescriptor(interfaceType, messageHandlerType, serviceLifetime));
+        }
+
+        return this;
+    }
+}

src/Underground.Outbox/Configuration/OutboxServiceConfiguration.cs

@@ -2,29 +2,9 @@
 
 namespace Underground.Outbox.Configuration;
 
-public class OutboxServiceConfiguration
+public class OutboxServiceConfiguration : ServiceConfiguration
 {
-    /// <summary>
-    /// Number of messages to process in a single batch.
-    /// The whole batch is processed within a single transaction. If you want to have a transaction per message, set this to 1.
-    /// </summary>
-    public int BatchSize { get; set; } = 5;
-
-    public int ParallelProcessingOfPartitions { get; set; } = 4;
-
-    internal List<ServiceDescriptor> HandlersWithLifetime = [];
-
-    public OutboxServiceConfiguration AddHandler<TMessageHandlerType>()
-    {
-        return AddHandler(typeof(TMessageHandlerType));
-    }
-
-    public OutboxServiceConfiguration AddHandler<TMessageHandlerType>(ServiceLifetime serviceLifetime)
-    {
-        return AddHandler(typeof(TMessageHandlerType), serviceLifetime);
-    }
-
-    public OutboxServiceConfiguration AddHandler(HandlerType messageHandlerType, ServiceLifetime serviceLifetime = ServiceLifetime.Transient)
+    public override ServiceConfiguration AddHandler(HandlerType messageHandlerType, ServiceLifetime serviceLifetime = ServiceLifetime.Transient)
     {
         var interfaceType = messageHandlerType.GetInterface("Underground.Outbox.IOutboxMessageHandler`1");
 

src/Underground.Outbox/Configuration/ServiceConfiguration.cs

@@ -0,0 +1,31 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Underground.Outbox.Configuration;
+
+public abstract class ServiceConfiguration
+{
+    /// <summary>
+    /// Number of messages to process in a single batch.
+    /// The whole batch is processed within a single transaction. If you want to have a transaction per message, set this to 1.
+    /// </summary>
+    public int BatchSize { get; set; } = 5;
+
+    public int ParallelProcessingOfPartitions { get; set; } = 4;
+
+    internal List<ServiceDescriptor> HandlersWithLifetime = [];
+
+    // TODO: add constraint on the generic type? like IInboxMessageHandler?
+    public ServiceConfiguration AddHandler<TMessageHandlerType>()
+    {
+        return AddHandler(typeof(TMessageHandlerType));
+    }
+
+    public ServiceConfiguration AddHandler<TMessageHandlerType>(ServiceLifetime serviceLifetime)
+    {
+        return AddHandler(typeof(TMessageHandlerType), serviceLifetime);
+    }
+
+#pragma warning disable CA1716 // Identifiers should not match keywords
+    public abstract ServiceConfiguration AddHandler(HandlerType messageHandlerType, ServiceLifetime serviceLifetime = ServiceLifetime.Transient);
+#pragma warning restore CA1716 // Identifiers should not match keywords
+}

src/Underground.Outbox/Data/IDbContext.cs

@@ -0,0 +1,24 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.ChangeTracking;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+
+namespace Underground.Outbox.Data;
+
+public interface IDbContext : IAsyncDisposable
+{
+    public DatabaseFacade Database { get; }
+    public ChangeTracker ChangeTracker { get; }
+#pragma warning disable CA1716 // Identifiers should not match keywords
+    public DbSet<TEntity> Set<TEntity>() where TEntity : class;
+#pragma warning restore CA1716 // Identifiers should not match keywords
+
+    /// <summary>
+    /// Asynchronously saves all changes made in this context to the database.
+    /// </summary>
+    /// <param name="cancellationToken">A <see cref="CancellationToken" /> to observe while waiting for the task to complete.</param>
+    /// <returns>
+    /// A task that represents the asynchronous save operation. The task result contains the
+    /// number of state entries written to the database.
+    /// </returns>
+    Task<int> SaveChangesAsync(CancellationToken cancellationToken = default);
+}